#fixed code
                                PeerEventServiceClient(Peer peer, Endpoint endpoint, Properties properties, PeerOptions peerOptions) {

        this.channelBuilder = endpoint.getChannelBuilder();
        this.filterBlock = peerOptions.isRegisterEventsForFilteredBlocks();
        this.peer = peer;
        name = peer.getName();
        url = peer.getUrl();
        channelName = peer.getChannel().getName();
        this.peerOptions = peerOptions;
        clientTLSCertificateDigest = endpoint.getClientTLSCertificateDigest();

        this.channelEventQue = peer.getChannel().getChannelEventQue();

        if (null == properties) {

            peerEventRegistrationWaitTimeMilliSecs = PEER_EVENT_REGISTRATION_WAIT_TIME;

        } else {

            String peerEventRegistrationWaitTime = properties.getProperty("peerEventRegistrationWaitTime", Long.toString(PEER_EVENT_REGISTRATION_WAIT_TIME));

            long tempPeerWaitTimeMilliSecs = PEER_EVENT_REGISTRATION_WAIT_TIME;

            try {
                tempPeerWaitTimeMilliSecs = Long.parseLong(peerEventRegistrationWaitTime);
            } catch (NumberFormatException e) {
                logger.warn(format("Peer event service registration %s wait time %s not parsable.", name, peerEventRegistrationWaitTime), e);
            }

            peerEventRegistrationWaitTimeMilliSecs = tempPeerWaitTimeMilliSecs;
        }

    }
