#vulnerable code 
                                @Override
    public void start() {
        consumerWorkers = new ArrayList<>(consumerThreads);
        CountDownLatch startupLatch = new CountDownLatch(consumerThreads);
        for (int i = 0; i < consumerThreads; i++) {
            KafkaConsumer<String, Request> consumer = new KafkaConsumer<>(consumerProps);
            RequestConsumerWorker worker = new RequestConsumerWorker(topic, consumer, messageDispatcher, startupLatch);
            consumerExecutor.submit(worker);
        }

        try {
            startupLatch.await(5000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            logger.error("Error while waiting for server consumers to subscribe", e);
        }
    } 
                            #location 8 
                            #vulnerability type RESOURCE_LEAK
#fixed code
                                @Override
    public void start() {
        disruptor.handleEventsWith(eventHandler);
        disruptor.start();

        RingBuffer<ServerEvent> ringBuffer = disruptor.getRingBuffer();
        requestConsumer.startConsumers(ringBuffer);
    }