imbajin commented on code in PR #3157:
URL: https://github.com/apache/hugegraph/pull/3157#discussion_r3890227305
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -127,7 +156,7 @@ public TTLResponse putTTL(String key, String value, long
ttl) throws PDException
private void onEvent(WatchResponse value, Consumer<T> consumer) {
log.debug("receive message for {},event Count:{}", value,
value.getEventsCount());
- clientId.compareAndSet(0L, value.getClientId());
+ watchClientId.compareAndSet(0L, value.getClientId());
Review Comment:
‼️ Critical. The global watch ID can be restored by another live
subscription while a failed subscription is waiting to reconnect. For example,
A clears the ID after disconnecting, B receives an event and writes the old ID
back here, then A reconnects with that old ID. The server registers observers
with `putIfAbsent(key@clientId, observer)`, so A's stale server observer can
win and the new stream appears started but never receives later events. Please
use a per-subscription/generation ID or atomically replace the stale server
observer, and test the one-failed/one-still-active sequence.
##########
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,
Review Comment:
⚠️ Important. `CANCELLED` is treated as permanently non-retryable, but
`AbstractClient.resetStub()` closes the shared channel with `shutdownNow()`.
That internal refresh can deliver `CANCELLED` to unrelated live watches, which
then enter `stopWatch()` and are removed forever even though `KvClient` is
still open. Please distinguish explicit client close from internal channel
replacement, or otherwise make these cancellations recoverable, and add a
multi-subscription channel-reset 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()
+ .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);
Review Comment:
‼️ Critical. `requestReconnect()` resets the observer and watch ID but
leaves the cached async stub intact. After `Leader_Changed` or an asynchronous
`UNAVAILABLE`, `reconnect()` calls `getStub()` and reuses the same channel; the
server-side `onError` callback never enters `streamingCall()`'s synchronous
peer retry. A leader move can therefore retry the old follower forever. Please
invalidate/rotate the async stub before scheduling the reconnect and cover this
with a real two-peer leader-failover test.
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -237,7 +377,7 @@ private void acquire() {
}
}
- private void release() {
+ private void release(Semaphore semaphore) {
Review Comment:
⚠️ Important. Permit ownership is still racy: this check-then-release is not
atomic, and `tryAcquire()` can return true for a nonzero client ID without
acquiring a permit while close/failure paths still call `release()`. Concurrent
`Starting`, error, or close callbacks can therefore raise the permit count
above one, allowing multiple reconnects to issue `clientId=0` streams and
record inconsistent server IDs. Please model acquisition ownership explicitly
(or use an atomic state machine) and assert that the effective permit count
never exceeds one under concurrent callbacks.
--
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]
