fix: Fix race condition in GrpcDirectStreamController

[mutianf]

Fix the race condition in GrpcDirectStreamController when controller.request() is called when there's a retry attempt. This could cause IllegalStateException:

java.lang.IllegalStateException: Not started

	at com.google.common.base.Preconditions.checkState(Preconditions.java:502)
	at io.grpc.internal.ClientCallImpl.request(ClientCallImpl.java:451)
	at com.google.api.gax.grpc.GrpcDirectStreamController.request(GrpcDirectStreamController.java:95)
	at com.google.api.gax.grpc.ExceptionResponseObserver$1.request(ExceptionResponseObserver.java:67)
	at com.google.api.gax.rpc.ServerStreamingAttemptCallable.onRequest(ServerStreamingAttemptCallable.java:345)
	at com.google.api.gax.rpc.ServerStreamingAttemptCallable.access$200(ServerStreamingAttemptCallable.java:97)
	at com.google.api.gax.rpc.ServerStreamingAttemptCallable$1.request(ServerStreamingAttemptCallable.java:168)
	at com.google.api.gax.tracing.TracedResponseObserver$1.request(TracedResponseObserver.java:83)
	at com.google.api.gax.rpc.ServerStreamingAttemptCallable.onRequest(ServerStreamingAttemptCallable.java:345)
	at com.google.api.gax.rpc.ServerStreamingAttemptCallable.access$200(ServerStreamingAttemptCallable.java:97)
	at com.google.api.gax.rpc.ServerStreamingAttemptCallable$1.request(ServerStreamingAttemptCallable.java:168)

This get triggered when GrpcDirectStreamController is created from a retry thread and controller.request() get called from another thread:

this.hasStarted = true;

// controller.request() called from the another thread

clientCall.start(new ResponseObserverAdapter(), new Metadata());

[sonarcloud]

[gapic-generator-java-root] Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
1 Code Smell

100.0% Coverage
0.0% Duplication

[blakeli0]

Why this line is moved from before clientCall.start to after it? I don't know the exact use case of startCommon, but it could be an issue if clientCall.start is taking some time to complete and startCommon is being called another time.

[mutianf]

hasStarted is meant to be a barrier before we could interact with a stream, grpc has the invariant that you can't call request before a stream is started. Before the barrier is opened, we simply increment the request count. The original assumption is that the only thing that could call request() this is in onStart() and onResponse(). However this is not the case when retry is enabled where controller could be created from a different thread.

I don't think clientCall.start() taking a long time will be a problem here. GrpcDirectStreamController#start() is only called from here and here for bidi, which means that every callable.call() will create a new controller instance, so I don't think there's a race condition here.

[blakeli0]

Thanks, it makes sense.