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 ab04ae11f Make sync ZK writes report their real outcome when the
caller is interrupted (#3072, #2685)
ab04ae11f is described below
commit ab04ae11fe35619cc5a33021ae020392e082471b
Author: Junkai Xue <[email protected]>
AuthorDate: Thu Nov 5 14:49:38 2026 -0800
Make sync ZK writes report their real outcome when the caller is
interrupted (#3072, #2685)
ZkBaseDataAccessor.set() could return false while the write had in fact been
applied. ZooKeeper's synchronous API queues the request to its send thread
before blocking the caller for the reply, so a thread interrupted during
that
wait gets InterruptedException even though the server still applies the
request. ZkClient turned that into ZkInterruptedException and
ZkBaseDataAccessor swallowed it into RetCode.ERROR, telling callers the
write
definitely did not happen when the outcome was actually unknown.
Two fixes, one per layer:
- ZkClient: the synchronous create/setData/delete/multi paths now submit
through the ZooKeeper async API and wait on a latch. If the calling
thread is
interrupted while the request is in flight, ZkClient keeps waiting for the
outcome ZooKeeper guarantees to deliver for every queued packet, either
the
server reply or CONNECTIONLOSS/SESSIONEXPIRED when the connection drops.
The
wait is bounded by the new zk.interrupted.write.wait.ms property (default:
the session timeout; 0 restores the previous abort-immediately behavior)
and
a second interrupt aborts it, so ZKHelixManager.disconnect(), which
re-interrupts its cleanup thread every second while the connection is
down,
is not held for the session timeout. The interrupt flag is restored before
returning so the caller still observes it. Only if the outcome is still
unknown when the wait ends is ZkInterruptedException thrown.
deleteRecursively/deleteRecursivelyAtomic let that exception through
instead
of wrapping it into ZkClientException.
- ZkBaseDataAccessor (and ZkCacheBaseDataAccessor.set on top of it):
create/set/update/multiSet/remove/removeWithExpectedVersion propagate
ZkInterruptedException instead of returning false, the same way
ZkBadVersionException is already propagated. A false return value now
always
means the write was not applied; an exception means it was aborted with an
unknown outcome and the caller should re-read. BaseDataAccessor javadoc
updated accordingly.
Tests: TestZkClientInFlightWrite unit-tests the wait itself (outcome
preserved under a pre-set interrupt, definite failures still reported, wait
disabled, bound elapsed, second interrupt aborts, property parsing) without
a
ZK server so the timing cases are deterministic. TestRawZkClient covers the
end-to-end create/setData/delete outcome under a pre-set interrupt against a
real server. TestZkBaseDataAccessor covers propagation from every sync
operation including the create-on-missing-node fallbacks, remove()'s
recursive-delete fallback, and the cache accessor.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
---
.../java/org/apache/helix/BaseDataAccessor.java | 28 +++-
.../helix/manager/zk/ZkBaseDataAccessor.java | 23 ++-
.../helix/manager/zk/ZkCacheBaseDataAccessor.java | 4 +
.../helix/manager/zk/TestZkBaseDataAccessor.java | 72 ++++++++
.../zookeeper/constant/ZkSystemPropertyKeys.java | 18 ++
.../apache/helix/zookeeper/zkclient/ZkClient.java | 179 +++++++++++++++++---
.../zookeeper/impl/client/TestRawZkClient.java | 45 +++++
.../zkclient/TestZkClientInFlightWrite.java | 184 +++++++++++++++++++++
8 files changed, 526 insertions(+), 27 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/BaseDataAccessor.java
b/helix-core/src/main/java/org/apache/helix/BaseDataAccessor.java
index 63245b544..fc1f31aed 100644
--- a/helix-core/src/main/java/org/apache/helix/BaseDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/BaseDataAccessor.java
@@ -41,7 +41,9 @@ public interface BaseDataAccessor<T> {
* @param path path to the ZNode to create
* @param record the data to write to the ZNode
* @param options Set the type of ZNode see the valid values in {@link
AccessOption}
- * @return true if creation succeeded, false otherwise (e.g. if the ZNode
exists)
+ * @return true if creation succeeded, false otherwise (e.g. if the ZNode
exists).
+ * If the calling thread is interrupted while the operation is in
flight, the
+ * implementation throws instead of returning false, because the
outcome is unknown.
*/
boolean create(String path, T record, int options);
@@ -53,7 +55,9 @@ public interface BaseDataAccessor<T> {
* @param record the data to write to the ZNode
* @param options Set the type of ZNode see the valid values in {@link
AccessOption}
* @param ttl TTL of the node in milliseconds, if options supports it
- * @return true if creation succeeded, false otherwise (e.g. if the ZNode
exists)
+ * @return true if creation succeeded, false otherwise (e.g. if the ZNode
exists).
+ * If the calling thread is interrupted while the operation is in
flight, the
+ * implementation throws instead of returning false, because the
outcome is unknown.
*/
default boolean create(String path, T record, int options, long ttl) {
throw new UnsupportedOperationException("create with TTL support is not
implemented.");
@@ -65,7 +69,9 @@ public interface BaseDataAccessor<T> {
* @param path path to the ZNode to set
* @param record the data to write to the ZNode
* @param options Set the type of ZNode see the valid values in {@link
AccessOption}
- * @return true if data was successfully set, false otherwise
+ * @return true if data was successfully set, false otherwise.
+ * If the calling thread is interrupted while the operation is in
flight, the
+ * implementation throws instead of returning false, because the
outcome is unknown.
*/
boolean set(String path, T record, int options);
@@ -78,7 +84,9 @@ public interface BaseDataAccessor<T> {
* @param options Set the type of ZNode see the valid values in {@link
AccessOption}
* @param expectVersion the expected version of the data to be overwritten,
-1 means match any
* version
- * @return true if data was successfully set, false otherwise (e.g. if the
version mismatches)
+ * @return true if data was successfully set, false otherwise (e.g. if the
version mismatches).
+ * If the calling thread is interrupted while the operation is in
flight, the
+ * implementation throws instead of returning false, because the
outcome is unknown.
*/
boolean set(String path, T record, int expectVersion, int options);
@@ -89,7 +97,9 @@ public interface BaseDataAccessor<T> {
* @param path path to the ZNode to update
* @param updater an update routine for the data to merge in
* @param options Set the type of ZNode see the valid values in {@link
AccessOption}
- * @return true if data update succeeded, false otherwise
+ * @return true if data update succeeded, false otherwise.
+ * If the calling thread is interrupted while the operation is in
flight, the
+ * implementation throws instead of returning false, because the
outcome is unknown.
*/
boolean update(String path, DataUpdater<T> updater, int options);
@@ -108,7 +118,9 @@ public interface BaseDataAccessor<T> {
* This will remove the ZNode and all its descendants if any
* @param path path to the root ZNode to remove
* @param options Set the type of ZNode see the valid values in {@link
AccessOption}
- * @return true if the removal succeeded, false otherwise
+ * @return true if the removal succeeded, false otherwise.
+ * If the calling thread is interrupted while the operation is in
flight, the
+ * implementation throws instead of returning false, because the
outcome is unknown.
*/
boolean remove(String path, int options);
@@ -119,7 +131,9 @@ public interface BaseDataAccessor<T> {
* @param options Set the type of ZNode see the valid values in {@link
AccessOption}
* @param expectedVersion the expected version of the node to be removed, -1
means match any
* version
- * @return true if the removal succeeded, false otherwise
+ * @return true if the removal succeeded, false otherwise.
+ * If the calling thread is interrupted while the operation is in
flight, the
+ * implementation throws instead of returning false, because the
outcome is unknown.
*/
default boolean removeWithExpectedVersion(String path, int options, int
expectedVersion) {
throw new NotImplementedException("removeWithExpectedVersion is not
implemented");
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 06ea97fb9..63a5d95d2 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
@@ -51,6 +51,7 @@ import org.apache.helix.zookeeper.zkclient.ZkClient;
import org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks;
import org.apache.helix.zookeeper.zkclient.exception.ZkBadVersionException;
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.ZkNodeExistsException;
import org.apache.helix.zookeeper.zkclient.serialize.PathBasedZkSerializer;
@@ -311,6 +312,8 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
// if parent node created/exists, retry
retry = true;
}
+ } catch (ZkInterruptedException e1) {
+ throw e1;
} catch (Exception e1) {
LOG.error("Exception while creating path: " + parentPath, e1);
result._retCode = RetCode.ERROR;
@@ -320,6 +323,12 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
LOG.warn("Node already exists. path: " + path);
result._retCode = RetCode.NODE_EXISTS;
return result;
+ } catch (ZkInterruptedException e) {
+ // The calling thread was interrupted while the request was in flight,
so the outcome is
+ // unknown: ZooKeeper may still have applied it. Returning ERROR would
tell the caller the
+ // write definitely did not happen, which it cannot know. Propagate
instead so that a
+ // false return value always means "not applied". Same for the other
sync operations.
+ throw e;
} catch (Exception e) {
LOG.error("Exception while creating path: " + path, e);
result._retCode = RetCode.ERROR;
@@ -391,12 +400,14 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
result._retCode = RetCode.ERROR;
return result;
}
+ } catch (ZkInterruptedException e1) {
+ throw e1;
} catch (Exception e1) {
LOG.error("Exception while setting path by creating: " + path, e1);
result._retCode = RetCode.ERROR;
return result;
}
- } catch (ZkBadVersionException e) {
+ } catch (ZkBadVersionException | ZkInterruptedException e) {
LOG.debug("Exception while setting path: " + path, e);
throw e;
} catch (Exception e) {
@@ -471,11 +482,15 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
result._retCode = RetCode.ERROR;
return result;
}
+ } catch (ZkInterruptedException e1) {
+ throw e1;
} catch (Exception e1) {
LOG.error("Exception while updating path by creating: " + path, e1);
result._retCode = RetCode.ERROR;
return result;
}
+ } catch (ZkInterruptedException e) {
+ throw e;
} catch (Exception e) {
LOG.error("Exception while updating path: " + path, e);
result._retCode = RetCode.ERROR;
@@ -524,6 +539,8 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
_zkClient.multi(ops);
} catch (ZkBadVersionException e) {
retry = true;
+ } catch (ZkInterruptedException e) {
+ throw e;
} catch (Exception e1) {
LOG.error("Exception while updating paths: " + updaterByPath.keySet(),
e1);
result._retCode = RetCode.ERROR;
@@ -775,6 +792,8 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
// despite real error, operation will throw exception when path not
empty, and in this
// case, we try to delete recursively
_zkClient.delete(path);
+ } catch (ZkInterruptedException e) {
+ throw e;
} catch (ZkException e) {
LOG.debug("Failed to delete {} with opts {}, err: {}. Try recursive
delete", path, options,
e.getMessage());
@@ -801,6 +820,8 @@ public class ZkBaseDataAccessor<T> implements
BaseDataAccessor<T> {
// despite real error, operation will throw exception when path not
empty, and in this
// case, we do NOT try to delete recursively
_zkClient.delete(path, expectedVersion);
+ } catch (ZkInterruptedException e) {
+ throw e;
} catch (ZkException zkException) {
LOG.error("Failed to delete {} with opts {} and expected version {}.",
path, options,
expectedVersion, zkException);
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 2be1a4f76..9691c365d 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
@@ -40,6 +40,7 @@ import org.apache.helix.zookeeper.zkclient.IZkChildListener;
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
import org.apache.helix.zookeeper.zkclient.ZkClient;
import org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks;
+import org.apache.helix.zookeeper.zkclient.exception.ZkInterruptedException;
import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
import org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer;
import org.apache.zookeeper.KeeperException.Code;
@@ -278,6 +279,9 @@ public class ZkCacheBaseDataAccessor<T> implements
HelixPropertyStore<T> {
// no cache
success = _baseAccessor.set(serverPath, data, expectVersion, options);
}
+ } catch (ZkInterruptedException e) {
+ // Outcome unknown, see ZkBaseDataAccessor#doSet. Returning false would
misreport it.
+ throw e;
} catch (Exception e) {
} finally {
if (cache != null) {
diff --git
a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkBaseDataAccessor.java
b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkBaseDataAccessor.java
index 56337f547..fe7c9d390 100644
---
a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkBaseDataAccessor.java
+++
b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkBaseDataAccessor.java
@@ -21,7 +21,9 @@ package org.apache.helix.manager.zk;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Date;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@@ -42,9 +44,13 @@ import
org.apache.helix.zookeeper.exception.ZkClientException;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.exception.ZkBadVersionException;
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.ZkMarshallingError;
import org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer;
+import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
+import org.mockito.ArgumentMatchers;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -912,4 +918,70 @@ public class TestZkBaseDataAccessor extends ZkUnitTestBase
{
+ new Date(System.currentTimeMillis()));
}
+ /*
+ * An interrupt while a request is in flight leaves the outcome unknown
(ZooKeeper may still
+ * apply it), so the sync operations must propagate it rather than report a
definite failure.
+ */
+ @Test
+ public void testSyncOperationsPropagateInterrupt() {
+ RealmAwareZkClient zkClient = Mockito.mock(RealmAwareZkClient.class);
+ ZkInterruptedException interrupted = new ZkInterruptedException(new
InterruptedException());
+ // The constructor above re-asserts the interrupt on the current thread;
clear it.
+ Thread.interrupted();
+
Mockito.doThrow(interrupted).when(zkClient).create(ArgumentMatchers.anyString(),
+ ArgumentMatchers.any(), ArgumentMatchers.any(CreateMode.class),
ArgumentMatchers.anyLong());
+
Mockito.doThrow(interrupted).when(zkClient).writeDataGetStat(ArgumentMatchers.anyString(),
+ ArgumentMatchers.any(), ArgumentMatchers.anyInt());
+ Mockito.doThrow(interrupted).when(zkClient)
+ .readData(ArgumentMatchers.anyString(),
ArgumentMatchers.any(Stat.class));
+
Mockito.doThrow(interrupted).when(zkClient).delete(ArgumentMatchers.anyString());
+ Mockito.doThrow(interrupted).when(zkClient)
+ .delete(ArgumentMatchers.anyString(), ArgumentMatchers.anyInt());
+ Mockito.doThrow(interrupted).when(zkClient).multi(ArgumentMatchers.any());
+
+ ZkBaseDataAccessor<ZNRecord> accessor = new ZkBaseDataAccessor<>(zkClient);
+ String path = "/" + _rootPath + "/interrupted";
+ ZNRecord record = new ZNRecord("interrupted");
+ DataUpdater<ZNRecord> updater = current -> record;
+
+ assertThrowsInterrupted(() -> accessor.create(path, record,
AccessOption.PERSISTENT));
+ assertThrowsInterrupted(() -> accessor.set(path, record,
AccessOption.PERSISTENT));
+ assertThrowsInterrupted(() -> accessor.set(path, record, 3,
AccessOption.PERSISTENT));
+ assertThrowsInterrupted(() -> accessor.update(path, updater,
AccessOption.PERSISTENT));
+ assertThrowsInterrupted(() -> accessor.remove(path,
AccessOption.PERSISTENT));
+ assertThrowsInterrupted(
+ () -> accessor.removeWithExpectedVersion(path,
AccessOption.PERSISTENT, 0));
+ Map<String, DataUpdater<ZNRecord>> updaters = new HashMap<>();
+ updaters.put(path, updater);
+ Mockito.doReturn(record).when(zkClient)
+ .readData(ArgumentMatchers.anyString(),
ArgumentMatchers.any(Stat.class));
+ assertThrowsInterrupted(() -> accessor.multiSet(updaters));
+
+ // The interrupt must also surface from the create-on-missing-node
fallback of set/update.
+ Mockito.doThrow(new ZkNoNodeException()).when(zkClient).writeDataGetStat(
+ ArgumentMatchers.anyString(), ArgumentMatchers.any(),
ArgumentMatchers.anyInt());
+ Mockito.doThrow(new ZkNoNodeException()).when(zkClient)
+ .readData(ArgumentMatchers.anyString(),
ArgumentMatchers.any(Stat.class));
+ assertThrowsInterrupted(() -> accessor.set(path, record,
AccessOption.PERSISTENT));
+ assertThrowsInterrupted(() -> accessor.update(path, updater,
AccessOption.PERSISTENT));
+
+ // ... and from remove()'s recursive-delete fallback for a non-empty node.
+ Mockito.doThrow(new ZkException("not
empty")).when(zkClient).delete(ArgumentMatchers.anyString());
+
Mockito.doThrow(interrupted).when(zkClient).deleteRecursively(ArgumentMatchers.anyString());
+ assertThrowsInterrupted(() -> accessor.remove(path,
AccessOption.PERSISTENT));
+
+ // The cache accessor on top must not swallow it either (property store
users).
+ ZkCacheBaseDataAccessor<ZNRecord> cacheAccessor =
+ new ZkCacheBaseDataAccessor<>(accessor, Collections.emptyList());
+ assertThrowsInterrupted(() -> cacheAccessor.set(path, record,
AccessOption.PERSISTENT));
+ }
+
+ private static void assertThrowsInterrupted(Runnable operation) {
+ try {
+ operation.run();
+ Assert.fail("Expected ZkInterruptedException");
+ } catch (ZkInterruptedException expected) {
+ // expected
+ }
+ }
}
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/ZkSystemPropertyKeys.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/ZkSystemPropertyKeys.java
index 3190acb52..97e581665 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/ZkSystemPropertyKeys.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/ZkSystemPropertyKeys.java
@@ -100,4 +100,22 @@ public class ZkSystemPropertyKeys {
*/
public static final String ZK_OPERATION_RETRY_TIMEOUT_MS =
"zk.operation.retry.timeout.ms";
+
+ /**
+ * This property bounds, in milliseconds, how long a synchronous ZkClient
write (create, setData,
+ * delete) keeps waiting for the outcome of a request that was already
handed to ZooKeeper when
+ * the calling thread got interrupted.
+ * <p>
+ * ZooKeeper queues a request to its send thread before blocking the caller
for the reply, so an
+ * interrupt during that wait does not stop the server from applying the
request. Instead of
+ * reporting such a write as failed, ZkClient waits up to this long for the
real outcome, which
+ * ZooKeeper guarantees to deliver either as the server reply or as a
connection loss, and then
+ * restores the thread's interrupt flag. A second interrupt during the wait
aborts it. Only if
+ * the outcome is still unknown when the wait ends is {@code
ZkInterruptedException} thrown.
+ * <p>
+ * A value of 0 disables the wait and aborts immediately with an unknown
outcome (the behavior
+ * prior to this property). A negative or unset value uses the ZooKeeper
session timeout, which
+ * is the longest the client can take to notice a broken connection.
+ */
+ public static final String ZK_INTERRUPTED_WRITE_WAIT_MS =
"zk.interrupted.write.wait.ms";
}
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 3d46983ac..0aec48e8f 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
@@ -35,6 +35,7 @@ import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
@@ -71,6 +72,7 @@ import
org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer;
import org.apache.helix.zookeeper.zkclient.util.ExponentialBackoffStrategy;
import org.apache.helix.zookeeper.zkclient.util.ZkPathRecursiveWatcherTrie;
import org.apache.zookeeper.AddWatchMode;
+import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.ConnectionLossException;
@@ -151,6 +153,9 @@ public class ZkClient implements Watcher {
// TODO PVo remove this later
private Thread _zookeeperEventThread;
private volatile boolean _closed;
+ // How long a sync write waits for its in-flight request after the caller is
interrupted.
+ // Negative means "use the session timeout". See
ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS.
+ private final long _interruptedWriteWaitMs;
private PathBasedZkSerializer _pathBasedZkSerializer;
private ZkClientMonitor _monitor;
// When _usePersistWatcher is true, ZKClient will register itself as persist
watcher to Zk Server.
@@ -254,6 +259,7 @@ public class ZkClient implements Watcher {
_connection = zkConnection;
_pathBasedZkSerializer = zkSerializer;
_operationRetryTimeoutInMillis = operationRetryTimeout;
+ _interruptedWriteWaitMs = readInterruptedWriteWaitMs();
_isNewSessionEventFired = false;
_asyncCallRetryThread = new ZkAsyncRetryThread(zkConnection.getServers());
@@ -1124,11 +1130,17 @@ public class ZkClient implements Watcher {
final String actualPath;
if (mode.isTTL()) {
- actualPath = retryUntilConnected(() ->
getExpectedZookeeper(expectedSessionId)
- .create(path, dataBytes, acl, mode, null, ttl));
+ actualPath = retryUntilConnected(
+ () -> this.<String>syncWrite(path,
getExpectedZookeeper(expectedSessionId),
+ (zk, write) -> zk.create(path, dataBytes, acl, mode,
+ (AsyncCallback.Create2Callback) (rc, p, ctx, name, stat) ->
+ write.complete(rc, name), null, ttl)));
} else {
- actualPath = retryUntilConnected(() ->
getExpectedZookeeper(expectedSessionId)
- .create(path, dataBytes, acl, mode));
+ actualPath = retryUntilConnected(
+ () -> this.<String>syncWrite(path,
getExpectedZookeeper(expectedSessionId),
+ (zk, write) -> zk.create(path, dataBytes, acl, mode,
+ (AsyncCallback.StringCallback) (rc, p, ctx, name) ->
write.complete(rc, name),
+ null)));
}
record(path, dataBytes, startT, ZkClientMonitor.AccessType.WRITE);
@@ -1831,6 +1843,9 @@ public class ZkClient implements Watcher {
// if there is any other failure case.
try {
delete(path);
+ } catch (ZkInterruptedException e) {
+ // Outcome unknown (see awaitInFlightWrite); do not report it as a plain
failure.
+ throw e;
} catch (Exception e) {
LOG.error("zkclient {}, Failed to delete {}, exception {}", _uid, path,
e);
throw new ZkClientException("Failed to delete " + path, e);
@@ -1865,6 +1880,9 @@ public class ZkClient implements Watcher {
try {
opResults = multi(ops);
+ } catch (ZkInterruptedException e) {
+ // Outcome unknown (see awaitInFlightWrite); do not report it as a plain
failure.
+ throw e;
} catch (Exception e) {
LOG.error("zkclient {}, Failed to delete paths {}, exception {}", _uid,
paths, e);
throw new ZkClientException("Failed to delete paths " + paths, e);
@@ -2130,6 +2148,122 @@ public class ZkClient implements Watcher {
}
}
+ /**
+ * Outcome holder for a write submitted through the ZooKeeper async API.
Used by the synchronous
+ * write paths so that an interrupt of the calling thread does not lose the
outcome of a request
+ * ZooKeeper has already queued. See {@link #awaitInFlightWrite(String,
InFlightWrite, long)}.
+ */
+ @VisibleForTesting
+ static class InFlightWrite<T> {
+ private final CountDownLatch _done = new CountDownLatch(1);
+ volatile int _rc = KeeperException.Code.APIERROR.intValue();
+ private volatile T _result;
+
+ void complete(int rc, T result) {
+ _rc = rc;
+ _result = result;
+ _done.countDown();
+ }
+
+ /** 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);
+ }
+ }
+
+ /** Issues one ZooKeeper write through the async API, completing the given
holder. */
+ @FunctionalInterface
+ private interface WriteSubmitter<T> {
+ void submit(ZooKeeper zk, InFlightWrite<T> write);
+ }
+
+ /**
+ * Runs a synchronous write (create, setData, delete, multi) by submitting
it through the
+ * ZooKeeper async API and waiting for the outcome, so that an interrupt of
the calling thread
+ * does not lose the result of a request ZooKeeper has already queued. Meant
to be called inside
+ * {@link #retryUntilConnected(Callable)}, which handles the retryable error
codes.
+ */
+ private <T> T syncWrite(final String path, final ZooKeeper zk, final
WriteSubmitter<T> submitter)
+ throws KeeperException, InterruptedException {
+ return syncWrite(path, zk, new InFlightWrite<>(), submitter);
+ }
+
+ private <T> T syncWrite(final String path, final ZooKeeper zk, final
InFlightWrite<T> write,
+ final WriteSubmitter<T> submitter) throws KeeperException,
InterruptedException {
+ submitter.submit(zk, write);
+ long interruptedWaitMs =
+ _interruptedWriteWaitMs < 0 ? zk.getSessionTimeout() :
_interruptedWriteWaitMs;
+ return awaitInFlightWrite(path, write, interruptedWaitMs);
+ }
+
+ private ZooKeeper currentZookeeper() {
+ return ((ZkConnection) getConnection()).getZookeeper();
+ }
+
+ /**
+ * Waits for a write that has already been submitted to ZooKeeper and
returns its result, or
+ * throws the KeeperException matching its error code.
+ * <p>
+ * ZooKeeper's synchronous API queues the request to the send thread before
blocking for the
+ * reply, so an interrupt during that wait surfaces as InterruptedException
even though the
+ * server still applies the request. The synchronous write paths therefore
submit through the
+ * async API and come here: if the calling thread is interrupted while
waiting, we keep waiting
+ * up to {@code interruptedWaitMs} ({@link
ZkSystemPropertyKeys#ZK_INTERRUPTED_WRITE_WAIT_MS},
+ * defaulting to the session timeout) because every queued packet is
guaranteed to be completed,
+ * either with the server reply or with CONNECTIONLOSS / SESSIONEXPIRED once
the connection
+ * drops. A second interrupt during that wait aborts it, so a caller that
really wants out (such
+ * as ZKHelixManager#disconnect re-interrupting its cleanup thread while the
connection is down)
+ * is not held for the session timeout. The interrupt flag is restored
before returning so
+ * callers still observe the interrupt; only if the outcome is still unknown
when the wait ends
+ * is the InterruptedException propagated.
+ */
+ @VisibleForTesting
+ static <T> T awaitInFlightWrite(final String path, final InFlightWrite<T>
write,
+ final long interruptedWaitMs) throws KeeperException,
InterruptedException {
+ try {
+ write._done.await();
+ } catch (InterruptedException e) {
+ boolean done = awaitOutcomeAfterInterrupt(write._done,
interruptedWaitMs);
+ // Preserve the interrupt for the caller whether we report the outcome
or give up.
+ Thread.currentThread().interrupt();
+ if (!done) {
+ LOG.warn("Caller interrupted while a write to {} was in flight; giving
up after at most {}"
+ + " ms with the outcome unknown", path, interruptedWaitMs);
+ throw e;
+ }
+ LOG.info("Caller interrupted while a write to {} was in flight; waited
for its outcome: {}",
+ path, KeeperException.Code.get(write._rc));
+ }
+ if (write._rc != KeeperException.Code.OK.intValue()) {
+ throw write.failure(path);
+ }
+ return write._result;
+ }
+
+ private static boolean awaitOutcomeAfterInterrupt(CountDownLatch done, long
waitMs) {
+ try {
+ return done.await(waitMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ // Interrupted again: the caller wants out now. Report the outcome only
if it already arrived.
+ return done.getCount() == 0;
+ }
+ }
+
+ @VisibleForTesting
+ static long readInterruptedWriteWaitMs() {
+ String value =
System.getProperty(ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS);
+ if (value == null || value.trim().isEmpty()) {
+ return -1;
+ }
+ try {
+ return Long.parseLong(value.trim());
+ } catch (NumberFormatException e) {
+ LOG.warn("Invalid value {} for {}, falling back to the session timeout",
value,
+ ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS);
+ return -1;
+ }
+ }
+
/**
* @param <T>
* @param callable
@@ -2276,14 +2410,9 @@ public class ZkClient implements Watcher {
boolean success;
try {
try {
- retryUntilConnected(new Callable<Object>() {
-
- @Override
- public Object call() throws Exception {
- getConnection().delete(path, expectedVersion);
- return null;
- }
- });
+ retryUntilConnected(() -> this.<Void>syncWrite(path,
currentZookeeper(),
+ (zk, write) -> zk.delete(path, expectedVersion,
+ (AsyncCallback.VoidCallback) (rc, p, ctx) ->
write.complete(rc, null), null)));
success = true;
} catch (ZkNoNodeException e) {
success = false;
@@ -2436,8 +2565,9 @@ public class ZkClient implements Watcher {
try {
final byte[] data = serialize(datat, path);
checkDataSizeLimit(path, data);
- final Stat stat = (Stat) retryUntilConnected(
- (Callable<Object>) () -> getConnection().writeDataReturnStat(path,
data, expectedVersion));
+ final Stat stat = retryUntilConnected(() -> this.<Stat>syncWrite(path,
currentZookeeper(),
+ (zk, write) -> zk.setData(path, data, expectedVersion,
+ (AsyncCallback.StatCallback) (rc, p, ctx, s) ->
write.complete(rc, s), null)));
record(path, data, startT, ZkClientMonitor.AccessType.WRITE);
return stat;
} catch (Exception e) {
@@ -2941,12 +3071,23 @@ public class ZkClient implements Watcher {
throw new NullPointerException("ops must not be null.");
}
- return retryUntilConnected(new Callable<List<OpResult>>() {
+ if (!ops.iterator().hasNext()) {
+ // Nothing to do; ZooKeeper's synchronous multi also returns early for
an empty transaction.
+ return Collections.emptyList();
+ }
- @Override
- public List<OpResult> call() throws Exception {
- return getConnection().multi(ops);
- }
+ return retryUntilConnected(() -> {
+ // Mirror ZooKeeper's synchronous multi, whose failure exception carries
no path. (It also
+ // attaches the per-op results, but KeeperException#setMultiResults is
package-private and
+ // nothing in Helix reads them.)
+ InFlightWrite<List<OpResult>> write = new
InFlightWrite<List<OpResult>>() {
+ @Override
+ KeeperException failure(String path) {
+ return KeeperException.create(KeeperException.Code.get(_rc));
+ }
+ };
+ return syncWrite(null, currentZookeeper(), write, (zk, w) ->
zk.multi(ops,
+ (AsyncCallback.MultiCallback) (rc, p, ctx, results) ->
w.complete(rc, results), null));
});
}
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 ed25303c0..948cdd88a 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
@@ -1341,4 +1341,49 @@ public class TestRawZkClient extends ZkTestBase {
Assert.assertFalse(_zkClient.exists(grandParent));
Assert.assertFalse(_zkClient.exists(newNode));
}
+
+ /*
+ * ZooKeeper's synchronous API hands a request to the send thread before
blocking for the reply,
+ * so a thread interrupted during that wait used to get
ZkInterruptedException even though the
+ * server still applied the write. ZkClient now waits out the in-flight
request and reports its
+ * real outcome, with the interrupt flag left set for the caller to observe.
+ */
+ @Test
+ public void testSyncWriteOutcomePreservedWhenThreadInterrupted() {
+ String path = TEST_ROOT + "/interruptedWrite";
+ _zkClient.createPersistent(TEST_ROOT, true);
+ _zkClient.delete(path);
+ try {
+ Thread.currentThread().interrupt();
+ _zkClient.createPersistent(path, "v1");
+ Assert.assertTrue(Thread.interrupted(), "Interrupt flag must survive
create");
+ Assert.assertEquals(_zkClient.readData(path), "v1");
+
+ Thread.currentThread().interrupt();
+ Stat stat = _zkClient.writeDataGetStat(path, "v2", -1);
+ Assert.assertTrue(Thread.interrupted(), "Interrupt flag must survive
setData");
+ Assert.assertNotNull(stat);
+ Assert.assertEquals(stat.getVersion(), 1);
+ Assert.assertEquals(_zkClient.readData(path), "v2");
+
+ Thread.currentThread().interrupt();
+ Assert.assertTrue(_zkClient.delete(path));
+ Assert.assertTrue(Thread.interrupted(), "Interrupt flag must survive
delete");
+ Assert.assertFalse(_zkClient.exists(path));
+
+ // A definite failure is still reported as that failure, not as an
interrupt.
+ Thread.currentThread().interrupt();
+ try {
+ _zkClient.writeData(path, "v3");
+ Assert.fail("Expected ZkNoNodeException");
+ } catch (ZkNoNodeException expected) {
+ // expected
+ }
+ Assert.assertTrue(Thread.interrupted(), "Interrupt flag must survive a
failed setData");
+ } finally {
+ // Never leak an interrupt into the next test.
+ Thread.interrupted();
+ _zkClient.delete(path);
+ }
+ }
}
diff --git
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/zkclient/TestZkClientInFlightWrite.java
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/zkclient/TestZkClientInFlightWrite.java
new file mode 100644
index 000000000..e1143490a
--- /dev/null
+++
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/zkclient/TestZkClientInFlightWrite.java
@@ -0,0 +1,184 @@
+package org.apache.helix.zookeeper.zkclient;
+
+/*
+ * 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.
+ */
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.helix.zookeeper.constant.ZkSystemPropertyKeys;
+import org.apache.helix.zookeeper.zkclient.ZkClient.InFlightWrite;
+import org.apache.zookeeper.KeeperException;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Unit tests for the wait that the synchronous ZkClient writes perform when
the calling thread is
+ * interrupted while the request is already queued to ZooKeeper. The outcome
is driven by hand
+ * here so the timing-sensitive cases are deterministic; the end-to-end
behavior against a real
+ * server is covered in TestRawZkClient.
+ */
+public class TestZkClientInFlightWrite {
+ private static final String PATH = "/in-flight";
+ private static final long WAIT_MS = TimeUnit.SECONDS.toMillis(30);
+
+ @BeforeClass
+ public void loadZkClient() {
+ // Initialize ZkClient (and its logger) before any test interrupts the
thread: logging setup
+ // on an interrupted thread fails and would poison the class for the rest
of the run.
+ ZkClient.readInterruptedWriteWaitMs();
+ }
+
+ @AfterMethod
+ public void clearInterrupt() {
+ Thread.interrupted();
+ }
+
+ @Test
+ public void testOutcomeReturnedWithoutInterrupt() throws Exception {
+ InFlightWrite<String> write = new InFlightWrite<>();
+ write.complete(KeeperException.Code.OK.intValue(), "result");
+ Assert.assertEquals(ZkClient.awaitInFlightWrite(PATH, write, WAIT_MS),
"result");
+ Assert.assertFalse(Thread.currentThread().isInterrupted());
+ }
+
+ @Test
+ public void testInterruptedCallerStillGetsSuccess() throws Exception {
+ InFlightWrite<String> write = new InFlightWrite<>();
+ completeLater(write, KeeperException.Code.OK, "result");
+ Thread.currentThread().interrupt();
+ Assert.assertEquals(ZkClient.awaitInFlightWrite(PATH, write, WAIT_MS),
"result");
+ Assert.assertTrue(Thread.interrupted(), "Interrupt flag must be
preserved");
+ }
+
+ @Test
+ public void testInterruptedCallerStillGetsDefiniteFailure() {
+ InFlightWrite<String> write = new InFlightWrite<>();
+ completeLater(write, KeeperException.Code.NONODE, null);
+ Thread.currentThread().interrupt();
+ try {
+ ZkClient.awaitInFlightWrite(PATH, write, WAIT_MS);
+ Assert.fail("Expected NoNodeException");
+ } catch (KeeperException.NoNodeException expected) {
+ Assert.assertEquals(expected.getPath(), PATH);
+ } catch (Exception e) {
+ Assert.fail("Expected NoNodeException", e);
+ }
+ Assert.assertTrue(Thread.interrupted(), "Interrupt flag must be
preserved");
+ }
+
+ @Test
+ public void testInterruptedCallerGivesUpWhenWaitDisabled() {
+ InFlightWrite<String> write = new InFlightWrite<>();
+ Thread.currentThread().interrupt();
+ try {
+ ZkClient.awaitInFlightWrite(PATH, write, 0);
+ Assert.fail("Expected InterruptedException");
+ } catch (InterruptedException expected) {
+ // expected
+ } catch (Exception e) {
+ Assert.fail("Expected InterruptedException", e);
+ }
+ Assert.assertTrue(Thread.interrupted(), "Interrupt flag must be
preserved");
+ }
+
+ @Test
+ public void testInterruptedCallerGivesUpWhenOutcomeNeverArrives() {
+ InFlightWrite<String> write = new InFlightWrite<>();
+ Thread.currentThread().interrupt();
+ long start = System.nanoTime();
+ try {
+ ZkClient.awaitInFlightWrite(PATH, write, 200);
+ Assert.fail("Expected InterruptedException");
+ } catch (InterruptedException expected) {
+ // expected
+ } catch (Exception e) {
+ Assert.fail("Expected InterruptedException", e);
+ }
+ Assert.assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)
>= 200,
+ "Must wait out the configured bound before giving up");
+ Assert.assertTrue(Thread.interrupted(), "Interrupt flag must be
preserved");
+ }
+
+ @Test
+ public void testSecondInterruptAbortsTheWait() {
+ InFlightWrite<String> write = new InFlightWrite<>();
+ Thread caller = Thread.currentThread();
+ Thread interrupter = new Thread(() -> {
+ sleepQuietly(200);
+ caller.interrupt();
+ });
+ interrupter.start();
+ Thread.currentThread().interrupt();
+ long start = System.nanoTime();
+ try {
+ ZkClient.awaitInFlightWrite(PATH, write, WAIT_MS);
+ Assert.fail("Expected InterruptedException");
+ } catch (InterruptedException expected) {
+ // expected
+ } catch (Exception e) {
+ Assert.fail("Expected InterruptedException", e);
+ }
+ Assert.assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)
< WAIT_MS / 2,
+ "A second interrupt must abort the wait well before the bound");
+ Assert.assertTrue(Thread.interrupted(), "Interrupt flag must be
preserved");
+ }
+
+ @Test
+ public void testReadInterruptedWriteWaitMs() {
+ String original =
System.getProperty(ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS);
+ try {
+ System.clearProperty(ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS);
+ Assert.assertEquals(ZkClient.readInterruptedWriteWaitMs(), -1);
+ System.setProperty(ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS, "
");
+ Assert.assertEquals(ZkClient.readInterruptedWriteWaitMs(), -1);
+ System.setProperty(ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS,
"not-a-number");
+ Assert.assertEquals(ZkClient.readInterruptedWriteWaitMs(), -1);
+ System.setProperty(ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS,
"0");
+ Assert.assertEquals(ZkClient.readInterruptedWriteWaitMs(), 0);
+ System.setProperty(ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS, "
1500 ");
+ Assert.assertEquals(ZkClient.readInterruptedWriteWaitMs(), 1500);
+ } finally {
+ if (original == null) {
+
System.clearProperty(ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS);
+ } else {
+ System.setProperty(ZkSystemPropertyKeys.ZK_INTERRUPTED_WRITE_WAIT_MS,
original);
+ }
+ }
+ }
+
+ private static <T> void completeLater(InFlightWrite<T> write,
KeeperException.Code code,
+ T result) {
+ Thread completer = new Thread(() -> {
+ sleepQuietly(200);
+ write.complete(code.intValue(), result);
+ });
+ completer.setDaemon(true);
+ completer.start();
+ }
+
+ private static void sleepQuietly(long ms) {
+ try {
+ Thread.sleep(ms);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+}