#vulnerable code 
                                void connectEnvelope(Envelope envelope) throws TransactionException {

        if (shutdown) {
            logger.warn(format("Peer %s not connecting is shutdown ", peer));
            return;
        }

        ManagedChannel lmanagedChannel = managedChannel;

        if (lmanagedChannel == null || lmanagedChannel.isTerminated() || lmanagedChannel.isShutdown()) {

            lmanagedChannel = channelBuilder.build();
            managedChannel = lmanagedChannel;

        }

        try {

            DeliverGrpc.DeliverStub broadcast = DeliverGrpc.newStub(lmanagedChannel);

            // final DeliverResponse[] ret = new DeliverResponse[1];
            //   final List<DeliverResponse> retList = new ArrayList<>();
            final List<Throwable> throwableList = new ArrayList<>();
            final CountDownLatch finishLatch = new CountDownLatch(1);

            so = new StreamObserver<DeliverResponse>() {

                @Override
                public void onNext(DeliverResponse resp) {

                    // logger.info("Got Broadcast response: " + resp);
                    logger.trace(format("DeliverResponse channel %s peer %s resp status value:%d  status %s, typecase %s ",
                            channelName, peer.getName(), resp.getStatusValue(), resp.getStatus(), resp.getTypeCase()));

                    final DeliverResponse.TypeCase typeCase = resp.getTypeCase();

                    if (typeCase == STATUS) {

                        logger.debug(format("DeliverResponse channel %s peer %s setting done.",
                                channelName, peer.getName()));

                        if (resp.getStatus() == Common.Status.SUCCESS) { // unlike you may think this only happens when all blocks are fetched.
                            peer.setLastConnectTime(System.currentTimeMillis());
                            peer.resetReconnectCount();
                        } else {

                            throwableList.add(new TransactionException(format("Channel %s peer %s Status returned failure code %d (%s) during peer service event registration",
                                    channelName, peer.getName(), resp.getStatusValue(), resp.getStatus().name())));
                        }

                    } else if (typeCase == FILTERED_BLOCK || typeCase == BLOCK) {
                        if (typeCase == BLOCK) {
                            logger.trace(format("Channel %s peer %s got event block hex hashcode: %016x, block number: %d",
                                    channelName, peer.getName(), resp.getBlock().hashCode(), resp.getBlock().getHeader().getNumber()));
                        } else {
                            logger.trace(format("Channel %s peer %s got event block hex hashcode: %016x, block number: %d",
                                    channelName, peer.getName(), resp.getFilteredBlock().hashCode(), resp.getFilteredBlock().getNumber()));
                        }

                        peer.setLastConnectTime(System.currentTimeMillis());
                        long reconnectCount = peer.getReconnectCount();
                        if (reconnectCount > 1) {

                            logger.info(format("Peer eventing service reconnected after %d attempts on channel %s, peer %s, url %s",
                                    reconnectCount, channelName, name, url));

                        }
                        peer.resetReconnectCount();

                        BlockEvent blockEvent = new BlockEvent(peer, resp);
                        peer.setLastBlockSeen(blockEvent);

                        channelEventQue.addBEvent(blockEvent);
                    } else {
                        logger.error(format("Channel %s peer %s got event block with unknown type: %s, %d",
                                channelName, peer.getName(), typeCase.name(), typeCase.getNumber()));

                        throwableList.add(new TransactionException(format("Channel %s peer %s Status got unknown type %s, %d",
                                channelName, peer.getName(), typeCase.name(), typeCase.getNumber())));

                    }
                    finishLatch.countDown();

                }

                @Override
                public void onError(Throwable t) {
                    ManagedChannel llmanagedChannel = managedChannel;
                    if (llmanagedChannel != null) {
                        llmanagedChannel.shutdownNow();
                        managedChannel = null;
                    }
                    if (!shutdown) {
                        final long reconnectCount = peer.getReconnectCount();
                        if (PEER_EVENT_RECONNECTION_WARNING_RATE > 1 && reconnectCount % PEER_EVENT_RECONNECTION_WARNING_RATE == 1) {
                            logger.warn(format("Received error on peer eventing service on channel %s, peer %s, url %s, attempts %d. %s",
                                    channelName, name, url, reconnectCount, t.getMessage()));

                        } else {
                            logger.trace(format("Received error on peer eventing service on channel %s, peer %s, url %s, attempts %d. %s",
                                    channelName, name, url, reconnectCount, t.getMessage()));

                        }

                        peer.reconnectPeerEventServiceClient(PeerEventServiceClient.this, t);

                    }
                    finishLatch.countDown();
                }

                @Override
                public void onCompleted() {
                    logger.debug(format("DeliverResponse onCompleted channel %s peer %s setting done.",
                            channelName, peer.getName()));
                    //            done = true;
                    //There should have been a done before this...
                    finishLatch.countDown();
                }
            };

            nso = filterBlock ? broadcast.deliverFiltered(so) : broadcast.deliver(so);

            nso.onNext(envelope);

            // try {
            if (!finishLatch.await(peerEventRegistrationWaitTimeMilliSecs, TimeUnit.MILLISECONDS)) {
                TransactionException ex = new TransactionException(format(
                        "Channel %s connect time exceeded for peer eventing service %s, timed out at %d ms.", channelName, name, peerEventRegistrationWaitTimeMilliSecs));
                throwableList.add(0, ex);

            }
            logger.trace("Done waiting for reply!");

            if (!throwableList.isEmpty()) {
                ManagedChannel llmanagedChannel = managedChannel;
                if (llmanagedChannel != null) {
                    llmanagedChannel.shutdownNow();
                    managedChannel = null;
                }
                Throwable throwable = throwableList.get(0);
                peer.reconnectPeerEventServiceClient(this, throwable);

            }

        } catch (InterruptedException e) {
            ManagedChannel llmanagedChannel = managedChannel;
            if (llmanagedChannel != null) {
                llmanagedChannel.shutdownNow();
                managedChannel = null;
            }
            logger.error(e); // not likely

            peer.reconnectPeerEventServiceClient(this, e);

        } finally {
            if (null != nso) {

                try {
                    nso.onCompleted();
                } catch (Exception e) {  //Best effort only report on debug
                    logger.debug(format("Exception completing connect with channel %s,  name %s, url %s %s",
                            channelName, name, url, e.getMessage()), e);
                }

            }
        }
    } 
                            #location 141 
                            #vulnerability type THREAD_SAFETY_VIOLATION