imbajin commented on code in PR #3157:
URL: https://github.com/apache/hugegraph/pull/3157#discussion_r3890884972
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -180,50 +200,170 @@ 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 {
+ return startWatch(subscription, true);
+ }
+
+ private boolean startWatch(WatchSubscription subscription,
+ boolean waitForPermit) 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;
+ }
+
+ if (waitForPermit) {
+ acquire(watchClientId, watchSemaphore);
+ } else if (!tryAcquire(watchClientId, watchSemaphore)) {
+ subscription.observer.compareAndSet(observer, null);
+ return false;
+ }
+ 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 {
+ if (!startWatch(subscription, false)) {
+ scheduleReconnect(subscription);
+ }
+ } catch (PDException e) {
Review Comment:
⚠️ Important. After peer retries are exhausted,
`AbstractClient.streamingCall()` wraps the final gRPC status as `PDException`.
`reconnect()` catches every `PDException` and schedules another attempt, so
permanent statuses such as permission or authentication failures are retried
forever without classification; `NON_RETRYABLE_WATCH_ERRORS` only protects the
async callback path. Preserve the gRPC status and stop/report non-retryable
synchronous failures.
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -56,13 +60,38 @@
@Slf4j
public class KvClient<T extends WatchResponse> extends AbstractClient
implements Closeable {
- private AtomicLong clientId = new AtomicLong(0);
- private Semaphore semaphore = new Semaphore(1);
- private AtomicBoolean closed = new AtomicBoolean(false);
- private Set<StreamObserver> observers = ConcurrentHashMap.newKeySet();
+ private static final long RECONNECT_DELAY_MS = 1000L;
+ private static final Set<Status.Code> NON_RETRYABLE_WATCH_ERRORS =
+ Set.of(Status.Code.CANCELLED,
+ Status.Code.INVALID_ARGUMENT,
+ Status.Code.NOT_FOUND,
+ Status.Code.ALREADY_EXISTS,
+ Status.Code.PERMISSION_DENIED,
+ Status.Code.FAILED_PRECONDITION,
+ Status.Code.OUT_OF_RANGE,
+ Status.Code.UNIMPLEMENTED,
+ Status.Code.DATA_LOSS,
+ Status.Code.UNAUTHENTICATED);
+
+ private final AtomicLong lockClientId = new AtomicLong(0);
+ private final AtomicLong watchClientId = new AtomicLong(0);
+ private final Semaphore lockSemaphore = new Semaphore(1);
+ private final Semaphore watchSemaphore = new Semaphore(1);
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final Set<WatchSubscription> subscriptions =
ConcurrentHashMap.newKeySet();
+ private final ScheduledExecutorService reconnectExecutor;
public KvClient(PDConfig pdConfig) {
+ this(pdConfig, Executors.newSingleThreadScheduledExecutor(runnable -> {
Review Comment:
⚠️ Important. This executor is created for every `KvClient`, but it is shut
down only by `KvClient.close()`. `SchemaDriver.destroy()` clears its caches and
singleton without closing the client that owns these reconnect tasks, so old
watches can continue after destroy and invoke callbacks against destroyed
state. Make the driver lifecycle close the client, or give the client a
destroy-safe ownership contract, and cover destroy with an active-watch test.
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -180,50 +200,170 @@ 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 {
+ return startWatch(subscription, true);
+ }
+
+ private boolean startWatch(WatchSubscription subscription,
+ boolean waitForPermit) 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;
+ }
+
+ if (waitForPermit) {
+ acquire(watchClientId, watchSemaphore);
+ } else if (!tryAcquire(watchClientId, watchSemaphore)) {
+ subscription.observer.compareAndSet(observer, null);
+ return false;
+ }
+ if (closed.get()) {
+ subscription.observer.compareAndSet(observer, null);
+ release(watchSemaphore);
+ return false;
+ }
+
+ WatchRequest request = WatchRequest.newBuilder()
Review Comment:
⚠️ Important. `WatchRequest` is built after the subscription and semaphore
are installed, but outside the `try` that releases them. A null key (or another
unchecked builder failure) therefore leaves the observer registered and the
permit held; a later valid `listen` can block indefinitely on the watch
semaphore. Build the request inside the guarded cleanup path and add an
invalid-input test that verifies no permit or subscription is leaked.
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -180,50 +200,170 @@ 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 {
+ return startWatch(subscription, true);
+ }
+
+ private boolean startWatch(WatchSubscription subscription,
+ boolean waitForPermit) 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;
+ }
+
+ if (waitForPermit) {
+ acquire(watchClientId, watchSemaphore);
+ } else if (!tryAcquire(watchClientId, watchSemaphore)) {
+ subscription.observer.compareAndSet(observer, null);
+ return false;
+ }
+ 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;
Review Comment:
⚠️ Important. `startWatch()` returns true immediately after
`asyncServerStreamingCall()` returns, before any `Starting` frame or other
readiness signal. If the connection is blackholed or the server never sends the
first frame, the subscription remains installed and the semaphore stays held
forever, so reconnect cannot make progress. Add a first-frame deadline/watchdog
that cleans up and reschedules the subscription when readiness is not observed.
--
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]
