#vulnerable code 
                                public static void main(String[] args) throws Exception {

        //Create queues
        BlockingQueue<Record> appleQueue = new LinkedBlockingQueue<Record>();
        BlockingQueue<Record> orangeQueue = new LinkedBlockingQueue<Record>();
        BlockingQueue<Record> defaultQueue = new LinkedBlockingQueue<Record>();

        // Build easy batch engines
        Engine engine1 = buildBatchEngine(appleQueue);
        Engine engine2 = buildBatchEngine(orangeQueue);
        Engine engine3 = buildBatchEngine(defaultQueue);

        //create a 3 threads pool to call Easy Batch engines in parallel
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        //submit workers to executor service
        Future<Report> reportFuture1 = executorService.submit(engine1);
        Future<Report> reportFuture2 = executorService.submit(engine2);
        Future<Report> reportFuture3 = executorService.submit(engine3);

        //create a content based record dispatcher to dispatch records to previously created queues
        RecordDispatcher recordDispatcher = new ContentBasedRecordDispatcherBuilder()
                .when(new AppleRecordPredicate()).dispatchTo(appleQueue)
                .when(new OrangeRecordPredicate()).dispatchTo(orangeQueue)
                .otherwise(defaultQueue)
                .build();

        //read data source and dispatch record to queues based on their content
        StringRecordReader stringRecordReader = new StringRecordReader("1,apple\n2,orange\n3,banana\n4,apple\n5,pear");
        stringRecordReader.open();
        while (stringRecordReader.hasNextRecord()) {
            Record record = stringRecordReader.readNextRecord();
            recordDispatcher.dispatchRecord(record);
        }
        stringRecordReader.close();

        //send poison records when all input data has been dispatched to workers
        recordDispatcher.dispatchRecord(new PoisonRecord());

        //wait for easy batch instances termination and get partial reports
        Report report1 = reportFuture1.get();
        Report report2 = reportFuture2.get();
        Report report3 = reportFuture3.get();

        //merge partial reports into a global one
        ReportMerger reportMerger = new DefaultReportMerger();
        Report finalReport = reportMerger.mergerReports(report1, report2, report3);
        System.out.println(finalReport);

        //shutdown executor service
        executorService.shutdown();

    } 
                            #location 35 
                            #vulnerability type RESOURCE_LEAK
#fixed code
                                public static void main(String[] args) throws Exception {

        String fruits = "1,apple\n2,orange\n3,banana\n4,apple\n5,pear";

        // Create queues
        BlockingQueue<Record> appleQueue = new LinkedBlockingQueue<Record>();
        BlockingQueue<Record> orangeQueue = new LinkedBlockingQueue<Record>();
        BlockingQueue<Record> defaultQueue = new LinkedBlockingQueue<Record>();

        // Create a content based record dispatcher to dispatch records to according queues based on their content
        ContentBasedRecordDispatcher recordDispatcher = new ContentBasedRecordDispatcherBuilder()
                .when(new AppleRecordPredicate()).dispatchTo(appleQueue)
                .when(new OrangeRecordPredicate()).dispatchTo(orangeQueue)
                .otherwise(defaultQueue)
                .build();

        // Build a master engine that will read records from the data source and dispatch them to worker engines
        Engine masterEngine = aNewEngine()
                .reader(new StringRecordReader(fruits))
                .processor(recordDispatcher)
                .batchProcessEventListener(new PoisonRecordBroadcaster(recordDispatcher))
                .build();

        // Build easy batch engines
        Engine workerEngine1 = buildWorkerEngine(appleQueue);
        Engine workerEngine2 = buildWorkerEngine(orangeQueue);
        Engine workerEngine3 = buildWorkerEngine(defaultQueue);

        // Create a threads pool to call Easy Batch engines in parallel
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

        // Submit master and worker engines to executor service
        executorService.submit(masterEngine);
        executorService.submit(workerEngine1);
        executorService.submit(workerEngine2);
        executorService.submit(workerEngine3);

        // Shutdown executor service
        executorService.shutdown();

    }