#vulnerable code 
                             @Override
	public void start() {

		for (int i = 0; i < topicHandlers.size(); i++) {
			ConsumerWorker<String, DefaultMessage> consumer = new ConsumerWorker<>(configs, topicHandlers,processExecutor);
			consumers.add(consumer);
			fetcheExecutor.submit(consumer);
		}
	} 
                            #location 7 
                            #vulnerability type RESOURCE_LEAK
#fixed code
                             @Override
	public void start() {
		createKafkaConsumer();
		//按主题数创建ConsumerWorker线程
		for (int i = 0; i < topicHandlers.size(); i++) {
			ConsumerWorker consumer = new ConsumerWorker();
			consumerWorks.add(consumer);
			fetcheExecutor.submit(consumer);
		}
	}