#vulnerable code 
                                @Test
    public void clientToServerBackpressure() throws InterruptedException {
        Object lock = new Object();

        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(channel);
        BackpressureDetector clientBackpressureDetector = new BackpressureDetector(madMultipleCutoff);
        Sequence seq = new Sequence(200, clientBackpressureDetector);

        Flowable<NumberProto.Number> rxRequest = Flowable
                .fromIterable(seq)
                .doOnNext(i -> System.out.println(i + " -->"))
                .map(BackpressureIntegrationTest::protoNum);


        Single<NumberProto.Number> rxResponse = stub.requestPressure(rxRequest);

        rxResponse.subscribe(
                n -> {
                    System.out.println("Client done. " + n.getNumber(0));
                    synchronized (lock) {
                        lock.notify();
                    }
                },
                t -> {
                    t.printStackTrace();
                    synchronized (lock) {
                        lock.notify();
                    }
                });

        synchronized (lock) {
            lock.wait(TimeUnit.SECONDS.toMillis(20));
        }

        assertThat(clientBackpressureDetector.backpressureDelayOcurred()).isTrue();
    } 
                            #location 5 
                            #vulnerability type THREAD_SAFETY_VIOLATION
#fixed code
                                @Test
    public void clientToServerBackpressure() throws InterruptedException {
        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(channel);

        Flowable<NumberProto.Number> rxRequest = Flowable
                .fromIterable(IntStream.range(0, NUMBER_OF_STREAM_ELEMENTS)::iterator)
                .doOnNext(i -> System.out.println(i + " --> "))
                .doOnNext(i -> updateNumberOfWaits(clientLastValueTime, clientNbOfWaits))
                .map(BackpressureIntegrationTest::protoNum);

        TestObserver<NumberProto.Number> rxResponse = stub.requestPressure(rxRequest).test();

        rxResponse.awaitTerminalEvent(5, TimeUnit.SECONDS);
        rxResponse.assertComplete()
                .assertValue(v -> v.getNumber(0) == NUMBER_OF_STREAM_ELEMENTS - 1);

        assertThat(clientNbOfWaits.get()).isEqualTo(1);
    }