You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hugegraph.apache.org by "dpol1 (via GitHub)" <gi...@apache.org> on 2026/06/22 17:01:01 UTC
[PR] fix(core): self-heal etcd meta watch on transport reconnect [hugegraph]
dpol1 opened a new pull request, #3062:
URL: https://github.com/apache/hugegraph/pull/3062
## Purpose of the PR
- close #3036
`CachedSchemaTransactionV2` registers a JVM-global watch on the meta store so that a
schema change on one node clears the schema cache on the other nodes. That watch goes
through `EtcdMetaDriver.listen` / `listenPrefix`, which handed jetcd the bare
`Consumer<WatchResponse>` overload. That overload drops `onError` and `onCompleted`, so
when jetcd ends a watch (for example after a transport reconnect tears down the gRPC
stream) the listener stopped receiving events with no log line and no exception. The node
kept serving stale schema and nothing reported it. `PdMetaDriver` does not have this
problem because its `KvClient` already re-subscribes on error.
## Main Changes
- `EtcdMetaDriver.listen` / `listenPrefix` now register a `Watch.Listener` instead of the
bare `Consumer` overload. On `onError` or `onCompleted` the driver re-subscribes after a
1s backoff on a daemon thread, which is the same recovery `PdMetaDriver` gets through
`KvClient`. A `WARN` is logged before each re-subscribe, so the failure is now visible.
- Removed `CachedSchemaTransactionV2.resetMetaListenerForReconnect()`. It was a manual
stopgap added in #3011 with no callers, and it never shipped in a tagged release. Now
that the driver recovers on its own, the JVM-global register-once flag is correct to stay
set, so the manual reset has nothing left to do. The lifecycle comment on
`metaEventListenerRegistered` is updated to describe the self-heal behavior.
- The `MetaDriver` interface is unchanged, so neither implementor needs edits beyond
`EtcdMetaDriver`.
Known limitation: the re-subscribe opens a fresh watch without a stored revision, so any
cache-clear events emitted during the short reconnect window are not replayed. This is the
same behavior as `PdMetaDriver` / `KvClient`. The change removes the permanent silent
failure. It does not add gap replay.
## Verifying these changes
- [x] Need tests and can be verified as follows:
- New `EtcdMetaDriverTest` (JUnit + Mockito) mocks the jetcd `Client`/`Watch`, captures
the registered `Watch.Listener`, and asserts that both `onError` and `onCompleted`
trigger a re-subscribe (a second `watch(...)` call), and that `onNext` still reaches
the consumer.
- Local run: `EtcdMetaDriverTest` 4/4, `CachedSchemaTransactionTest` 18/18, and
`MetaManagerSchemaCacheClearEventTest` 6/6 after the stopgap removal.
- Red-green: commenting out the re-subscribe makes the three recovery tests fail;
restoring it makes them pass.
## Does this PR potentially affect the following parts?
- [ ] Dependencies
- [ ] Modify configurations
- [ ] The public API
- [ ] Other affects (typed here)
- [x] Nope
## Documentation Status
- [ ] `Doc - TODO`
- [ ] `Doc - Done`
- [x] `Doc - No Need`
--
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
Re: [PR] fix(core): self-heal etcd meta watch on transport reconnect [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin merged PR #3062:
URL: https://github.com/apache/hugegraph/pull/3062
--
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
Re: [PR] fix(core): self-heal etcd meta watch on transport reconnect [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #3062:
URL: https://github.com/apache/hugegraph/pull/3062#discussion_r3457108136
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/EtcdMetaDriver.java:
##########
@@ -314,9 +341,35 @@ public <T> void listen(String key, Consumer<T> consumer) {
@SuppressWarnings("unchecked")
@Override
public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
- ByteSequence sequence = toByteSequence(prefix);
WatchOption option = WatchOption.newBuilder().isPrefix(true).build();
- this.client.getWatchClient().watch(sequence, option, (Consumer<WatchResponse>) consumer);
+ this.watchKey(toByteSequence(prefix), option,
+ (Consumer<WatchResponse>) consumer);
+ }
+
+ /**
+ * Subscribe a self-healing watch. Unlike the bare {@code Consumer} overload,
+ * this surfaces {@code onError}/{@code onCompleted}: when the underlying
+ * watch terminates (e.g. a transport reconnect drops the gRPC stream) it
+ * re-subscribes after a short backoff, so the listener is not silently lost.
+ * Mirrors the re-subscribe behaviour PdMetaDriver already gets from KvClient.
+ */
+ private void watchKey(ByteSequence key, WatchOption option,
+ Consumer<WatchResponse> consumer) {
+ Watch.Listener listener = Watch.listener(
+ consumer,
+ throwable -> this.scheduleReWatch(key, option, consumer, throwable),
Review Comment:
‼️ Critical: this creates duplicate active watches on retryable jetcd reconnects.
HugeGraph uses `jetcd-core:0.5.9`, and `WatchImpl.handleError()` already calls the application listener's `onError(...)` and then internally `reschedule()`s the same `WatcherImpl` for retryable gRPC errors. With this PR, `onError` also schedules a fresh `watchKey(...)`, while the original watcher is still resumable and not closed/tracked here.
That means one transient transport reconnect can leave two active watchers for the same key/prefix, and repeated reconnects can multiply meta-event delivery and retained watch resources. This affects all `EtcdMetaDriver.listen/listenPrefix` consumers, not only schema cache clear.
Please avoid opening a fresh watch from retryable `onError`. Either rely on jetcd's built-in reschedule path, or track the returned `Watch.Watcher` and replace it only after a confirmed terminal close/cancel path where jetcd will not resume the original watcher. The test should model jetcd's real `onError -> reschedule same watcher` behavior, not just a mocked listener callback.
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/EtcdMetaDriver.java:
##########
@@ -314,9 +341,35 @@ public <T> void listen(String key, Consumer<T> consumer) {
@SuppressWarnings("unchecked")
@Override
public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
- ByteSequence sequence = toByteSequence(prefix);
WatchOption option = WatchOption.newBuilder().isPrefix(true).build();
- this.client.getWatchClient().watch(sequence, option, (Consumer<WatchResponse>) consumer);
+ this.watchKey(toByteSequence(prefix), option,
+ (Consumer<WatchResponse>) consumer);
+ }
+
+ /**
+ * Subscribe a self-healing watch. Unlike the bare {@code Consumer} overload,
+ * this surfaces {@code onError}/{@code onCompleted}: when the underlying
+ * watch terminates (e.g. a transport reconnect drops the gRPC stream) it
+ * re-subscribes after a short backoff, so the listener is not silently lost.
+ * Mirrors the re-subscribe behaviour PdMetaDriver already gets from KvClient.
+ */
+ private void watchKey(ByteSequence key, WatchOption option,
+ Consumer<WatchResponse> consumer) {
+ Watch.Listener listener = Watch.listener(
+ consumer,
+ throwable -> this.scheduleReWatch(key, option, consumer, throwable),
+ () -> this.scheduleReWatch(key, option, consumer, null));
+ // Watcher intentionally not closed: process-lifetime watch, recreated
+ // on self-heal; the prior watcher's stream has already terminated.
+ this.client.getWatchClient().watch(key, option, listener);
+ }
+ private void scheduleReWatch(ByteSequence key, WatchOption option,
+ Consumer<WatchResponse> consumer,
+ Throwable cause) {
+ LOG.warn("etcd meta watch dropped, re-subscribing", cause);
+ this.reWatchExecutor.schedule(
Review Comment:
‼️ Critical: a failed delayed re-watch attempt stops recovery permanently.
The scheduled task calls `watchKey(...)` directly. If `watch(...)` throws once while the watch client is still closed/unavailable, the task fails and no further retry is scheduled. Since this PR also removes the upper-layer reset hook and leaves `metaEventListenerRegistered` permanently true, the schema-cache-clear listener can again be lost forever after a reconnect failure.
Please wrap the scheduled body, log failed retry attempts, and keep retrying with bounded/backoff behavior while the driver is active. Add a regression test where the first replacement `watch(...)` call throws and a later attempt succeeds.
##########
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/EtcdMetaDriverTest.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hugegraph.meta;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hugegraph.testutil.Assert;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.Watch;
+import io.etcd.jetcd.options.WatchOption;
+import io.etcd.jetcd.watch.WatchResponse;
+
+/**
+ * Unit tests for {@link EtcdMetaDriver}'s self-healing watch: a dropped watch
+ * (onError / onCompleted) must be re-subscribed so the listener is not silently
+ * lost after a transport reconnect (issue #3036). Uses a mock jetcd
+ * {@link Client} via the package-private test constructor; no live etcd needed.
+ */
+public class EtcdMetaDriverTest {
+
+ @Test
+ public void testListenReSubscribesOnError() {
+ Watch watch = Mockito.mock(Watch.class);
+ EtcdMetaDriver driver = newDriver(watch);
+
+ driver.listen("k", response -> { });
+ captureListener(watch).onError(new RuntimeException("watch dropped"));
+
+ // The watch terminated, so the driver must re-subscribe: a second
+ // watch() call lands once the (0ms) backoff task runs.
+ Mockito.verify(watch, Mockito.timeout(2000).times(2))
+ .watch(Mockito.any(ByteSequence.class),
+ Mockito.any(WatchOption.class),
+ Mockito.any(Watch.Listener.class));
+ }
+
+ @Test
+ public void testListenReSubscribesOnCompleted() {
+ Watch watch = Mockito.mock(Watch.class);
+ EtcdMetaDriver driver = newDriver(watch);
+
+ driver.listen("k", response -> { });
+ captureListener(watch).onCompleted();
+
+ Mockito.verify(watch, Mockito.timeout(2000).times(2))
+ .watch(Mockito.any(ByteSequence.class),
+ Mockito.any(WatchOption.class),
+ Mockito.any(Watch.Listener.class));
+ }
+
+ @Test
+ public void testListenPrefixReSubscribesOnError() {
+ Watch watch = Mockito.mock(Watch.class);
Review Comment:
⚠️ Important: this test only proves a second `watch(...)` call happens; it does not prove the re-created prefix watch preserves the original key and prefix `WatchOption`.
Please capture the second invocation's `ByteSequence` and `WatchOption` and assert it still watches `"prefix"` with prefix semantics enabled. Otherwise a future regression could downgrade this to an exact-key watch while the test still passes.
--
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
Re: [PR] fix(core): self-heal etcd meta watch on transport reconnect [hugegraph]
Posted by "dpol1 (via GitHub)" <gi...@apache.org>.
dpol1 commented on code in PR #3062:
URL: https://github.com/apache/hugegraph/pull/3062#discussion_r3457937601
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/EtcdMetaDriver.java:
##########
@@ -314,9 +341,35 @@ public <T> void listen(String key, Consumer<T> consumer) {
@SuppressWarnings("unchecked")
@Override
public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
- ByteSequence sequence = toByteSequence(prefix);
WatchOption option = WatchOption.newBuilder().isPrefix(true).build();
- this.client.getWatchClient().watch(sequence, option, (Consumer<WatchResponse>) consumer);
+ this.watchKey(toByteSequence(prefix), option,
+ (Consumer<WatchResponse>) consumer);
+ }
+
+ /**
+ * Subscribe a self-healing watch. Unlike the bare {@code Consumer} overload,
+ * this surfaces {@code onError}/{@code onCompleted}: when the underlying
+ * watch terminates (e.g. a transport reconnect drops the gRPC stream) it
+ * re-subscribes after a short backoff, so the listener is not silently lost.
+ * Mirrors the re-subscribe behaviour PdMetaDriver already gets from KvClient.
+ */
+ private void watchKey(ByteSequence key, WatchOption option,
+ Consumer<WatchResponse> consumer) {
+ Watch.Listener listener = Watch.listener(
+ consumer,
+ throwable -> this.scheduleReWatch(key, option, consumer, throwable),
Review Comment:
Address by d2bd2a2
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/EtcdMetaDriver.java:
##########
@@ -314,9 +341,35 @@ public <T> void listen(String key, Consumer<T> consumer) {
@SuppressWarnings("unchecked")
@Override
public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
- ByteSequence sequence = toByteSequence(prefix);
WatchOption option = WatchOption.newBuilder().isPrefix(true).build();
- this.client.getWatchClient().watch(sequence, option, (Consumer<WatchResponse>) consumer);
+ this.watchKey(toByteSequence(prefix), option,
+ (Consumer<WatchResponse>) consumer);
+ }
+
+ /**
+ * Subscribe a self-healing watch. Unlike the bare {@code Consumer} overload,
+ * this surfaces {@code onError}/{@code onCompleted}: when the underlying
+ * watch terminates (e.g. a transport reconnect drops the gRPC stream) it
+ * re-subscribes after a short backoff, so the listener is not silently lost.
+ * Mirrors the re-subscribe behaviour PdMetaDriver already gets from KvClient.
+ */
+ private void watchKey(ByteSequence key, WatchOption option,
+ Consumer<WatchResponse> consumer) {
+ Watch.Listener listener = Watch.listener(
+ consumer,
+ throwable -> this.scheduleReWatch(key, option, consumer, throwable),
+ () -> this.scheduleReWatch(key, option, consumer, null));
+ // Watcher intentionally not closed: process-lifetime watch, recreated
+ // on self-heal; the prior watcher's stream has already terminated.
+ this.client.getWatchClient().watch(key, option, listener);
+ }
+ private void scheduleReWatch(ByteSequence key, WatchOption option,
+ Consumer<WatchResponse> consumer,
+ Throwable cause) {
+ LOG.warn("etcd meta watch dropped, re-subscribing", cause);
+ this.reWatchExecutor.schedule(
Review Comment:
Address by d2bd2a2
##########
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/EtcdMetaDriverTest.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hugegraph.meta;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hugegraph.testutil.Assert;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.Watch;
+import io.etcd.jetcd.options.WatchOption;
+import io.etcd.jetcd.watch.WatchResponse;
+
+/**
+ * Unit tests for {@link EtcdMetaDriver}'s self-healing watch: a dropped watch
+ * (onError / onCompleted) must be re-subscribed so the listener is not silently
+ * lost after a transport reconnect (issue #3036). Uses a mock jetcd
+ * {@link Client} via the package-private test constructor; no live etcd needed.
+ */
+public class EtcdMetaDriverTest {
+
+ @Test
+ public void testListenReSubscribesOnError() {
+ Watch watch = Mockito.mock(Watch.class);
+ EtcdMetaDriver driver = newDriver(watch);
+
+ driver.listen("k", response -> { });
+ captureListener(watch).onError(new RuntimeException("watch dropped"));
+
+ // The watch terminated, so the driver must re-subscribe: a second
+ // watch() call lands once the (0ms) backoff task runs.
+ Mockito.verify(watch, Mockito.timeout(2000).times(2))
+ .watch(Mockito.any(ByteSequence.class),
+ Mockito.any(WatchOption.class),
+ Mockito.any(Watch.Listener.class));
+ }
+
+ @Test
+ public void testListenReSubscribesOnCompleted() {
+ Watch watch = Mockito.mock(Watch.class);
+ EtcdMetaDriver driver = newDriver(watch);
+
+ driver.listen("k", response -> { });
+ captureListener(watch).onCompleted();
+
+ Mockito.verify(watch, Mockito.timeout(2000).times(2))
+ .watch(Mockito.any(ByteSequence.class),
+ Mockito.any(WatchOption.class),
+ Mockito.any(Watch.Listener.class));
+ }
+
+ @Test
+ public void testListenPrefixReSubscribesOnError() {
+ Watch watch = Mockito.mock(Watch.class);
Review Comment:
Address by d2bd2a2
--
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