This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
The following commit(s) were added to refs/heads/master by this push:
new 5612d2426 Fix 4 reopened flaky CI tests; fix a real disconnect() hang
on broken ZK connection
5612d2426 is described below
commit 5612d24266667c84509a17a80b6f274cf99518b0
Author: Junkai Xue <[email protected]>
AuthorDate: Mon Nov 2 14:33:16 2026 -0800
Fix 4 reopened flaky CI tests; fix a real disconnect() hang on broken ZK
connection
Addresses the reopened [Failed CI Test] issues #3149, #3144, #3158, #3175.
- #3149 TestZkConnectionLost.testDisconnectWhenConnectionBreak: disconnect()
could hang when the ZK connection broke during the callback-handler
cleanup.
ZkClient.waitForRetry() swallowed an interrupt that arrived during the
retry
backoff sleep and cleared the interrupt flag, so the cleanup thread
(which the
manager stops via that single interrupt) kept retrying and then blocked in
waitUntilConnected() for the whole operation retry timeout. waitForRetry()
now throws ZkInterruptedException like the other interrupt points (note:
any
caller interrupted during the backoff now sees the operation aborted,
which
is what an interrupt at the other points already did). ZKHelixManager
re-interrupts the cleanup thread every second while the connection is down
instead of joining it unbounded, and tolerates the disconnecting thread
itself being interrupted. Added a unit test that times out without the
ZkClient fix. The test also records the stuck thread stack on failure.
- #3144 TestIndependentTaskRebalancer.testThresholdFailure: the
invoked-class
set and run-count map were plain HashSet/HashMap written concurrently by
task
threads of several participants; two concurrent first insertions can lose
an
element. Use concurrent collections with atomic updates and poll instead
of
a fixed sleep.
- #3158 TestZkCacheAsyncOpSingleThread: the test compared the live
ZkCallbackCache content without the cache read lock while the zk event
thread mutated it under the write lock (ConcurrentModificationException /
torn reads). Compare under the read lock like production readers do;
TestHelper.verifyZkCache additionally treats a concurrently removed entry
or
a concurrent modification as "not stable yet" so polling callers retry.
- #3175 TestZeroReplicaAvoidance.testWagedRebalancer: root cause not
reproducible; failure records now carry the validated snapshot's version
and modified time, the validated map, and (clearly labeled as a separate,
possibly newer read) the current ideal state preference list/mapping.
---
.../apache/helix/manager/zk/ZKHelixManager.java | 22 ++++++++-
.../src/test/java/org/apache/helix/TestHelper.java | 22 +++++++++
.../helix/integration/TestZkConnectionLost.java | 6 ++-
.../rebalancer/TestZeroReplicaAvoidance.java | 54 ++++++++++++++++++----
.../task/TestIndependentTaskRebalancer.java | 36 +++++++++------
.../manager/zk/TestZkCacheAsyncOpSingleThread.java | 43 +++++++++++++----
.../apache/helix/zookeeper/zkclient/ZkClient.java | 7 ++-
.../zookeeper/impl/client/TestRawZkClient.java | 54 ++++++++++++++++++++++
8 files changed, 207 insertions(+), 37 deletions(-)
diff --git
a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
index 3473bed08..cf549de2d 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
@@ -116,6 +116,8 @@ public class ZKHelixManager implements HelixManager,
IZkStateListener {
private static final int FLAPPING_TIME_WINDOW = 300000; // Default to 300 sec
public static final int DEFAULT_MAX_DISCONNECT_THRESHOLD = 600; // Default
to be a large number
private static final int DEFAULT_WAIT_CONNECTED_TIMEOUT = 10 * 1000; //
wait until connected for up to 10 seconds.
+ // How often disconnect() re-checks the connection while waiting for the
callback handler cleanup.
+ private static final long CLEANUP_RECHECK_INTERVAL_MS = 1000L;
protected final String _zkAddress;
private final String _clusterName;
@@ -975,8 +977,24 @@ public class ZKHelixManager implements HelixManager,
IZkStateListener {
}
try {
- cleanupThread.join();
- } catch (InterruptedException ex) {
+ // Wait for the cleanup to finish. The state listener above
interrupts the cleanup thread
+ // once the connection breaks, but a single interrupt can be
consumed by the code the
+ // thread is running at that moment (for example a sleep in a retry
loop) and never reach
+ // the blocking ZK call. So while the cleanup thread is alive and
the connection is not
+ // active, keep interrupting it. This keeps the original behavior
for a healthy
+ // connection (no time bound on the cleanup) while guaranteeing that
disconnect() cannot
+ // hang forever on a cleanup that can no longer make progress.
+ while (cleanupThread.isAlive()) {
+ cleanupThread.join(CLEANUP_RECHECK_INTERVAL_MS);
+ if (cleanupThread.isAlive() && !_zkclient.waitUntilConnected(0,
TimeUnit.MILLISECONDS)) {
+ LOG.warn("ZkClient is not connected. Interrupt the cleanup
thread {} of {} again.",
+ cleanupThread.getName(), _instanceName);
+ cleanupThread.interrupt();
+ }
+ }
+ } catch (InterruptedException | ZkInterruptedException ex) {
+ // The disconnecting thread itself got interrupted
(waitUntilConnected() reports this as
+ // ZkInterruptedException). Stop waiting and make sure the cleanup
thread stops too.
cleanupThread.interrupt();
}
} finally {
diff --git a/helix-core/src/test/java/org/apache/helix/TestHelper.java
b/helix-core/src/test/java/org/apache/helix/TestHelper.java
index 206617861..138af90e0 100644
--- a/helix-core/src/test/java/org/apache/helix/TestHelper.java
+++ b/helix-core/src/test/java/org/apache/helix/TestHelper.java
@@ -26,6 +26,7 @@ import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -625,6 +626,20 @@ public class TestHelper {
public static boolean verifyZkCache(List<String> paths, List<String>
pathsExcludeForStat,
Map<String, ZNode> cache, Map<String, ZNode> zkMap, boolean
needVerifyStat) {
+ try {
+ return doVerifyZkCache(pathsExcludeForStat, cache, zkMap,
needVerifyStat);
+ } catch (ConcurrentModificationException e) {
+ // The cache passed in is usually the live ZkCallbackCache map. Its
ZNode child sets are plain
+ // HashSets that the zk event thread keeps mutating while this method
iterates over them, so a
+ // comparison can race with an in-flight update. The cache is simply not
stable yet in that
+ // case, so report a mismatch and let the polling caller retry instead
of failing the test.
+ System.err.println("zkCache was modified concurrently during
verification, will retry: " + e);
+ return false;
+ }
+ }
+
+ private static boolean doVerifyZkCache(List<String> pathsExcludeForStat,
+ Map<String, ZNode> cache, Map<String, ZNode> zkMap, boolean
needVerifyStat) {
// equal size
if (zkMap.size() != cache.size()) {
System.err
@@ -643,6 +658,13 @@ public class TestHelper {
ZNode cacheNode = cache.get(path);
ZNode zkNode = zkMap.get(path);
+ if (cacheNode == null) {
+ // The live cache map is weakly consistent: the entry was removed
between the keySet
+ // iteration and the get, i.e. the cache is still changing. Report a
mismatch for retry.
+ System.err.println("path: " + path + " was removed from the cache
during verification");
+ return false;
+ }
+
if (zkNode == null) {
// in cache but not on zk
System.err.println("path: " + path + " in cache but not on zk:
inCacheNode: " + cacheNode);
diff --git
a/helix-core/src/test/java/org/apache/helix/integration/TestZkConnectionLost.java
b/helix-core/src/test/java/org/apache/helix/integration/TestZkConnectionLost.java
index ee019152a..997d420e2 100644
---
a/helix-core/src/test/java/org/apache/helix/integration/TestZkConnectionLost.java
+++
b/helix-core/src/test/java/org/apache/helix/integration/TestZkConnectionLost.java
@@ -20,6 +20,7 @@ package org.apache.helix.integration;
*/
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
@@ -136,8 +137,11 @@ public class TestZkConnectionLost extends TaskTestBase {
// disconnect() with a broken ZK connection legitimately takes a while
(internal
// retries / timeouts), especially on slow or overloaded machines.
testThread.join(TestHelper.WAIT_DURATION);
+ // Capture where disconnect() is stuck before failing, otherwise a hang
leaves no trace.
+ String stuckAt = disconnected.get() ? "" : ", disconnect thread stack: "
+ + Arrays.toString(testThread.getStackTrace());
Assert.assertTrue(disconnected.get(),
- "disconnect() should complete within " + TestHelper.WAIT_DURATION +
" ms");
+ "disconnect() should complete within " + TestHelper.WAIT_DURATION +
" ms" + stuckAt);
Assert.assertFalse(controllerManager.isConnected());
} finally {
if (testThread.isAlive()) {
diff --git
a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestZeroReplicaAvoidance.java
b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestZeroReplicaAvoidance.java
index 44a3f23e5..2314970de 100644
---
a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestZeroReplicaAvoidance.java
+++
b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestZeroReplicaAvoidance.java
@@ -27,6 +27,7 @@ import java.util.Map;
import org.apache.helix.ExternalViewChangeListener;
import org.apache.helix.HelixManager;
+import org.apache.helix.HelixProperty;
import org.apache.helix.HelixManagerFactory;
import org.apache.helix.IdealStateChangeListener;
import org.apache.helix.InstanceType;
@@ -209,7 +210,7 @@ public class TestZeroReplicaAvoidance extends ZkTestBase
for (String partition : is.getPartitionSet()) {
Map<String, String> evStateMap = ev.getRecord().getMapField(partition);
- validateMap(is.getResourceName(), partition, replica, evStateMap,
stateModelDef);
+ validateMap("ExternalView", ev, is, partition, replica, evStateMap,
stateModelDef);
}
}
@@ -222,22 +223,32 @@ public class TestZeroReplicaAvoidance extends ZkTestBase
BuiltInStateModelDefinitions.valueOf(is.getStateModelDefRef()).getStateModelDefinition();
for (String partition : is.getPartitionSet()) {
- validateMap(is.getResourceName(), partition, replica,
is.getInstanceStateMap(partition),
+ validateMap("IdealState", is, is, partition, replica,
is.getInstanceStateMap(partition),
stateModelDef);
}
}
- private void validateMap(String resource, String partition, int replica,
- Map<String, String> instanceStateMap, StateModelDefinition
stateModelDef) {
+ /**
+ * @param source which snapshot is being validated (external view
or ideal state)
+ * @param snapshot the property delivered with the callback that is
being validated
+ * @param is the ideal state used for the static
configurations. For the external
+ * view path it is fetched separately and may be
newer than the snapshot.
+ * @param instanceStateMap the instance -> state map of the partition to
validate
+ */
+ private void validateMap(String source, HelixProperty snapshot, IdealState
is, String partition,
+ int replica, Map<String, String> instanceStateMap, StateModelDefinition
stateModelDef) {
+ String resource = is.getResourceName();
if (instanceStateMap == null || instanceStateMap.isEmpty()) {
recordValidationFailure(
- String.format("Resource %s partition %s has no active replica!",
resource, partition));
+ String.format("Resource %s partition %s has no active replica!%s",
resource, partition,
+ describeContext(source, snapshot, is, partition,
instanceStateMap)));
return;
}
if (instanceStateMap.size() < replica) {
recordValidationFailure(
- String.format("Resource %s partition %s has %d active replica, less
than required %d!",
- resource, partition, instanceStateMap.size(), replica));
+ String.format("Resource %s partition %s has %d active replica, less
than required %d!%s",
+ resource, partition, instanceStateMap.size(), replica,
+ describeContext(source, snapshot, is, partition,
instanceStateMap)));
return;
}
@@ -252,12 +263,37 @@ public class TestZeroReplicaAvoidance extends ZkTestBase
}
if (topCount > 1) {
recordValidationFailure(
- String.format("Resource %s partition %s has %d replica in %s, more
than 1!", resource,
- partition, topCount, topState));
+ String.format("Resource %s partition %s has %d replica in %s, more
than 1!%s", resource,
+ partition, topCount, topState,
+ describeContext(source, snapshot, is, partition,
instanceStateMap)));
}
}
}
+ /**
+ * Describe the snapshot that failed the validation. The failure is observed
on a callback thread
+ * and only surfaces later through the recorded message, so the message has
to carry the version
+ * of the snapshot that was validated. For the external view path the ideal
state is a separate,
+ * possibly newer read; it is labeled as such and reported with its own
version so that the two
+ * can be correlated with the controller logs instead of being mistaken for
one consistent view.
+ */
+ private String describeContext(String source, HelixProperty snapshot,
IdealState is,
+ String partition, Map<String, String> validatedMap) {
+ StringBuilder sb = new StringBuilder();
+ sb.append(String.format(" [source=%s, time=%d, snapshotVersion=%d,
snapshotModifiedTime=%d, "
+ + "validatedMap=%s, snapshotPreferenceList=%s", source,
System.currentTimeMillis(),
+ snapshot.getRecord().getVersion(),
snapshot.getRecord().getModifiedTime(), validatedMap,
+ snapshot.getRecord().getListField(partition)));
+ if (snapshot != is) {
+ sb.append(String.format(", currentIsVersion=%d,
currentIsModifiedTime=%d, "
+ + "currentIsPreferenceList=%s, currentIsMap=%s (ideal state read
separately, "
+ + "may be newer than the snapshot)", is.getRecord().getVersion(),
+ is.getRecord().getModifiedTime(), is.getPreferenceList(partition),
+ is.getInstanceStateMap(partition)));
+ }
+ return sb.append(']').toString();
+ }
+
private void recordValidationFailure(String message) {
// Record the failure instead of calling Assert.fail(). The validation
runs on the Helix
// callback threads, where a thrown assertion error is swallowed by the
callback handler and
diff --git
a/helix-core/src/test/java/org/apache/helix/integration/task/TestIndependentTaskRebalancer.java
b/helix-core/src/test/java/org/apache/helix/integration/task/TestIndependentTaskRebalancer.java
index 0cd3a1009..055627de9 100644
---
a/helix-core/src/test/java/org/apache/helix/integration/task/TestIndependentTaskRebalancer.java
+++
b/helix-core/src/test/java/org/apache/helix/integration/task/TestIndependentTaskRebalancer.java
@@ -24,6 +24,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.collect.ImmutableMap;
@@ -55,12 +56,15 @@ import org.testng.Assert;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import org.testng.collections.Sets;
public class TestIndependentTaskRebalancer extends TaskTestBase {
private static final Logger LOG =
LoggerFactory.getLogger(TestIndependentTaskRebalancer.class);
- private Set<String> _invokedClasses = Sets.newHashSet();
- private Map<String, Integer> _runCounts = Maps.newHashMap();
+ // Both collections are written by the task threads of several participants
concurrently and read
+ // by the test thread, so they must be thread-safe. A plain HashSet can
silently lose one of two
+ // concurrent first insertions (both threads allocate the backing table and
one allocation wins),
+ // which made the invoked class check flaky.
+ private final Set<String> _invokedClasses = ConcurrentHashMap.newKeySet();
+ private final Map<String, Integer> _runCounts = new ConcurrentHashMap<>();
private static final AtomicBoolean _failureCtl = new AtomicBoolean(true);
@BeforeClass
@@ -195,19 +199,20 @@ public class TestIndependentTaskRebalancer extends
TaskTestBase {
LOG.info("Job context after completion: " + (jobCtx != null ? "attempts="
+ jobCtx.getPartitionNumAttempts(0) : "null"));
- // Additional wait to ensure all task run() methods have completed
- // This handles the race between workflow completion and task execution
finishing
- Thread.sleep(500);
+ // Both tasks record themselves at the very beginning of run(), so they
must be visible once the
+ // workflow is COMPLETED. Poll briefly anyway to absorb any scheduling
slack on a slow machine.
+ boolean bothInvoked = TestHelper.verify(
+ () -> _invokedClasses.contains(TaskOne.class.getName())
+ && _invokedClasses.contains(TaskTwo.class.getName()),
+ TestHelper.WAIT_DURATION);
// Log the invoked classes for debugging
LOG.info("Invoked classes: " + _invokedClasses);
LOG.info("Run counts: " + _runCounts);
// Ensure that each class was invoked
- Assert.assertTrue(_invokedClasses.contains(TaskOne.class.getName()),
- "TaskOne was not invoked. Invoked classes: " + _invokedClasses);
- Assert.assertTrue(_invokedClasses.contains(TaskTwo.class.getName()),
- "TaskTwo was not invoked. Invoked classes: " + _invokedClasses);
+ Assert.assertTrue(bothInvoked,
+ "Both TaskOne and TaskTwo should have been invoked. Invoked classes: "
+ _invokedClasses);
}
@Test(dependsOnMethods = "testThresholdFailure", timeOut = 600000)
@@ -335,10 +340,9 @@ public class TestIndependentTaskRebalancer extends
TaskTestBase {
}
_shouldFail = shouldFail;
- // Initialize the count for this instance if not already done
- if (!_runCounts.containsKey(instanceName)) {
- _runCounts.put(instanceName, 0);
- }
+ // Initialize the count for this instance if not already done (atomic,
tasks on the same
+ // participant may be constructed concurrently)
+ _runCounts.putIfAbsent(instanceName, 0);
_instanceName = instanceName;
}
@@ -346,7 +350,9 @@ public class TestIndependentTaskRebalancer extends
TaskTestBase {
public synchronized TaskResult run() {
LOG.info("TaskOne.run() invoked on instance: " + _instanceName + ",
shouldFail: " + _shouldFail);
_invokedClasses.add(getClass().getName());
- _runCounts.put(_instanceName, _runCounts.get(_instanceName) + 1);
+ // run() is synchronized per task object only; different tasks on the
same participant run
+ // concurrently, so the increment must be atomic.
+ _runCounts.merge(_instanceName, 1, Integer::sum);
LOG.info("Current invoked classes: " + _invokedClasses + ", run counts:
" + _runCounts);
// Fail the task if it should fail
diff --git
a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkCacheAsyncOpSingleThread.java
b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkCacheAsyncOpSingleThread.java
index e5fcc92c2..ca6cd9e8f 100644
---
a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkCacheAsyncOpSingleThread.java
+++
b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkCacheAsyncOpSingleThread.java
@@ -22,10 +22,13 @@ package org.apache.helix.manager.zk;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import org.apache.helix.AccessOption;
import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.store.zk.ZNode;
import org.apache.helix.TestHelper;
import org.apache.helix.ZkTestHelper;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
@@ -38,6 +41,28 @@ import org.testng.Assert;
import org.testng.annotations.Test;
public class TestZkCacheAsyncOpSingleThread extends ZkUnitTestBase {
+
+ /**
+ * Compare the content of the ZkCallbackCache with ZK while holding the
cache read lock, like
+ * every production reader of the cache does. The zk event thread mutates
the cached ZNodes
+ * (plain HashSet child sets) under the write lock, so comparing without the
lock races with
+ * in-flight updates (ConcurrentModificationException or torn reads). ZK is
read before taking
+ * the lock to keep the lock hold time short; the polling callers retry on a
mismatch.
+ */
+ private boolean verifyZkCacheWithReadLock(List<String> paths,
+ ZkCacheBaseDataAccessor<ZNRecord> accessor, boolean needVerifyStat) {
+ Map<String, ZNode> zkMap = new HashMap<>();
+ for (String path : paths) {
+ TestHelper.readZkRecursive(path, zkMap, _gZkClient);
+ }
+ accessor._zkCache.lockRead();
+ try {
+ return TestHelper.verifyZkCache(paths, null, accessor._zkCache._cache,
zkMap, needVerifyStat);
+ } finally {
+ accessor._zkCache.unlockRead();
+ }
+ }
+
@Test
public void testSessionExpirationWithSharedZkClient() throws Exception {
int curstateCnt = 10;
@@ -63,7 +88,7 @@ public class TestZkCacheAsyncOpSingleThread extends
ZkUnitTestBase {
new ZkCacheBaseDataAccessor<>(sharedBaseAccessor, null, null,
zkCacheInitPaths);
boolean ret =
- TestHelper.verifyZkCache(zkCacheInitPaths, accessor._zkCache._cache,
_gZkClient, true);
+ verifyZkCacheWithReadLock(zkCacheInitPaths, accessor, true);
Assert.assertTrue(ret, "zkCache doesn't match data on Zk");
// create 10 current states using external base accessor
@@ -152,7 +177,7 @@ public class TestZkCacheAsyncOpSingleThread extends
ZkUnitTestBase {
// TestHelper.printCache(accessor._zkCache);
boolean ret = TestHelper.verify(() ->
- TestHelper.verifyZkCache(zkCacheInitPaths, accessor._zkCache._cache,
_gZkClient, true),
+ verifyZkCacheWithReadLock(zkCacheInitPaths, accessor, true),
30 * 1000);
Assert.assertTrue(ret, "zkCache doesn't match data on Zk");
@@ -176,7 +201,7 @@ public class TestZkCacheAsyncOpSingleThread extends
ZkUnitTestBase {
// wait zkEventThread update zkCache
// verify wtCache
ret = TestHelper.verify(() -> {
- return TestHelper.verifyZkCache(zkCacheInitPaths,
accessor._zkCache._cache, _gZkClient, true);
+ return verifyZkCacheWithReadLock(zkCacheInitPaths, accessor, true);
}, 30 * 1000);
Assert.assertTrue(ret, "zkCache doesn't match data on Zk");
@@ -203,7 +228,7 @@ public class TestZkCacheAsyncOpSingleThread extends
ZkUnitTestBase {
// wait zkEventThread update zkCache using polling
ret = TestHelper.verify(() ->
- TestHelper.verifyZkCache(zkCacheInitPaths, accessor._zkCache._cache,
_gZkClient, true),
+ verifyZkCacheWithReadLock(zkCacheInitPaths, accessor, true),
30 * 1000);
Assert.assertTrue(ret, "zkCache doesn't match data on Zk");
@@ -224,7 +249,7 @@ public class TestZkCacheAsyncOpSingleThread extends
ZkUnitTestBase {
// wait zkEventThread update zkCache using polling
ret = TestHelper.verify(() ->
- TestHelper.verifyZkCache(zkCacheInitPaths, accessor._zkCache._cache,
_gZkClient, true),
+ verifyZkCacheWithReadLock(zkCacheInitPaths, accessor, true),
30 * 1000);
Assert.assertTrue(ret, "zkCache doesn't match data on Zk");
@@ -242,7 +267,7 @@ public class TestZkCacheAsyncOpSingleThread extends
ZkUnitTestBase {
// wait zkEventThread update zkCache using polling
ret = TestHelper.verify(() ->
- TestHelper.verifyZkCache(zkCacheInitPaths, accessor._zkCache._cache,
_gZkClient, true),
+ verifyZkCacheWithReadLock(zkCacheInitPaths, accessor, true),
30 * 1000);
Assert.assertTrue(ret, "zkCache doesn't match data on Zk");
@@ -273,7 +298,7 @@ public class TestZkCacheAsyncOpSingleThread extends
ZkUnitTestBase {
// TestHelper.printCache(accessor._zkCache._cache);
boolean ret =
- TestHelper.verifyZkCache(zkCacheInitPaths, accessor._zkCache._cache,
_gZkClient, true);
+ verifyZkCacheWithReadLock(zkCacheInitPaths, accessor, true);
Assert.assertTrue(ret, "zkCache doesn't match data on Zk");
// create 10 current states using this accessor
@@ -295,7 +320,7 @@ public class TestZkCacheAsyncOpSingleThread extends
ZkUnitTestBase {
// verify cache
// TestHelper.printCache(accessor._zkCache._cache);
- ret = TestHelper.verifyZkCache(zkCacheInitPaths, accessor._zkCache._cache,
_gZkClient, false);
+ ret = verifyZkCacheWithReadLock(zkCacheInitPaths, accessor, false);
Assert.assertTrue(ret, "zkCache doesn't match data on Zk");
// update each current state 10 times by this accessor
@@ -346,7 +371,7 @@ public class TestZkCacheAsyncOpSingleThread extends
ZkUnitTestBase {
// verify cache
// TestHelper.printCache(accessor._zkCache._cache);
- ret = TestHelper.verifyZkCache(zkCacheInitPaths, accessor._zkCache._cache,
_gZkClient, true);
+ ret = verifyZkCacheWithReadLock(zkCacheInitPaths, accessor, true);
// System.out.println("ret: " + ret);
Assert.assertTrue(ret, "zkCache doesn't match data on Zk");
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java
index 6a0fc9b5d..3d46983ac 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java
@@ -2221,7 +2221,12 @@ public class ZkClient implements Watcher {
LOG.debug("zkclient {} Wait for {} ms before retrying operation",
_uid, maxSleep);
Thread.sleep(maxSleep);
} catch (InterruptedException ex) {
- // we don't need to re-throw.
+ // An interrupt means the caller wants to abort the operation, exactly
like an interrupt
+ // received inside waitUntilConnected() or inside the ZK call itself.
Swallowing it here
+ // would clear the interrupt flag and let the retry loop continue; if
the connection has
+ // been lost in the meantime, the next waitUntilConnected() would then
block for the whole
+ // operation retry timeout with no way to stop it (see
HelixManager#disconnect cleanup).
+ throw new ZkInterruptedException(ex);
}
}
}
diff --git
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestRawZkClient.java
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestRawZkClient.java
index 0ce0349ab..ed25303c0 100644
---
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestRawZkClient.java
+++
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestRawZkClient.java
@@ -30,6 +30,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -52,6 +53,7 @@ import org.apache.helix.zookeeper.zkclient.ZkConnection;
import org.apache.helix.zookeeper.zkclient.ZkServer;
import org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks;
import org.apache.helix.zookeeper.zkclient.exception.ZkException;
+import org.apache.helix.zookeeper.zkclient.exception.ZkInterruptedException;
import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
import
org.apache.helix.zookeeper.zkclient.exception.ZkSessionMismatchedException;
import org.apache.helix.zookeeper.zkclient.exception.ZkTimeoutException;
@@ -784,6 +786,58 @@ public class TestRawZkClient extends ZkTestBase {
Assert.assertFalse(_zkClient.exists(path));
}
+ /*
+ * Tests that an interrupt arriving while retryUntilConnected() sleeps
between two retries aborts
+ * the operation with ZkInterruptedException and preserves the interrupt
flag, exactly like an
+ * interrupt that arrives inside the ZK call or inside waitUntilConnected().
Before this was fixed
+ * the interrupt was swallowed, the retry loop kept going and, once the
connection was actually
+ * lost, the operation blocked for the whole operation retry timeout with no
way to stop it. That
+ * made HelixManager#disconnect() hang on a broken ZK connection.
+ */
+ @Test(timeOut = 60 * 1000L)
+ public void testInterruptDuringRetryBackoff() throws Exception {
+ final ZkClient zkClient = new ZkClient(ZkTestBase.ZK_ADDR);
+ final Thread operationThread = Thread.currentThread();
+ final AtomicInteger attempts = new AtomicInteger();
+ // Interrupt the operation thread shortly after the first attempt failed,
i.e. while it is in
+ // the backoff sleep (a few hundred milliseconds for the first retries).
+ final Thread interrupter = new Thread(() -> {
+ try {
+ Thread.sleep(100L);
+ } catch (InterruptedException e) {
+ return;
+ }
+ operationThread.interrupt();
+ });
+ try {
+ long start = System.currentTimeMillis();
+ interrupter.start();
+ try {
+ zkClient.retryUntilConnected(() -> {
+ attempts.incrementAndGet();
+ // Simulate a connection loss while the client still reports
SyncConnected, which is the
+ // window in which retryUntilConnected() sleeps before retrying.
+ throw new KeeperException.ConnectionLossException();
+ });
+ Assert.fail("The interrupted operation should not be retried until it
succeeds.");
+ } catch (ZkInterruptedException expected) {
+ // The interrupt flag must be preserved for the caller. Clear it for
the rest of the test.
+ Assert.assertTrue(Thread.interrupted(), "The interrupt flag should be
preserved.");
+ }
+ long elapsed = System.currentTimeMillis() - start;
+ Assert.assertTrue(elapsed < 10_000L,
+ "The operation should be aborted promptly after the interrupt but
took " + elapsed + "ms");
+ Assert.assertTrue(attempts.get() <= 3,
+ "The operation should not keep retrying after the interrupt,
attempts: " + attempts.get());
+ } finally {
+ // Clear the flag first: a pending interrupt would otherwise turn the
join below into an
+ // InterruptedException that masks the real failure, and would disturb
closing the client.
+ Thread.interrupted();
+ interrupter.join();
+ zkClient.close();
+ }
+ }
+
/*
* Tests that when trying to create an ephemeral node, if connection to zk
service is lost,
* ConnectionLossException will be thrown in retryUntilConnected and
ephemeral creation will fail.
