#fixed code
                                @Test(timeout = 10000L)
    public void testNoDeadlockOnFullQueue() {
        //
        // Fixes https://github.com/awslabs/amazon-kinesis-client/issues/448
        //
        // This test is to verify that the drain of a blocked queue no longer deadlocks.
        // If the test times out before starting the subscriber it means something went wrong while filling the queue.
        // After the subscriber is started one of the things that can trigger a timeout is a deadlock.
        //

        final int[] sequenceNumberInResponse = { 0 };

        when(getRecordsRetrievalStrategy.getRecords(anyInt())).thenAnswer( i -> GetRecordsResponse.builder().records(
                Record.builder().data(SdkBytes.fromByteArray(new byte[] { 1, 2, 3 })).sequenceNumber(++sequenceNumberInResponse[0] + "").build())
                .nextShardIterator(NEXT_SHARD_ITERATOR).build());

        getRecordsCache.start(sequenceNumber, initialPosition);

        //
        // Wait for the queue to fill up, and the publisher to block on adding items to the queue.
        //
        log.info("Waiting for queue to fill up");
        while (getRecordsCache.getPublisherSession().prefetchRecordsQueue().size() < MAX_SIZE) {
            Thread.yield();
        }

        log.info("Queue is currently at {} starting subscriber", getRecordsCache.getPublisherSession().prefetchRecordsQueue().size());
        AtomicInteger receivedItems = new AtomicInteger(0);

        final int expectedItems = MAX_SIZE * 10;

        Object lock = new Object();

        final boolean[] isRecordNotInorder = { false };
        final String[] recordNotInOrderMessage = { "" };

        Subscriber<RecordsRetrieved> delegateSubscriber = new Subscriber<RecordsRetrieved>() {
            Subscription sub;
            int receivedSeqNum = 0;

            @Override
            public void onSubscribe(Subscription s) {
                sub = s;
                s.request(1);
            }

            @Override
            public void onNext(RecordsRetrieved recordsRetrieved) {
                receivedItems.incrementAndGet();
                if (Integer.parseInt(((PrefetchRecordsPublisher.PrefetchRecordsRetrieved) recordsRetrieved)
                        .lastBatchSequenceNumber()) != ++receivedSeqNum) {
                    isRecordNotInorder[0] = true;
                    recordNotInOrderMessage[0] = "Expected : " + receivedSeqNum + " Actual : "
                            + ((PrefetchRecordsPublisher.PrefetchRecordsRetrieved) recordsRetrieved)
                            .lastBatchSequenceNumber();
                }
                if (receivedItems.get() >= expectedItems) {
                    synchronized (lock) {
                        log.info("Notifying waiters");
                        lock.notifyAll();
                    }
                    sub.cancel();
                } else {
                    sub.request(1);
                }
            }

            @Override
            public void onError(Throwable t) {
                log.error("Caught error", t);
                throw new RuntimeException(t);
            }

            @Override
            public void onComplete() {
                fail("onComplete not expected in this test");
            }
        };

        Subscriber<RecordsRetrieved> subscriber = new ShardConsumerNotifyingSubscriber(delegateSubscriber, getRecordsCache);

        synchronized (lock) {
            log.info("Awaiting notification");
            Flowable.fromPublisher(getRecordsCache).subscribeOn(Schedulers.computation())
                    .observeOn(Schedulers.computation(), true, 8).subscribe(subscriber);
            try {
                lock.wait();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        verify(getRecordsRetrievalStrategy, atLeast(expectedItems)).getRecords(anyInt());
        assertThat(receivedItems.get(), equalTo(expectedItems));
        assertFalse(recordNotInOrderMessage[0], isRecordNotInorder[0]);
    }
#vulnerable code 
                                @Test(timeout = 10000L)
    public void testNoDeadlockOnFullQueue() {
        //
        // Fixes https://github.com/awslabs/amazon-kinesis-client/issues/448
        //
        // This test is to verify that the drain of a blocked queue no longer deadlocks.
        // If the test times out before starting the subscriber it means something went wrong while filling the queue.
        // After the subscriber is started one of the things that can trigger a timeout is a deadlock.
        //

        final int[] sequenceNumberInResponse = { 0 };

        when(getRecordsRetrievalStrategy.getRecords(anyInt())).thenAnswer( i -> GetRecordsResponse.builder().records(
                Record.builder().data(SdkBytes.fromByteArray(new byte[] { 1, 2, 3 })).sequenceNumber(++sequenceNumberInResponse[0] + "").build())
                .nextShardIterator(NEXT_SHARD_ITERATOR).build());

        getRecordsCache.start(sequenceNumber, initialPosition);

        //
        // Wait for the queue to fill up, and the publisher to block on adding items to the queue.
        //
        log.info("Waiting for queue to fill up");
        while (getRecordsCache.getPublisherSession().prefetchRecordsQueue().size() < MAX_SIZE) {
            Thread.yield();
        }

        log.info("Queue is currently at {} starting subscriber", getRecordsCache.getPublisherSession().prefetchRecordsQueue().size());
        AtomicInteger receivedItems = new AtomicInteger(0);
        
        final int expectedItems = MAX_SIZE * 10;

        Object lock = new Object();

        final boolean[] isRecordNotInorder = { false };
        final String[] recordNotInOrderMessage = { "" };

        Subscriber<RecordsRetrieved> delegateSubscriber = new Subscriber<RecordsRetrieved>() {
            Subscription sub;
            int receivedSeqNum = 0;

            @Override
            public void onSubscribe(Subscription s) {
                sub = s;
                s.request(1);
            }

            @Override
            public void onNext(RecordsRetrieved recordsRetrieved) {
                receivedItems.incrementAndGet();
                if (Integer.parseInt(((PrefetchRecordsPublisher.PrefetchRecordsRetrieved) recordsRetrieved)
                        .lastBatchSequenceNumber()) != ++receivedSeqNum) {
                    isRecordNotInorder[0] = true;
                    recordNotInOrderMessage[0] = "Expected : " + receivedSeqNum + " Actual : "
                            + ((PrefetchRecordsPublisher.PrefetchRecordsRetrieved) recordsRetrieved)
                            .lastBatchSequenceNumber();
                }
                if (receivedItems.get() >= expectedItems) {
                    synchronized (lock) {
                        log.info("Notifying waiters");
                        lock.notifyAll();
                    }
                    sub.cancel();
                } else {
                    sub.request(1);
                }
            }

            @Override
            public void onError(Throwable t) {
                log.error("Caught error", t);
                throw new RuntimeException(t);
            }

            @Override
            public void onComplete() {
                fail("onComplete not expected in this test");
            }
        };

        Subscriber<RecordsRetrieved> subscriber = new ShardConsumerNotifyingSubscriber(delegateSubscriber, getRecordsCache);

        synchronized (lock) {
            log.info("Awaiting notification");
            Flowable.fromPublisher(getRecordsCache).subscribeOn(Schedulers.computation())
                    .observeOn(Schedulers.computation(), true, 8).subscribe(subscriber);
            try {
                lock.wait();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        verify(getRecordsRetrievalStrategy, atLeast(expectedItems)).getRecords(anyInt());
        assertThat(receivedItems.get(), equalTo(expectedItems));
        assertFalse(recordNotInOrderMessage[0], isRecordNotInorder[0]);
    } 
                            #location 80 
                            #vulnerability type THREAD_SAFETY_VIOLATION