[Bug] after apache/skywalking-java#457, cross-thread grpc streaming is broken.

[h0cheung]

Search before asking

• I had searched in the issues and found no similar issues.

Apache SkyWalking Component

Java Agent (apache/skywalking-java)

What happened

If StreamObserver#onNext is called in another thread, we will see exception: IllegalArgumentException("ContextSnapshot can't be null.")

After apache/skywalking-java#457, we use a io.grpc.Context to pass the ContextSnapshot. But by default, io.grpc.Context uses ThreadLocal as storage. So it won't work across threads. Then the contextSnapshot will be null, which causes the exception above.

I think we should partially revert the PR. Using the constructor to pass contexts is good.

What you expected to happen

It should work normally

How to reproduce

the simplest way is write a handler like this:

    public StreamObserver<Request> connect(StreamObserver<Response> responseObserver) {
        new Thread(()-> {
            responseObserver.onNext(Response.newBuilder().build());
        }).start();
        return new StreamObserver<>() {
            @Override
            public void onNext(Request value) {
                
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onCompleted() {

            }
        }

Then call it with a client, such as grpcurl.

Anything else

No response

Are you willing to submit a pull request to fix on your own?

• Yes I am willing to submit a pull request on my own!

Code of Conduct

• I agree to follow this project's Code of Conduct

[wu-sheng]

As this is relative to the previous change, @devkanro could you take a look here?

@h0cheung TBO, I am not familiar with the way you used through gRPC. The new thread is created by yourself, and I am not sure whether the plugin could adopt that.

[wu-sheng]

Usually, we return the observer and send the data through the return of gRPC method.

StreamObserver<SegmentObject> upstreamSegmentStreamObserver = serviceStub.withDeadlineAfter(
                Config.Collector.GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS
            ).collect(new StreamObserver<Commands>() {
....

upstreamSegmentStreamObserver.onNext(upstreamSegment);

[wu-sheng]

But you created another thread to do the send, I am not familiar with this use case.

The user thread usually breaks the tracing context, and users usually should propagate the context.
Meanwhile, I understand, this time is a little tricky, because it breaks plugin's internal propagation mechanism due the ThreadLocal as you mentioned.

What I am not sure about is, whether this ThreadLocal thing was added with purposes, and important inside gRPC plugin.

[h0cheung]

The actual scenario is subscribing. These observers are registered to a centralized subscription management component, which runs in its own thread. It will send data to the client via the stream when a specific event occurs.

[wu-sheng]

I tried to change the codes like this, is this fitting your use case?

apache/skywalking-java#741