imbajin commented on code in PR #3157:
URL: https://github.com/apache/hugegraph/pull/3157#discussion_r3790078606
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -180,49 +186,122 @@ public void onNext(WatchResponse value) {
@Override
public void onError(Throwable t) {
- release();
- if (!closed.get()) {
- clientId.set(0);
- listenWrapper.accept(key, consumer);
- }
+ requestReconnect(subscription, this);
}
@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);
+ 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();
+ if (closed.get()) {
+ subscription.observer.compareAndSet(observer, null);
+ release();
+ return false;
+ }
+
+ WatchRequest request = WatchRequest.newBuilder()
+ .setClientId(clientId.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();
+ if (e instanceof PDException) {
+ throw (PDException) e;
+ }
throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, e);
}
}
+ private void requestReconnect(WatchSubscription subscription,
+ StreamObserver<WatchResponse>
sourceObserver) {
+ if (closed.get() ||
+ !subscription.observer.compareAndSet(sourceObserver, null)) {
+ return;
+ }
+ clientId.set(0L);
Review Comment:
‼️ Critical. A single watch failure resets the shared `clientId` to zero,
but the same field is used by `lock`, `unlock`, and `keepAlive`
(KvClient.java:357-416). The server stores locks under that ID and `unlock`
only deletes a lock when `owned(key, clientId)` matches
(KvService.java:230-273); after this watch reconnect obtains a new ID, an
existing lock remains under the old ID and cannot be unlocked or renewed,
expiring only at its TTL. Separate watch identity from lock identity, or
coordinate reconnects so the lock session ID is never reset.
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -180,49 +186,122 @@ public void onNext(WatchResponse value) {
@Override
public void onError(Throwable t) {
- release();
- if (!closed.get()) {
- clientId.set(0);
- listenWrapper.accept(key, consumer);
- }
+ requestReconnect(subscription, this);
}
@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);
+ 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();
+ if (closed.get()) {
+ subscription.observer.compareAndSet(observer, null);
+ release();
+ return false;
+ }
+
+ WatchRequest request = WatchRequest.newBuilder()
+ .setClientId(clientId.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:
‼️ Critical. `startWatch()` treats a normal return from `streamingCall()` as
a successful watch, but `AbstractClient.streamingCall()` catches the terminal
`StatusRuntimeException` after its retry limit and returns without invoking the
observer (AbstractClient.java:252-266). The observer remains installed and the
semaphore can remain held, so `reconnect()` never schedules another attempt;
the new `TestKvClient.streamingCall()` test double also bypasses this
production path by throwing `PDException` directly. Propagate the final
transport failure or return an explicit failure, clean the observer/semaphore
state, and add a test that exercises the production streaming path.
##########
hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java:
##########
@@ -180,49 +186,122 @@ public void onNext(WatchResponse value) {
@Override
public void onError(Throwable t) {
Review Comment:
⚠️ Important. `onError(Throwable t)` discards the status and unconditionally
schedules another watch; `onCompleted()` follows the same path. A terminal
authentication/validation failure therefore becomes an unbounded one-second RPC
retry with no way for the caller to observe or stop the subscription. Classify
retryable transport/leader-transition errors separately, and terminate/report
non-retryable failures instead of rescheduling them forever.
##########
hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/client/KvClientTest.java:
##########
@@ -67,6 +97,189 @@ public void testCreateBlockingStub() {
}
}
+ @Test
+ public void testTransportInitializationDoesNotCloseClient() throws
Exception {
+ AtomicReference<String> grpcAddress = new AtomicReference<>();
+ Server server = ServerBuilder.forPort(0)
+ .addService(new PDGrpc.PDImplBase() {
+ @Override
+ public void getMembers(
+ Pdpb.GetMembersRequest
request,
+
StreamObserver<Pdpb.GetMembersResponse> observer) {
+ Metapb.Member leader =
+ Metapb.Member.newBuilder()
+
.setGrpcUrl(grpcAddress.get())
+ .build();
+ observer.onNext(
+
Pdpb.GetMembersResponse.newBuilder()
+
.setLeader(leader)
+
.build());
+ observer.onCompleted();
+ }
+ })
+ .build()
+ .start();
+ grpcAddress.set("127.0.0.1:" + server.getPort());
+ InitializableKvClient testClient =
+ new InitializableKvClient(PDConfig.of(grpcAddress.get())
+ .setAuthority(user, pwd));
+
+ try {
+ testClient.initializeTransport();
+
+ assertThat(isClosed(testClient)).isFalse();
+ testClient.close();
+ assertThat(isClosed(testClient)).isTrue();
+ } finally {
+ testClient.close();
+ server.shutdownNow().awaitTermination(5L, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testReconnectRetriesAfterFirstFailure() throws Exception {
+ try (WatchTestContext context = newWatchTestContext()) {
+ Consumer<WatchResponse> consumer = mock(Consumer.class);
+ context.client.listen("key", consumer);
+ context.client.failNextCalls(1);
+
+ context.client.call(0).observer.onError(new
RuntimeException("disconnected"));
+ context.runNextReconnect();
+ assertThat(context.client.calls).hasSize(2);
+ assertThat(context.reconnectTasks).hasSize(1);
+
+ context.runNextReconnect();
+
+ assertThat(context.client.calls).hasSize(3);
+ assertThat(context.reconnectTasks).isEmpty();
+ assertThat(context.client.call(2).methodName)
+
.isEqualTo(KvServiceGrpc.getWatchMethod().getFullMethodName());
+
assertThat(context.client.call(2).request.getKey()).isEqualTo("key");
+
+ WatchResponse started = WatchResponse.newBuilder()
+
.setState(WatchState.Starting)
+ .setClientId(1L)
+ .build();
+ WatchEvent event =
WatchEvent.newBuilder().setType(WatchType.Put).build();
+ WatchResponse futureEvent = WatchResponse.newBuilder()
+
.setState(WatchState.Started)
+ .setClientId(1L)
+ .addEvents(event)
+ .build();
+ context.client.call(2).observer.onNext(started);
+ context.client.call(2).observer.onNext(futureEvent);
+ verify(consumer).accept(futureEvent);
+ }
+ }
+
+ @Test
+ public void testReconnectRetriesAfterConsecutiveFailures() throws
Exception {
+ try (WatchTestContext context = newWatchTestContext()) {
+ context.client.listen("key", response -> { });
+ context.client.failNextCalls(2);
+
+ context.client.call(0).observer.onError(new
RuntimeException("disconnected"));
+ context.runNextReconnect();
+ context.runNextReconnect();
+ context.runNextReconnect();
+
+ assertThat(context.client.calls).hasSize(4);
+ assertThat(context.reconnectTasks).isEmpty();
+ }
+ }
+
+ @Test
+ public void testLeaderChangedSchedulesReconnect() throws Exception {
+ try (WatchTestContext context = newWatchTestContext()) {
+ context.client.listen("key", response -> { });
+ StreamObserver<WatchResponse> observer =
context.client.call(0).observer;
+ observer.onNext(WatchResponse.newBuilder()
+ .setState(WatchState.Starting)
+ .setClientId(7L)
+ .build());
+
+ observer.onNext(
+
WatchResponse.newBuilder().setState(WatchState.Leader_Changed).build());
+
+ assertThat(context.reconnectTasks).hasSize(1);
+ assertThat(context.reconnectDelaysMs.get(0)).isGreaterThan(0L);
+ context.runNextReconnect();
+ assertThat(context.client.calls).hasSize(2);
+ assertThat(context.client.call(1).request.getClientId()).isZero();
+ }
+ }
+
+ @Test
+ public void testCompletedSchedulesReconnect() throws Exception {
+ try (WatchTestContext context = newWatchTestContext()) {
+ context.client.listen("key", response -> { });
+
+ context.client.call(0).observer.onCompleted();
+
+ assertThat(context.reconnectTasks).hasSize(1);
+ context.runNextReconnect();
+ assertThat(context.client.calls).hasSize(2);
+ }
+ }
+
+ @Test
+ public void testObserverSchedulesOnlyOneReconnect() throws Exception {
+ try (WatchTestContext context = newWatchTestContext()) {
+ context.client.listen("key", response -> { });
+ StreamObserver<WatchResponse> observer =
context.client.call(0).observer;
+
+ observer.onError(new RuntimeException("disconnected"));
+ observer.onCompleted();
+
+ assertThat(context.reconnectTasks).hasSize(1);
+ }
+ }
+
+ @Test
+ public void testStaleObserverDoesNotScheduleReconnect() throws Exception {
+ try (WatchTestContext context = newWatchTestContext()) {
+ context.client.listen("key", response -> { });
+ StreamObserver<WatchResponse> staleObserver =
context.client.call(0).observer;
+ staleObserver.onError(new RuntimeException("disconnected"));
+ context.runNextReconnect();
+
+ staleObserver.onCompleted();
Review Comment:
⚠️ Important. `testStaleObserverDoesNotScheduleReconnect()` only calls
`onCompleted()` on the stale observer; it never sends a stale `Started`
response containing an event after the new observer is installed. A regression
removing the identity check in `KvClient.java:162-164` would still pass this
test and could duplicate business callbacks. Send an event through the stale
observer and assert the consumer is untouched, while verifying the current
observer still delivers events.
--
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]
