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 653e16e9b Handle unknown ZooKeeper return codes and fix tests broken
by 3.9.3 (#3203, #3204, #3205, #3206)
653e16e9b is described below
commit 653e16e9bfac7cb5cb103f44cf1ca3941009e0ff
Author: Junkai Xue <[email protected]>
AuthorDate: Fri Nov 6 09:48:44 2026 -0800
Handle unknown ZooKeeper return codes and fix tests broken by 3.9.3 (#3203,
#3204, #3205, #3206)
The ZooKeeper 3.9.3 upgrade (#3093) made four CI tests fail.
testHelixDataAccessorReadData exposed a product-side gap. Helix interprets
async return codes with KeeperException.Code.get(rc) in ZkAsyncCallbacks,
ZkBaseDataAccessor, ZkCacheBaseDataAccessor and ZkClient. Before 3.9 that
lookup returned null for a code the client does not know and Helix treated
the null as a failed operation (ZkAsyncCallbacks even caught the resulting
NullPointerException on purpose to skip retrying). Since ZooKeeper 3.9
(ZOOKEEPER-4377) the lookup throws IllegalArgumentException, which turned
"unknown code means the operation failed" into an exception escaping from
the callback, and in ZkBaseDataAccessor.get() into a
HelixMetaDataAccessException even when the caller asked not to throw. All
those sites now go through ZkAsyncCallbacks.getCode(), which maps a code
this
client version does not know (possible with a newer server) to SYSTEMERROR
with a warning, so it is handled like any other failed, non-retryable
operation. The needRetry() implementations no longer need their
catch-everything blocks. MockZkClient, the only producer of such a code,
faked a failed read with return code 4, which was never a ZooKeeper code; it
now uses CONNECTIONLOSS.
The three TestZkMetaClient listener tests failed in meta-client's
TestUtil.getZkWatch, the same reflection helper that was already updated in
helix-core and zookeeper-api: ZooKeeper 3.7 moved the client watch registry
from the ZooKeeper#watchManager field behind the package-private
getWatchManager() accessor.
Tests run locally against 3.9.3: TestZkMetaClient; TestHelixDataAccessor,
TestZKWatch, TestZkBaseDataAccessor, TestZkCacheAsyncOpSingleThread,
TestWtCacheAsyncOpSingleThread, TestZkHelixAdmin and TestZkBasis in
helix-core; TestZkClientAsyncRetry and TestZooKeeperConnection in
zookeeper-api.
---
.../helix/manager/zk/ZkBaseDataAccessor.java | 26 ++++++------
.../helix/manager/zk/ZkCacheBaseDataAccessor.java | 2 +-
.../java/org/apache/helix/mock/MockZkClient.java | 5 ++-
.../apache/helix/metaclient/impl/zk/TestUtil.java | 28 +++++++------
.../apache/helix/zookeeper/zkclient/ZkClient.java | 10 ++---
.../zkclient/callback/ZkAsyncCallbacks.java | 47 +++++++++++++---------
6 files changed, 66 insertions(+), 52 deletions(-)
diff --git
a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
index 63a5d95d2..0816294b0 100644
---
a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
+++
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
@@ -639,17 +639,17 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
}
ZkAsyncCallbacks.GetDataCallbackHandler cb = cbList[i];
- if (Code.get(cb.getRc()) == Code.OK) {
+ if (ZkAsyncCallbacks.getCode(cb.getRc()) == Code.OK) {
@SuppressWarnings("unchecked")
T record = (T) _zkClient.deserialize(cb._data, paths.get(i));
records.set(i, record);
if (stats != null) {
stats.set(i, cb._stat);
}
- } else if (Code.get(cb.getRc()) != Code.NONODE && throwException) {
+ } else if (ZkAsyncCallbacks.getCode(cb.getRc()) != Code.NONODE &&
throwException) {
throw new HelixMetaDataAccessException(String
.format("Failed to read node %s, return code: %s", paths.get(i),
- Code.get(cb.getRc())));
+ ZkAsyncCallbacks.getCode(cb.getRc())));
} else {
pathFailToRead.put(paths.get(i), cb.getRc());
}
@@ -889,7 +889,7 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
cb.waitForSuccess();
String path = paths.get(i);
- if (Code.get(cb.getRc()) == Code.NONODE) {
+ if (ZkAsyncCallbacks.getCode(cb.getRc()) == Code.NONODE) {
String parentPath = HelixUtil.getZkParentPath(path);
parentPaths.set(i, parentPath);
failOnNoNode = true;
@@ -899,7 +899,7 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
needCreate[i] = false;
// if succeeds, record what paths we've created
- if (Code.get(cb.getRc()) == Code.OK && pathsCreated != null) {
+ if (ZkAsyncCallbacks.getCode(cb.getRc()) == Code.OK && pathsCreated
!= null) {
if (pathsCreated.get(i) == null) {
pathsCreated.set(i, new ArrayList<String>());
}
@@ -927,7 +927,7 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
continue;
}
- Code rc = Code.get(parentCb.getRc());
+ Code rc = ZkAsyncCallbacks.getCode(parentCb.getRc());
// if parent is created, retry create child
if (rc == Code.OK || rc == Code.NODEEXISTS) {
@@ -977,7 +977,7 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
for (int i = 0; i < cbList.length; i++) {
ZkAsyncCallbacks.CreateCallbackHandler cb = cbList[i];
- success[i] = (Code.get(cb.getRc()) == Code.OK);
+ success[i] = (ZkAsyncCallbacks.getCode(cb.getRc()) == Code.OK);
}
return success;
@@ -1052,7 +1052,7 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
for (int i = 0; i < cbList.length; i++) {
ZkAsyncCallbacks.SetDataCallbackHandler cb = cbList[i];
cb.waitForSuccess();
- Code rc = Code.get(cb.getRc());
+ Code rc = ZkAsyncCallbacks.getCode(cb.getRc());
switch (rc) {
case OK:
setStats.set(i, cb.getStat());
@@ -1079,7 +1079,7 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
continue;
}
- Code rc = Code.get(createCb.getRc());
+ Code rc = ZkAsyncCallbacks.getCode(createCb.getRc());
switch (rc) {
case OK:
setStats.set(i, ZNode.ZERO_STAT);
@@ -1102,12 +1102,12 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
for (int i = 0; i < cbList.length; i++) {
ZkAsyncCallbacks.SetDataCallbackHandler cb = cbList[i];
- Code rc = Code.get(cb.getRc());
+ Code rc = ZkAsyncCallbacks.getCode(cb.getRc());
if (rc == Code.OK) {
success[i] = true;
} else if (rc == Code.NONODE) {
ZkAsyncCallbacks.CreateCallbackHandler createCb = createCbList[i];
- if (Code.get(createCb.getRc()) == Code.OK) {
+ if (ZkAsyncCallbacks.getCode(createCb.getRc()) == Code.OK) {
success[i] = true;
}
}
@@ -1229,7 +1229,7 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
cb.waitForSuccess();
- switch (Code.get(cb.getRc())) {
+ switch (ZkAsyncCallbacks.getCode(cb.getRc())) {
case OK:
updateData.set(i, newDataList.get(i));
setStats.set(i, cb.getStat());
@@ -1259,7 +1259,7 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
continue;
}
- switch (Code.get(createCb.getRc())) {
+ switch (ZkAsyncCallbacks.getCode(createCb.getRc())) {
case OK:
needUpdate[i] = false;
updateData.set(i, newDataList.get(i));
diff --git
a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
index 9691c365d..66b6c84a7 100644
---
a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
+++
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
@@ -483,7 +483,7 @@ public class ZkCacheBaseDataAccessor<T> implements
HelixPropertyStore<T> {
boolean[] success = new boolean[size];
for (int i = 0; i < size; i++) {
ZkAsyncCallbacks.CreateCallbackHandler cb = createCbList[i];
- success[i] = (Code.get(cb.getRc()) == Code.OK);
+ success[i] = (ZkAsyncCallbacks.getCode(cb.getRc()) == Code.OK);
updateCache(cache, pathsCreatedList.get(i), success[i],
serverPaths.get(i),
records.get(i), ZNode.ZERO_STAT);
diff --git a/helix-core/src/test/java/org/apache/helix/mock/MockZkClient.java
b/helix-core/src/test/java/org/apache/helix/mock/MockZkClient.java
index 6b834d287..53118db3f 100644
--- a/helix-core/src/test/java/org/apache/helix/mock/MockZkClient.java
+++ b/helix-core/src/test/java/org/apache/helix/mock/MockZkClient.java
@@ -27,6 +27,7 @@ import org.apache.helix.manager.zk.ZNRecordSerializer;
import org.apache.helix.manager.zk.ZkAsyncCallbacks;
import org.apache.helix.zookeeper.impl.client.ZkClient;
import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.zookeeper.KeeperException;
public class MockZkClient extends ZkClient implements HelixZkClient {
Map<String, byte[]> _dataMap;
@@ -42,7 +43,9 @@ public class MockZkClient extends ZkClient implements
HelixZkClient {
final ZkAsyncCallbacks.GetDataCallbackHandler cb) {
if (_dataMap.containsKey(path)) {
if (_dataMap.get(path) == null) {
- cb.processResult(4, path, null, _dataMap.get(path), null);
+ // Simulate a failed read with a real ZooKeeper error code. ZooKeeper
3.9+ rejects
+ // unknown codes in KeeperException.Code.get() instead of returning
null.
+ cb.processResult(KeeperException.Code.CONNECTIONLOSS.intValue(), path,
null, null, null);
} else {
cb.processResult(0, path, null, _dataMap.get(path), null);
}
diff --git
a/meta-client/src/test/java/org/apache/helix/metaclient/impl/zk/TestUtil.java
b/meta-client/src/test/java/org/apache/helix/metaclient/impl/zk/TestUtil.java
index a0aca15e8..4f9d0c1f1 100644
---
a/meta-client/src/test/java/org/apache/helix/metaclient/impl/zk/TestUtil.java
+++
b/meta-client/src/test/java/org/apache/helix/metaclient/impl/zk/TestUtil.java
@@ -65,34 +65,36 @@ public class TestUtil {
ZkConnection connection = ((ZkConnection) zkClient.getConnection());
ZooKeeper zk = connection.getZookeeper();
- java.lang.reflect.Field field = getField(zk.getClass(), "watchManager");
- field.setAccessible(true);
- Object watchManager = field.get(zk);
+ // Since ZooKeeper 3.7 the client watch manager lives in ClientCnxn and is
reached through the
+ // package-private ZooKeeper#getWatchManager() instead of a field.
+ java.lang.reflect.Method getWatchManager =
ZooKeeper.class.getDeclaredMethod("getWatchManager");
+ getWatchManager.setAccessible(true);
+ Object watchManager = getWatchManager.invoke(zk);
java.lang.reflect.Field field2 = getField(watchManager.getClass(),
"dataWatches");
field2.setAccessible(true);
- HashMap<String, Set<Watcher>> dataWatches =
- (HashMap<String, Set<Watcher>>) field2.get(watchManager);
+ Map<String, Set<Watcher>> dataWatches =
+ (Map<String, Set<Watcher>>) field2.get(watchManager);
field2 = getField(watchManager.getClass(), "existWatches");
field2.setAccessible(true);
- HashMap<String, Set<Watcher>> existWatches =
- (HashMap<String, Set<Watcher>>) field2.get(watchManager);
+ Map<String, Set<Watcher>> existWatches =
+ (Map<String, Set<Watcher>>) field2.get(watchManager);
field2 = getField(watchManager.getClass(), "childWatches");
field2.setAccessible(true);
- HashMap<String, Set<Watcher>> childWatches =
- (HashMap<String, Set<Watcher>>) field2.get(watchManager);
+ Map<String, Set<Watcher>> childWatches =
+ (Map<String, Set<Watcher>>) field2.get(watchManager);
field2 = getField(watchManager.getClass(), "persistentWatches");
field2.setAccessible(true);
- HashMap<String, Set<Watcher>> persistentWatches =
- (HashMap<String, Set<Watcher>>) field2.get(watchManager);
+ Map<String, Set<Watcher>> persistentWatches =
+ (Map<String, Set<Watcher>>) field2.get(watchManager);
field2 = getField(watchManager.getClass(), "persistentRecursiveWatches");
field2.setAccessible(true);
- HashMap<String, Set<Watcher>> persistentRecursiveWatches =
- (HashMap<String, Set<Watcher>>) field2.get(watchManager);
+ Map<String, Set<Watcher>> persistentRecursiveWatches =
+ (Map<String, Set<Watcher>>) field2.get(watchManager);
lists.put("dataWatches", new ArrayList<>(dataWatches.keySet()));
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 95ca167be..50703a7e0 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
@@ -1730,7 +1730,7 @@ public class ZkClient implements Watcher {
callbackHandler.waitForSuccess();
- KeeperException.Code code =
KeeperException.Code.get(callbackHandler.getRc());
+ KeeperException.Code code =
ZkAsyncCallbacks.getCode(callbackHandler.getRc());
if (code == KeeperException.Code.OK) {
LOG.info("zkclient {}, sycnOnNewSession with sessionID {} async return
code: {} and proceeds",
_uid, sessionId, code);
@@ -1913,7 +1913,7 @@ public class ZkClient implements Watcher {
for (int i = 0; i < opResults.size(); i++) {
if (opResults.get(i) instanceof OpResult.ErrorResult) {
failedPathsMap.put(ops.get(i).getPath(),
- KeeperException.Code.get(((OpResult.ErrorResult)
opResults.get(i)).getErr()));
+ ZkAsyncCallbacks.getCode(((OpResult.ErrorResult)
opResults.get(i)).getErr()));
}
}
@@ -2221,7 +2221,7 @@ public class ZkClient implements Watcher {
/** The exception to throw for a non-OK result code. Mirrors ZooKeeper's
synchronous API. */
KeeperException failure(String path) {
- return KeeperException.create(KeeperException.Code.get(_rc), path);
+ return KeeperException.create(ZkAsyncCallbacks.getCode(_rc), path);
}
}
@@ -2286,7 +2286,7 @@ public class ZkClient implements Watcher {
throw e;
}
LOG.info("Caller interrupted while a write to {} was in flight; waited
for its outcome: {}",
- path, KeeperException.Code.get(write._rc));
+ path, ZkAsyncCallbacks.getCode(write._rc));
}
if (write._rc != KeeperException.Code.OK.intValue()) {
throw write.failure(path);
@@ -3147,7 +3147,7 @@ public class ZkClient implements Watcher {
InFlightWrite<List<OpResult>> write = new
InFlightWrite<List<OpResult>>() {
@Override
KeeperException failure(String path) {
- return KeeperException.create(KeeperException.Code.get(_rc));
+ return KeeperException.create(ZkAsyncCallbacks.getCode(_rc));
}
};
return syncWrite(null, currentZookeeper(), write, (zk, w) ->
zk.multi(ops,
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/callback/ZkAsyncCallbacks.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/callback/ZkAsyncCallbacks.java
index 61c97c231..2b2cb64dd 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/callback/ZkAsyncCallbacks.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/callback/ZkAsyncCallbacks.java
@@ -37,6 +37,26 @@ import org.slf4j.LoggerFactory;
public class ZkAsyncCallbacks {
private static Logger LOG = LoggerFactory.getLogger(ZkAsyncCallbacks.class);
+ /**
+ * Translates an async return code into a {@link Code}. A code this client
version does not know
+ * (possible with a newer server; KeeperException.Code#get returned null for
it before ZooKeeper
+ * 3.9 and throws IllegalArgumentException since) is reported as {@link
Code#SYSTEMERROR}, so
+ * callers treat it as a failed, non-retryable operation instead of failing
on the lookup itself.
+ */
+ public static Code getCode(int rc) {
+ Code code = null;
+ try {
+ code = Code.get(rc);
+ } catch (IllegalArgumentException e) {
+ // handled below together with the pre-3.9 null result
+ }
+ if (code == null) {
+ LOG.warn("Unknown ZooKeeper return code {}, treating it as {}", rc,
Code.SYSTEMERROR);
+ return Code.SYSTEMERROR;
+ }
+ return code;
+ }
+
public static class GetDataCallbackHandler extends DefaultCallback
implements DataCallback {
public byte[] _data;
public Stat _stat;
@@ -173,18 +193,12 @@ public class ZkAsyncCallbacks {
@Override
protected boolean needRetry(int rc) {
- try {
- switch (KeeperException.Code.get(rc)) {
- /** Connection to the server has been lost */
- case CONNECTIONLOSS:
- return true;
- default:
- return false;
- }
- } catch (ClassCastException | NullPointerException ex) {
- LOG.error("Session {} failed to handle unknown return code {}. Skip
retrying. ex {}",
- _sessionId, rc, ex);
- return false;
+ switch (getCode(rc)) {
+ /** Connection to the server has been lost */
+ case CONNECTIONLOSS:
+ return true;
+ default:
+ return false;
}
}
}
@@ -198,7 +212,7 @@ public class ZkAsyncCallbacks {
public void callback(int rc, String path, Object ctx) {
if (rc != 0) {
- LOG.debug(" {}, rc:{}, path {}", this, Code.get(rc), path);
+ LOG.debug(" {}, rc:{} ({}), path {}", this, rc, getCode(rc), path);
}
if (ctx != null && ctx instanceof ZkAsyncCallMonitorContext) {
@@ -304,8 +318,7 @@ public class ZkAsyncCallbacks {
* @return true if the error is transient and the operation may succeed
when being retried.
*/
protected boolean needRetry(int rc) {
- try {
- switch (Code.get(rc)) {
+ switch (getCode(rc)) {
/** Connection to the server has been lost */
case CONNECTIONLOSS:
/** The session has been expired by the server */
@@ -315,10 +328,6 @@ public class ZkAsyncCallbacks {
return true;
default:
return false;
- }
- } catch (ClassCastException | NullPointerException ex) {
- LOG.error("Failed to handle unknown return code {}. Skip retrying.",
rc, ex);
- return false;
}
}
}
