bitflicker64 commented on code in PR #3157:
URL: https://github.com/apache/hugegraph/pull/3157#discussion_r3885719895
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -180,50 +200,144 @@ public void onNext(WatchResponse value) {
@Override
public void onError(Throwable t) {
- release();
- if (!closed.get()) {
- clientId.set(0);
- listenWrapper.accept(key, consumer);
+ if (isRetryableWatchError(t)) {
+ requestReconnect(subscription, this);
+ } else {
+ stopWatch(subscription, this, t);
}
}
@Override
public void onCompleted() {
-
+ requestReconnect(subscription, this);
}
};
}
public void listen(String key, Consumer<T> consumer) throws PDException {
- long value = clientId.get();
- StreamObserver<WatchResponse> observer = getObserver(key, consumer,
listenWrapper, value);
- acquire();
+ listen(key, consumer, false);
+ }
+
+ public void listenPrefix(String prefix, Consumer<T> consumer) throws
PDException {
+ listen(prefix, consumer, true);
+ }
+
+ private void listen(String key, Consumer<T> consumer, boolean prefix)
throws PDException {
+ WatchSubscription subscription = new WatchSubscription(key, consumer,
prefix);
+ subscriptions.add(subscription);
try {
- WatchRequest k =
-
WatchRequest.newBuilder().setClientId(clientId.get()).setKey(key).build();
- streamingCall(KvServiceGrpc.getWatchMethod(), k, observer, 1);
- } catch (Exception e) {
- release();
- throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, e);
+ if (!startWatch(subscription)) {
+ throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE,
+ "KvClient is closed");
+ }
+ } catch (PDException e) {
+ subscription.observer.set(null);
+ subscriptions.remove(subscription);
+ throw e;
}
}
- public void listenPrefix(String prefix, Consumer<T> consumer) throws
PDException {
- long value = clientId.get();
- StreamObserver<WatchResponse> observer =
- getObserver(prefix, consumer, prefixListenWrapper, value);
- acquire();
+ private boolean startWatch(WatchSubscription subscription) throws
PDException {
+ if (closed.get()) {
+ return false;
+ }
+
+ StreamObserver<WatchResponse> observer = getObserver(subscription);
+ subscription.observer.set(observer);
+ if (closed.get()) {
+ subscription.observer.compareAndSet(observer, null);
+ return false;
+ }
+
+ acquire(watchClientId, watchSemaphore);
Review Comment:
โ ๏ธ All reconnects share one thread that can block here indefinitely, so one
stalled stream stops every watch from recovering.
`acquire(watchClientId, watchSemaphore)` performs an untimed
`semaphore.acquire()` whenever `watchClientId == 0` (`KvClient.java:340-352`).
The permit only comes back when some watch receives `Starting`
(`KvClient.java:185`), fails (`KvClient.java:286`), or is stopped
(`KvClient.java:301`). Reconnects all run on the single-thread executor created
at `KvClient.java:85-89`.
That combination is exercised by an ordinary leader change:
`KvWatchSubject.notifyClientChangeLeader()` sends `Leader_Changed` to every
observer and then calls `removeClient(...)`, which completes each stream, so
all subscriptions call `requestReconnect` (resetting `watchClientId` to 0) and
queue on that one thread. The first task re-issues its stream and returns; the
second parks in `acquire` until the first stream answers. Since `resetStub()`
builds channels with `ManagedChannelBuilder.forTarget(host).usePlaintext()` and
no keepalive (`AbstractClient.java:142`), a half-open connection that never
delivers `Starting` and never errors parks that thread for good, and no
subscription in the client can reconnect.
Requested change: replace the untimed `acquire` with `tryAcquire(timeout,
unit)` and treat a timeout as a failed attempt that reschedules, and/or give
the reconnect executor more than one thread so a stalled subscription cannot
block the others.
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -180,50 +200,144 @@ public void onNext(WatchResponse value) {
@Override
public void onError(Throwable t) {
- release();
- if (!closed.get()) {
- clientId.set(0);
- listenWrapper.accept(key, consumer);
+ if (isRetryableWatchError(t)) {
+ requestReconnect(subscription, this);
+ } else {
+ stopWatch(subscription, this, t);
}
}
@Override
public void onCompleted() {
-
+ requestReconnect(subscription, this);
}
};
}
public void listen(String key, Consumer<T> consumer) throws PDException {
- long value = clientId.get();
- StreamObserver<WatchResponse> observer = getObserver(key, consumer,
listenWrapper, value);
- acquire();
+ listen(key, consumer, false);
+ }
+
+ public void listenPrefix(String prefix, Consumer<T> consumer) throws
PDException {
+ listen(prefix, consumer, true);
+ }
+
+ private void listen(String key, Consumer<T> consumer, boolean prefix)
throws PDException {
+ WatchSubscription subscription = new WatchSubscription(key, consumer,
prefix);
+ subscriptions.add(subscription);
try {
- WatchRequest k =
-
WatchRequest.newBuilder().setClientId(clientId.get()).setKey(key).build();
- streamingCall(KvServiceGrpc.getWatchMethod(), k, observer, 1);
- } catch (Exception e) {
- release();
- throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, e);
+ if (!startWatch(subscription)) {
+ throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE,
+ "KvClient is closed");
+ }
+ } catch (PDException e) {
+ subscription.observer.set(null);
+ subscriptions.remove(subscription);
+ throw e;
}
}
- public void listenPrefix(String prefix, Consumer<T> consumer) throws
PDException {
- long value = clientId.get();
- StreamObserver<WatchResponse> observer =
- getObserver(prefix, consumer, prefixListenWrapper, value);
- acquire();
+ private boolean startWatch(WatchSubscription subscription) throws
PDException {
+ if (closed.get()) {
+ return false;
+ }
+
+ StreamObserver<WatchResponse> observer = getObserver(subscription);
+ subscription.observer.set(observer);
+ if (closed.get()) {
+ subscription.observer.compareAndSet(observer, null);
+ return false;
+ }
+
+ acquire(watchClientId, watchSemaphore);
+ if (closed.get()) {
+ subscription.observer.compareAndSet(observer, null);
+ release(watchSemaphore);
+ return false;
+ }
+
+ WatchRequest request = WatchRequest.newBuilder()
+ .setClientId(watchClientId.get())
+ .setKey(subscription.key)
+ .build();
try {
- WatchRequest k =
-
WatchRequest.newBuilder().setClientId(clientId.get()).setKey(prefix).build();
- streamingCall(KvServiceGrpc.getWatchPrefixMethod(), k, observer,
1);
+ if (subscription.prefix) {
+ streamingCall(KvServiceGrpc.getWatchPrefixMethod(), request,
observer, 1);
+ } else {
+ streamingCall(KvServiceGrpc.getWatchMethod(), request,
observer, 1);
+ }
+ return true;
} catch (Exception e) {
- release();
+ release(watchSemaphore);
+ if (e instanceof PDException) {
+ throw (PDException) e;
+ }
throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, e);
}
}
- private void acquire() {
+ private void requestReconnect(WatchSubscription subscription,
+ StreamObserver<WatchResponse>
sourceObserver) {
+ if (closed.get() ||
+ !subscription.observer.compareAndSet(sourceObserver, null)) {
+ return;
+ }
+ watchClientId.set(0L);
+ release(watchSemaphore);
+ scheduleReconnect(subscription);
+ }
+
+ private static boolean isRetryableWatchError(Throwable throwable) {
+ Status.Code code = Status.fromThrowable(throwable).getCode();
+ return !NON_RETRYABLE_WATCH_ERRORS.contains(code);
+ }
+
+ private void stopWatch(WatchSubscription subscription,
+ StreamObserver<WatchResponse> sourceObserver,
+ Throwable throwable) {
+ if (!subscription.observer.compareAndSet(sourceObserver, null)) {
+ return;
+ }
+ release(watchSemaphore);
+ subscriptions.remove(subscription);
+ log.error("Watch for key {} stopped after a non-retryable error: {}",
Review Comment:
๐งน A permanently stopped watch is invisible to the caller.
`stopWatch()` drops the subscription from `subscriptions` and records a log
line, but the consumer registered through `listen`/`listenPrefix` is never
told. Both production callers โ `PdMetaDriver.listen`/`listenPrefix`
(`hugegraph-server/hugegraph-core/.../meta/PdMetaDriver.java:110-126`) and
`SchemaDriver.listen` (`hugegraph-struct/.../SchemaDriver.java:199-205`) โ only
see the initial `PDException`, so after a `CANCELLED`, `UNAUTHENTICATED` or
`PERMISSION_DENIED` stream error the application keeps running as if it were
still subscribed and the only signal is a log line.
Requested change: give the caller a way to observe termination โ for example
an optional error/termination callback on `listen`, or a handle returned from
`listen` whose state can be queried โ so a permanently stopped watch is
detectable without scraping logs.
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -180,50 +200,144 @@ public void onNext(WatchResponse value) {
@Override
public void onError(Throwable t) {
- release();
- if (!closed.get()) {
- clientId.set(0);
- listenWrapper.accept(key, consumer);
+ if (isRetryableWatchError(t)) {
+ requestReconnect(subscription, this);
+ } else {
+ stopWatch(subscription, this, t);
}
}
@Override
public void onCompleted() {
-
+ requestReconnect(subscription, this);
}
};
}
public void listen(String key, Consumer<T> consumer) throws PDException {
- long value = clientId.get();
- StreamObserver<WatchResponse> observer = getObserver(key, consumer,
listenWrapper, value);
- acquire();
+ listen(key, consumer, false);
+ }
+
+ public void listenPrefix(String prefix, Consumer<T> consumer) throws
PDException {
+ listen(prefix, consumer, true);
+ }
+
+ private void listen(String key, Consumer<T> consumer, boolean prefix)
throws PDException {
+ WatchSubscription subscription = new WatchSubscription(key, consumer,
prefix);
+ subscriptions.add(subscription);
try {
- WatchRequest k =
-
WatchRequest.newBuilder().setClientId(clientId.get()).setKey(key).build();
- streamingCall(KvServiceGrpc.getWatchMethod(), k, observer, 1);
- } catch (Exception e) {
- release();
- throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, e);
+ if (!startWatch(subscription)) {
+ throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE,
+ "KvClient is closed");
+ }
+ } catch (PDException e) {
+ subscription.observer.set(null);
+ subscriptions.remove(subscription);
+ throw e;
}
}
- public void listenPrefix(String prefix, Consumer<T> consumer) throws
PDException {
- long value = clientId.get();
- StreamObserver<WatchResponse> observer =
- getObserver(prefix, consumer, prefixListenWrapper, value);
- acquire();
+ private boolean startWatch(WatchSubscription subscription) throws
PDException {
+ if (closed.get()) {
+ return false;
+ }
+
+ StreamObserver<WatchResponse> observer = getObserver(subscription);
+ subscription.observer.set(observer);
+ if (closed.get()) {
+ subscription.observer.compareAndSet(observer, null);
+ return false;
+ }
+
+ acquire(watchClientId, watchSemaphore);
+ if (closed.get()) {
+ subscription.observer.compareAndSet(observer, null);
+ release(watchSemaphore);
+ return false;
+ }
+
+ WatchRequest request = WatchRequest.newBuilder()
+ .setClientId(watchClientId.get())
+ .setKey(subscription.key)
+ .build();
try {
- WatchRequest k =
-
WatchRequest.newBuilder().setClientId(clientId.get()).setKey(prefix).build();
- streamingCall(KvServiceGrpc.getWatchPrefixMethod(), k, observer,
1);
+ if (subscription.prefix) {
+ streamingCall(KvServiceGrpc.getWatchPrefixMethod(), request,
observer, 1);
+ } else {
+ streamingCall(KvServiceGrpc.getWatchMethod(), request,
observer, 1);
+ }
+ return true;
} catch (Exception e) {
- release();
+ release(watchSemaphore);
+ if (e instanceof PDException) {
+ throw (PDException) e;
+ }
throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, e);
}
}
- private void acquire() {
+ private void requestReconnect(WatchSubscription subscription,
+ StreamObserver<WatchResponse>
sourceObserver) {
+ if (closed.get() ||
+ !subscription.observer.compareAndSet(sourceObserver, null)) {
+ return;
+ }
+ watchClientId.set(0L);
+ release(watchSemaphore);
+ scheduleReconnect(subscription);
+ }
+
+ private static boolean isRetryableWatchError(Throwable throwable) {
+ Status.Code code = Status.fromThrowable(throwable).getCode();
+ return !NON_RETRYABLE_WATCH_ERRORS.contains(code);
+ }
+
+ private void stopWatch(WatchSubscription subscription,
+ StreamObserver<WatchResponse> sourceObserver,
+ Throwable throwable) {
+ if (!subscription.observer.compareAndSet(sourceObserver, null)) {
+ return;
+ }
+ release(watchSemaphore);
+ subscriptions.remove(subscription);
+ log.error("Watch for key {} stopped after a non-retryable error: {}",
+ subscription.key, Status.fromThrowable(throwable),
throwable);
+ }
+
+ private void scheduleReconnect(WatchSubscription subscription) {
+ if (closed.get() ||
!subscription.reconnectScheduled.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ reconnectExecutor.schedule(() -> reconnect(subscription),
RECONNECT_DELAY_MS,
+ TimeUnit.MILLISECONDS);
+ } catch (RuntimeException e) {
+ subscription.reconnectScheduled.set(false);
+ if (!closed.get()) {
+ log.warn("Failed to schedule watch reconnect for key {}",
subscription.key, e);
+ }
+ }
+ }
+
+ private void reconnect(WatchSubscription subscription) {
+ subscription.reconnectScheduled.set(false);
+ if (closed.get()) {
+ return;
+ }
+ try {
+ startWatch(subscription);
+ } catch (PDException e) {
Review Comment:
โ ๏ธ `reconnect()` catches only `PDException`, so an unchecked failure ends
the retry chain permanently.
`reconnect()` clears `subscription.reconnectScheduled` at the top and then
calls `startWatch` guarded only by `catch (PDException e)`. Any
`RuntimeException` escapes into the `ScheduledExecutorService`, where it is
swallowed into the (discarded) `ScheduledFuture`. At that point the flag is
already `false`, `subscription.observer` is null, and nothing reschedules โ the
subscription is left permanently dead, which is the exact failure mode this PR
removes.
This is reachable through the production stub path: `startWatch` โ
`streamingCall` โ `AbstractClient.getStub()` โ `resetStub()`. `resetStub()`
assigns `leaderHost` from the members response (`AbstractClient.java:150`)
*before* it assigns `proxy.setStub(...)` (`AbstractClient.java:156`). If stub
creation throws for every host, `resetStub()` returns a non-empty `leaderHost`
while `proxy.getStub()` is still null, and `getStub()` then evaluates
`setAsyncParams(null, config)` โ NPE, which is unchecked and outside the `try`
in `streamingCall`.
No test covers it either: `TestKvClient.streamingCall` only ever throws
`PDException` (`KvClientTest.java:752`).
Requested change: catch `Throwable` (or at least `RuntimeException`) here
and route it through the same `scheduleReconnect(subscription)` path โ or move
the rescheduling into a `finally` โ and add a test whose stubbed
`streamingCall` throws an unchecked exception, asserting a further reconnect is
still scheduled.
--
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]
