imbajin commented on code in PR #3157:
URL: https://github.com/apache/hugegraph/pull/3157#discussion_r4041234497
##########
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:
Closing for this PR: retain the existing virtual dispatch and
super-delegating override behavior. No affected production replacement override
has been identified; custom transport extensions can be addressed when a
concrete downstream contract requires them. No private-delegate rewrite is
needed here.
##########
hugegraph-struct/src/main/java/org/apache/hugegraph/SchemaDriver.java:
##########
@@ -98,24 +117,75 @@ public static void init(PDConfig pdConfig) {
}
public static void init(PDConfig pdConfig, int cacheSize, long expiration)
{
- SchemaDriver instance = INSTANCE.get();
- if (instance != null) {
- throw new NotAllowException(
- "The SchemaDriver [cacheSize=%s, expiration=%s, " +
- "client=%s] has already been initialized and is not " +
- "allowed to be initialized again", instance.caches.limit(),
- instance.caches.expiration(), instance.client);
+ synchronized (LIFECYCLE_LOCK) {
+ if (destroying) {
+ throw new NotAllowException("The SchemaDriver is being
destroyed");
Review Comment:
Closing with the accepted lifecycle boundary: callers coordinate
initialization, use and destruction. This PR does not guarantee successful
graph construction concurrent with global driver destruction; the existing
fail-fast behavior remains. No getOrInit/wait/retry redesign is needed for
watch recovery.
##########
hugegraph-struct/src/main/java/org/apache/hugegraph/SchemaDriver.java:
##########
@@ -98,24 +117,75 @@ public static void init(PDConfig pdConfig) {
}
public static void init(PDConfig pdConfig, int cacheSize, long expiration)
{
- SchemaDriver instance = INSTANCE.get();
- if (instance != null) {
- throw new NotAllowException(
- "The SchemaDriver [cacheSize=%s, expiration=%s, " +
- "client=%s] has already been initialized and is not " +
- "allowed to be initialized again", instance.caches.limit(),
- instance.caches.expiration(), instance.client);
+ synchronized (LIFECYCLE_LOCK) {
+ if (destroying) {
+ throw new NotAllowException("The SchemaDriver is being
destroyed");
+ }
+ SchemaDriver instance = INSTANCE.get();
+ if (instance != null) {
+ throw new NotAllowException(
+ "The SchemaDriver [cacheSize=%s, expiration=%s, " +
+ "client=%s] has already been initialized and is not " +
+ "allowed to be initialized again",
instance.caches.limit(),
+ instance.caches.expiration(), instance.client);
+ }
+ INSTANCE.set(new SchemaDriver(pdConfig, cacheSize, expiration));
Review Comment:
Accepted for this PR: initialization remains serialized with destruction,
including its PD discovery work. Callers coordinate lifecycle transitions;
adding an initialization/cancellation state machine would broaden this fix
without a current production requirement. Resolving on that scope decision.
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java:
##########
@@ -98,79 +104,154 @@ public static <T extends AbstractStub> T setAsyncParams(T
stub, PDConfig config)
new Authentication(config.getUserName(),
config.getAuthority()));
}
- protected AbstractBlockingStub getBlockingStub() throws PDException {
+ protected synchronized AbstractBlockingStub getBlockingStub() throws
PDException {
Review Comment:
Accepted tradeoff for this PR: ordinary KV/lock requests may experience
transient delay while watch recovery rotates the shared transport.
Recovery-time isolation is not a requirement here. Transport separation can be
considered independently if that requirement changes; an unlocked getter alone
would not provide it. Resolving with this explicit limitation, not as a claim
that the delay was eliminated.
--
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]
