private StreamObserver<RemoteMessage> createStreamObserver() {
        int sleepTotalMillis = 0;
        int sleepMillis = 10;

        // Control the concurrency of gRPC streaming stub.
        // If over 10 created and not finished/error, this blocks the method.
        while (concurrentStreamObserverNumber.get() > 10) {
            try {
                Thread.sleep(sleepMillis);
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }

            sleepTotalMillis += sleepMillis;

            if (sleepTotalMillis > 60000) {
                log.warn(
                    "Remote client [{}] block times over 60 seconds. Current streaming number {}",
                    address, concurrentStreamObserverNumber.get()
                );
                // Reset sleepTotalMillis to avoid too many warn logs.
                sleepTotalMillis = 0;
            }
        }

        final StreamObserver<RemoteMessage> remoteMessageStreamObserver
            = getStub().withDeadlineAfter(remoteTimeout, TimeUnit.SECONDS)
                       .call(new StreamObserver<Empty>() {
                           @Override
                           public void onNext(Empty empty) {
                           }

                           @Override
                           public void onError(Throwable throwable) {
                               concurrentStreamObserverNumber.addAndGet(-1);
                               Status status = Status.fromThrowable(throwable);
                               if (Status.CANCELLED.getCode() == status.getCode()) {
                                   if (log.isDebugEnabled()) {
                                       log.debug(throwable.getMessage(), throwable);
                                   }
                                   return;
                               } else if (Status.RESOURCE_EXHAUSTED.getCode() == status.getCode()) {
                                   if (System.currentTimeMillis() - lastRemoteResourceExhaustedTime > 120_000) {
                                       // Only output the log every 120 seconds.
                                       log.warn(throwable.getMessage(), throwable);
                                       lastRemoteResourceExhaustedTime = System.currentTimeMillis();
                                   }
                                   return;
                               }
                               log.error(throwable.getMessage(), throwable);
                           }

                           @Override
                           public void onCompleted() {
                               concurrentStreamObserverNumber.addAndGet(-1);
                           }
                       });
        concurrentStreamObserverNumber.incrementAndGet();
        return remoteMessageStreamObserver;
    }
Create a gRPC stream observer to sending stream data, one stream observer could send multiple stream data by a
single consume. The max number of concurrency allowed at the same time is 10.

@return stream observer
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java