#vulnerable code 
                                @Test
    public void multipleItemTest() throws Exception {
        addItemsToReturn(100);

        setupNotifierAnswer(recordsPublisher.responses.size());

        synchronized (processedNotifier) {
            subscriber.startSubscriptions();
            processedNotifier.wait(5000);
        }

        verify(shardConsumer, times(100)).handleInput(argThat(eqProcessRecordsInput(processRecordsInput)),
                any(Subscription.class));
    } 
                            #location 12 
                            #vulnerability type THREAD_SAFETY_VIOLATION
#fixed code
                                @Test
    public void multipleItemTest() throws Exception {
        addItemsToReturn(100);

        setupNotifierAnswer(recordsPublisher.responses.size());

        synchronized (processedNotifier) {
            subscriber.startSubscriptions();
            processedNotifier.wait(5000);
        }

        verify(shardConsumer, times(100)).handleInput(any(ProcessRecordsInput.class), any(Subscription.class));
    }