#fixed code
                                @Test
    public void testSourceSinkStartResumeRollingEverySecond() throws Exception {
        //This is the config that is required to make the VanillaChronicle roll every second

        final String sourceBasePath = getVanillaTestPath("-source");
        final String sinkBasePath = getVanillaTestPath("-sink");
        assertNotNull(sourceBasePath);
        assertNotNull(sinkBasePath);

        final Chronicle source = ChronicleQueueBuilder.source(
                ChronicleQueueBuilder.vanilla(sourceBasePath)
                    .entriesPerCycle(1L << 20)
                    .cycleLength(1000, false)
                    .cycleFormat("yyyyMMddHHmmss")
                    .indexBlockSize(16L << 10)
                    .build())
            .bindAddress("localhost", BASE_PORT + 104)
            .build();

        ExcerptAppender appender = source.createAppender();
        System.out.print("writing 100 items will take take 10 seconds.");
        for (int i = 0; i < 100; i++) {
            appender.startExcerpt();
            int value = 1000000000 + i;
            appender.append(value).append(' '); //this space is really important.
            appender.finish();
            Thread.sleep(100);

            if(i % 10==0) {
                System.out.print(".");
            }
        }

        appender.close();
        System.out.print("\n");

        //create a tailer to get the first 50 items then exit the tailer
        final Chronicle sink1 = ChronicleQueueBuilder.sink(
                ChronicleQueueBuilder.vanilla(sinkBasePath)
                    .entriesPerCycle(1L << 20)
                    .cycleLength(1000, false)
                    .cycleFormat("yyyyMMddHHmmss")
                    .indexBlockSize(16L << 10)
                    .build())
            .connectAddress("localhost", BASE_PORT + 104)
            .build();

        final ExcerptTailer tailer1 = sink1.createTailer().toStart();

        System.out.println("Sink1 reading first 50 items then stopping");
        for( int count=0; count < 50 ;) {
            if(tailer1.nextIndex()) {
                assertEquals(1000000000 + count, tailer1.parseLong());
                tailer1.finish();

                count++;
            }
        }

        tailer1.close();
        sink1.close();
        //TODO: fix sink1.checkCounts(1, 1);

        //now resume the tailer to get the first 50 items
        final Chronicle sink2 = ChronicleQueueBuilder.sink(
            ChronicleQueueBuilder.vanilla(sinkBasePath)
                .entriesPerCycle(1L << 20)
                .cycleLength(1000, false)
                .cycleFormat("yyyyMMddHHmmss")
                .indexBlockSize(16L << 10)
                .build())
            .connectAddress("localhost", BASE_PORT + 104)
            .build();

        //Take the tailer to the last index (item 50) and start reading from there.
        final ExcerptTailer tailer2 = sink2.createTailer().toEnd();
        assertEquals(1000000000 + 49, tailer2.parseLong());
        tailer2.finish();
        
        System.out.println("Sink2 restarting to continue to read the next 50 items");
        for(int count=50 ; count < 100 ; ) {
            if(tailer2.nextIndex()) {
                assertEquals(1000000000 + count, tailer2.parseLong());
                tailer2.finish();

                count++;
            }
        }

        tailer2.close();
        sink2.close();
        //TODO: fix sink2.checkCounts(1, 1);

        sink2.clear();

        source.close();
        //TODO: fix source.checkCounts(1, 1);
        source.clear();

        assertFalse(new File(sourceBasePath).exists());
        assertFalse(new File(sinkBasePath).exists());
    }
#vulnerable code 
                                @Test
    public void testSourceSinkStartResumeRollingEverySecond() throws Exception {
        //This is the config that is required to make the VanillaChronicle roll every second

        final String sourceBasePath = getVanillaTestPath("-source");
        final String sinkBasePath = getVanillaTestPath("-sink");
        assertNotNull(sourceBasePath);
        assertNotNull(sinkBasePath);

        final ChronicleSource source = new ChronicleSource(
            ChronicleQueueBuilder.vanilla(sourceBasePath)
                .entriesPerCycle(1L << 20)
                .cycleLength(1000, false)
                .cycleFormat("yyyyMMddHHmmss")
                .indexBlockSize(16L << 10)
                .build(),
            8888);

        ExcerptAppender appender = source.createAppender();
        System.out.print("writing 100 items will take take 10 seconds.");
        for (int i = 0; i < 100; i++) {
            appender.startExcerpt();
            int value = 1000000000 + i;
            appender.append(value).append(' '); //this space is really important.
            appender.finish();
            Thread.sleep(100);

            if(i % 10==0) {
                System.out.print(".");
            }
        }

        appender.close();
        System.out.print("\n");

        //create a tailer to get the first 50 items then exit the tailer
        final ChronicleSink sink1 = new ChronicleSink(
            ChronicleQueueBuilder.vanilla(sinkBasePath)
                .entriesPerCycle(1L << 20)
                .cycleLength(1000, false)
                .cycleFormat("yyyyMMddHHmmss")
                .indexBlockSize(16L << 10)
                .build(),
            "localhost",
            8888);

        final ExcerptTailer tailer1 = sink1.createTailer().toStart();

        System.out.println("Sink1 reading first 50 items then stopping");
        for( int count=0; count < 50 ;) {
            if(tailer1.nextIndex()) {
                assertEquals(1000000000 + count, tailer1.parseLong());
                tailer1.finish();

                count++;
            }
        }

        tailer1.close();
        sink1.close();
        sink1.checkCounts(1, 1);

        //now resume the tailer to get the first 50 items
        final ChronicleSink sink2 = new ChronicleSink(
            ChronicleQueueBuilder.vanilla(sinkBasePath)
                .entriesPerCycle(1L << 20)
                .cycleLength(1000, false)
                .cycleFormat("yyyyMMddHHmmss")
                .indexBlockSize(16L << 10)
                .build(),
            "localhost",
            8888);

        //Take the tailer to the last index (item 50) and start reading from there.
        final ExcerptTailer tailer2 = sink2.createTailer().toEnd();
        assertEquals(1000000000 + 49, tailer2.parseLong());
        tailer2.finish();
        
        System.out.println("Sink2 restarting to continue to read the next 50 items");
        for(int count=50 ; count < 100 ; ) {
            if(tailer2.nextIndex()) {
                assertEquals(1000000000 + count, tailer2.parseLong());
                tailer2.finish();

                count++;
            }
        }

        tailer2.close();
        sink2.close();
        sink2.checkCounts(1, 1);

        sink2.clear();

        source.close();
        source.checkCounts(1, 1);
        source.clear();

        assertFalse(new File(sourceBasePath).exists());
        assertFalse(new File(sinkBasePath).exists());
    } 
                            #location 30 
                            #vulnerability type RESOURCE_LEAK