You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hugegraph.apache.org by "bitflicker64 (via GitHub)" <gi...@apache.org> on 2026/09/01 05:34:28 UTC
Re: [PR] fix(pd): keep KvClient watches alive after reconnect failures [hugegraph]
bitflicker64 commented on code in PR #3157:
URL: https://github.com/apache/hugegraph/pull/3157#discussion_r3901065262
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -75,6 +131,16 @@ protected AbstractBlockingStub createBlockingStub() {
return KvServiceGrpc.newBlockingStub(channel);
}
+ @Override
+ protected long stubResetTimeoutMillis() {
+ return Math.max(1L, Math.min(config.getGrpcTimeOut(), WATCH_START_TIMEOUT_MS));
Review Comment:
⚠️ Important. The watch-start budget is applied client-wide, so it also shrinks the PD reconnect budget for every blocking KV and lock call.
`stubResetTimeoutMillis()` is consumed by `AbstractClient.resetStub()` as the total deadline for the whole peer loop (AbstractClient.java:156-159), and `resetStub()` is reached from `getBlockingStub()` (AbstractClient.java:107-116) — not just from the watch path. That method backs `put`, `get`, `delete`, `deletePrefix`, `scanPrefix`, `putTTL`, `keepTTLAlive`, `lock`, `lockWithoutReentrant`, `isLocked`, `unlock` and `keepAlive`. With the default `PDConfig.grpcTimeOut = 60000` (PDConfig.java:34) this override returns 5000, and the derived per-peer deadline becomes `min(60000, remaining / remainingHosts)` — about 1.7 s each across three PD peers instead of 60 s.
Measured against three unreachable peers (`10.255.255.1-3:8686`), same `PDConfig`, same `getBlockingStub()` path:
| build | `KvClient.get()` | `MetaClient.getGraphs()` |
|---|---|---|
| base `7bb624b99` | 21328 / 35403 / 6511 ms | 37053 ms |
| head `f99b6bdd` | 4984 / 5213 / 5212 ms | 25071 ms |
At this head `KvClient` pins to ~5.0 s on every run — exactly `WATCH_START_TIMEOUT_MS` — while a sibling client on identical configuration keeps its multi-peer budget. So a configured `pd.grpc_timeout` of 60 s is silently honoured for `MetaClient` but capped at 5 s for `KvClient`. During a slow PD leader election or a GC pause longer than the per-peer slice, `lock()` and `keepAlive()` now fail with `PD_UNREACHABLE` where they previously waited. The same cap applies to store-node startup config reads (`PdConfigureListener.onApplicationEvent` calls `scanPrefix`/`put`) and to `PdMetaDriver`/`SchemaDriver`.
Requested change: scope the 5 s readiness budget to the watch-start path only — for example thread it through the async stub reset that `startWatch()` triggers, or gate the override on the async stub — and leave `getBlockingStub()` on the inherited `grpcTimeOut * hostCount` budget so `pd.grpc_timeout` keeps governing lock and KV traffic.
##########
hugegraph-struct/src/main/java/org/apache/hugegraph/SchemaDriver.java:
##########
@@ -98,24 +117,73 @@ 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) {
+ 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));
}
- INSTANCE.compareAndSet(null, new SchemaDriver(pdConfig, cacheSize,
- expiration));
}
public static void destroy() {
- SchemaDriver instance = INSTANCE.get();
- if (instance != null) {
- instance.caches.cancelScheduleCacheClean();
- instance.caches.destroyAll();
- INSTANCE.set(null);
+ SchemaDriver instance = null;
+ CountDownLatch completion;
+ boolean closeResources = false;
+ synchronized (LIFECYCLE_LOCK) {
Review Comment:
⚠️ Important. `destroy()` keeps `INSTANCE` published while the driver's `KvClient` is being closed, so callers can be handed a driver whose PD client is already shut down.
`INSTANCE` is only cleared in the `finally` at line 158-161, after `instance.closeResources()` (line 156) has already run `client.close()`. The new `testDestroyKeepsInstanceUntilResourcesAreClosed` asserts exactly this window. Meanwhile `SchemaGraph.schemaDriverInit()` (SchemaGraph.java:56-64) checks `SchemaDriver.getInstance() == null`, sees the still-published instance, skips `init()`, and returns it — and `SchemaGraph`'s constructor immediately calls `loadConfig()` -> `graphConfig()` -> `client.get()` on that driver.
By then `KvClient.close()` has set `closed = true`, shut down both reconnect executors and called `shutdownNow()` on the channel, so `AbstractClient.resetStub()` breaks out on its first `isShutdown()` check (AbstractClient.java:159) and `getBlockingStub()` throws `PD_UNREACHABLE`. On the base commit `destroy()` never closed the client, so a driver observed in this window still worked. `init()` is unusable in the window too: it takes `LIFECYCLE_LOCK`, sees the non-null `INSTANCE`, and throws `NotAllowException` (asserted by `testInitDoesNotWaitForDestroyCleanup`).
This is the mirror image of the earlier review note about clearing `INSTANCE` too early — the fix moved the hazard rather than removing it, because publication and usability are still two separate states.
Requested change: make the instance unreachable before its resources are torn down while still preventing a premature re-`init()` — for example clear `INSTANCE` inside the first `LIFECYCLE_LOCK` block and let the `destroying` flag alone block `init()` until cleanup finishes, so `getInstance()` never returns a driver with a closed client. Please add a test that calls `getInstance()` during the cleanup window and asserts the returned driver is either absent or still usable.
##########
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:
🧹 Minor. Routing `attemptConsumer` through a `ThreadLocal` makes it invisible to the documented 4-arg override point, which silently disables transport rotation.
This overload stores the consumer in `streamingAttemptConsumer` (line 63) and relies on the 4-arg `streamingCall` reading it back (line 335-338). A subclass that overrides the 4-arg method — which `testLegacyStreamingCallOverrideRemainsDispatchTarget` (KvClientTest.java:893, `LegacyStreamingOverrideKvClient` at KvClientTest.java:1502-1518) deliberately establishes as supported — never invokes the consumer.
`KvClient.recordAttemptChannel()` then never runs, so `subscription.attemptChannel` stays null. On `Leader_Changed`, `requestReconnect(..., rotateTransport = true)` copies that null into `reconnectChannel` (KvClient.java:429-432), `reconnect()` skips `invalidateAttemptStub()` (KvClient.java:556-558), the cached async stub survives, and the watch re-attaches to the same follower — the exact failure this PR is fixing, reintroduced for any subclass that keeps the legacy override.
The coupling is also implicit: the two calls only work because they happen on the same thread, and the value is keyed on a per-instance `ThreadLocal` whose entry is only reclaimed when the `finally` runs or the client is collected.
Requested change: pass the consumer explicitly rather than through thread state — have both public overloads delegate to one private implementation that takes `Consumer<Channel> attemptConsumer` as a parameter (null for the legacy signature), and drop the `ThreadLocal` field. That keeps the 4-arg override working as a dispatch target while making it obvious at the call site that an overrider must forward the consumer.
--
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org