imbajin commented on code in PR #3157:
URL: https://github.com/apache/hugegraph/pull/3157#discussion_r4040953789
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java:
##########
@@ -249,39 +326,70 @@ protected <ReqT, RespT> KVPair<Boolean, RespT>
concurrentBlockingUnaryCall(
protected <ReqT, RespT> void streamingCall(MethodDescriptor<ReqT, RespT>
method, ReqT request,
StreamObserver<RespT>
responseObserver,
int retry) throws PDException {
- AbstractStub stub = getStub();
+ AbstractStub stub;
+ Channel attemptChannel;
+ synchronized (this) {
+ stub = getStub();
+ AbstractStub currentStub = proxy.getStub();
+ attemptChannel = currentStub == null ? stub.getChannel() :
currentStub.getChannel();
+ Consumer<Channel> attemptConsumer =
this.streamingAttemptConsumer.get();
+ if (attemptConsumer != null) {
+ attemptConsumer.accept(attemptChannel);
+ }
+ }
try {
ClientCall<ReqT, RespT> call = stub.getChannel().newCall(method,
stub.getCallOptions());
ClientCalls.asyncServerStreamingCall(call, request,
responseObserver);
} catch (Exception e) {
log.error("rpc call with exception :", e);
if (e instanceof StatusRuntimeException) {
if (retry < proxy.getHostCount()) {
- synchronized (this) {
- proxy.setStub(null);
- }
+ invalidateAsyncStub(attemptChannel);
streamingCall(method, request, responseObserver, ++retry);
+ return;
}
}
+ throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE,
+ "RPC streaming call failed", e);
+ }
+ }
+
+ protected <ReqT, RespT> void streamingCall(MethodDescriptor<ReqT, RespT>
method, ReqT request,
+ StreamObserver<RespT>
responseObserver,
+ int retry,
+ Consumer<Channel>
attemptConsumer)
+ throws PDException {
+ Consumer<Channel> previous = this.streamingAttemptConsumer.get();
Review Comment:
A four-argument override that delegates to super retains the attempt
consumer; the ThreadLocal is restored/removed in finally. No production
override replacing this transport exists in the repository. Routing both
overloads directly to a private delegate would bypass the legacy virtual
override, contradicting the compatibility test. Custom transport/channel
tracking remains an extension-contract discussion, not a demonstrated
regression in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
