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

        // Input file tweets.csv
        File tweets = new File(args[0]);

        //Create queues
        BlockingQueue<Record> queue1 = new LinkedBlockingQueue<Record>();
        BlockingQueue<Record> queue2 = new LinkedBlockingQueue<Record>();

        // Build easy batch engines
        Engine engine1 = buildBatchEngine(queue1);
        Engine engine2 = buildBatchEngine(queue2);

        //create a 2 threads pool to call engines in parallel
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

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

        //create a record dispatcher to dispatch records to previously created queues
        RecordDispatcher recordDispatcher = new RoundRobinRecordDispatcher(Arrays.asList(queue1, queue2));

        //read data source and dispatch records to queues in round-robin fashion
        FlatFileRecordReader flatFileRecordReader = new FlatFileRecordReader(tweets);
        flatFileRecordReader.open();
        while (flatFileRecordReader.hasNextRecord()) {
            Record record = flatFileRecordReader.readNextRecord();
            recordDispatcher.dispatchRecord(record);
        }
        flatFileRecordReader.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();

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

        //shutdown executor service
        executorService.shutdown();

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

        // Input file tweets.csv
        File tweets = new File(args[0]);

        // Create queues
        BlockingQueue<Record> queue1 = new LinkedBlockingQueue<Record>();
        BlockingQueue<Record> queue2 = new LinkedBlockingQueue<Record>();

        // Create a round robin record dispatcher that will be used to distribute records to worker engines
        RoundRobinRecordDispatcher roundRobinRecordDispatcher = new RoundRobinRecordDispatcher(Arrays.asList(queue1, queue2));

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

        // Build worker engines
        Engine workerEngine1 = buildWorkerEngine(queue1);
        Engine workerEngine2 = buildWorkerEngine(queue2);

        // Create a thread pool to call master and worker engines in parallel
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

        // Submit workers to executor service
        executorService.submit(masterEngine);
        executorService.submit(workerEngine1);
        executorService.submit(workerEngine2);

        // Shutdown executor service
        executorService.shutdown();

    }