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 fc7843735 Fail fast on oversized ZK transactions and let dropInstance
fall back (#3073)
fc7843735 is described below
commit fc784373502899d2c2484cfdd9f04f576c1f7ec7
Author: Junkai Xue <[email protected]>
AuthorDate: Fri Nov 6 00:57:44 2026 -0800
Fail fast on oversized ZK transactions and let dropInstance fall back
(#3073)
ZKHelixAdmin.dropInstance() deletes the instance path and its config in one
ZK transaction (deleteRecursivelyAtomic, #2994). An instance that
accumulated
thousands of current/customized states produces a multi packet larger than
jute.maxbuffer. The server rejects such a packet by closing the connection
without replying, ZkClient.retryUntilConnected() sees a connection loss and
re-sends the same packet, so the drop never succeeds and never fails until
the
operation retry timeout.
ZkClient.multi() now measures the request packet (request header plus
MultiOperationRecord, serialized into a counting stream so nothing is
buffered) and throws the new ZkTransactionTooLargeException, a
ZkClientException, before sending anything when it exceeds the write size
limit (the client's jute.maxbuffer, ZNRecord.SIZE_LIMIT by default, the same
limit already applied to single writes). deleteRecursivelyAtomic() re-throws
it with the paths added; nothing has been deleted at that point.
ZKHelixAdmin.dropInstancePathsRecursively() catches that exception and
deletes the instance's sub-trees (everything two or more levels below the
instance path: individual current states, customized states, messages) in a
non-atomic pass, each sub-tree in one transaction when it fits and znode by
znode otherwise, then retries the atomic delete of the remaining instance
path, its direct children and the instance config. The instance is therefore
still never left half-dropped. The fallback re-checks that the instance has
no live instance first, wraps ZK failures into HelixException like the rest
of the method, and gives up with HelixException if the skeleton still does
not fit.
Tests: TestRawZkClient covers getTransactionSize against ZooKeeper's own
serialization and the fail-fast of multi()/deleteRecursivelyAtomic() on a
~1.5 MB tree without deleting anything. TestZkHelixAdmin drops an instance
whose current states (~3 MB of delete operations, one session too large for
a transaction on its own) exceed the limit and verifies both the instance
path and config are gone.
---
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 72 +++++++++++++++
.../apache/helix/manager/zk/TestZkHelixAdmin.java | 68 ++++++++++++++
.../exception/ZkTransactionTooLargeException.java | 35 +++++++
.../apache/helix/zookeeper/zkclient/ZkClient.java | 66 +++++++++++++-
.../zookeeper/impl/client/TestRawZkClient.java | 101 +++++++++++++++++++++
5 files changed, 341 insertions(+), 1 deletion(-)
diff --git
a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
index 5d917587e..99e2b56aa 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
@@ -99,12 +99,14 @@ import
org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.exception.ZkClientException;
+import org.apache.helix.zookeeper.exception.ZkTransactionTooLargeException;
import org.apache.helix.zookeeper.impl.client.FederatedZkClient;
import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.NetworkUtil;
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.zookeeper.KeeperException;
import org.apache.zookeeper.Op;
@@ -281,10 +283,23 @@ public class ZKHelixAdmin implements HelixAdmin {
String instanceConfigPath =
PropertyPathBuilder.instanceConfig(clusterName, instanceName);
String instancePath = PropertyPathBuilder.instance(clusterName,
instanceName);
int retryCnt = 0;
+ boolean subtreesDeleted = false;
while (true) {
try {
_zkClient.deleteRecursivelyAtomic(Arrays.asList(instancePath,
instanceConfigPath));
return;
+ } catch (ZkTransactionTooLargeException e) {
+ if (subtreesDeleted) {
+ String errorMessage = "Failed to drop instance: " + instanceName
+ + ". Its znodes still do not fit in one ZK transaction after
deleting its sub-trees";
+ logger.error(errorMessage, e);
+ throw new HelixException(errorMessage, e);
+ }
+ logger.warn("Instance {} has too many znodes to be dropped in one ZK
transaction ({}). "
+ + "Deleting its sub-trees non-atomically first, then dropping the
remaining paths "
+ + "atomically.", instanceName, e.getMessage());
+ deleteInstanceSubtrees(clusterName, instanceName);
+ subtreesDeleted = true;
} catch (ZkClientException e) {
if (retryCnt < 3 && e.getCause() instanceof ZkException && e.getCause()
.getCause() instanceof KeeperException.NotEmptyException) {
@@ -304,6 +319,63 @@ public class ZKHelixAdmin implements HelixAdmin {
}
}
+ /**
+ * Deletes, non-atomically, everything two or more levels below the instance
path, e.g. the
+ * individual current states, customized states and messages, which is where
an instance
+ * accumulates thousands of znodes. The instance path itself, its direct
children (CURRENTSTATES,
+ * MESSAGES, ...) and the instance config are left in place so that the
caller can still remove
+ * them in one atomic transaction afterwards, which keeps the guarantee that
an instance is never
+ * left half-dropped (e.g. without its config but with its instance path).
+ * <p>
+ * Removing only these deeper znodes is safe for an instance that is not
live, because the
+ * controller reads current states, customized states and messages of live
instances only.
+ * Liveness is re-checked here since the deletes below are not atomic and
can take a while.
+ */
+ private void deleteInstanceSubtrees(String clusterName, String instanceName)
{
+ if (_zkClient.exists(PropertyPathBuilder.liveInstance(clusterName,
instanceName))) {
+ throw new HelixException("Node " + instanceName + " is still alive for
cluster " + clusterName
+ + ", can't drop.");
+ }
+ String instancePath = PropertyPathBuilder.instance(clusterName,
instanceName);
+ try {
+ for (String child : getChildrenOrEmpty(instancePath)) {
+ String childPath = instancePath + "/" + child;
+ for (String grandChild : getChildrenOrEmpty(childPath)) {
+ deleteSubtree(childPath + "/" + grandChild);
+ }
+ }
+ } catch (ZkInterruptedException e) {
+ // Outcome of the in-flight delete is unknown; do not report it as a
plain failure.
+ throw e;
+ } catch (ZkClientException | ZkException e) {
+ String errorMessage = "Failed to drop instance: " + instanceName
+ + ". Could not delete its sub-trees non-atomically";
+ logger.error(errorMessage, e);
+ throw new HelixException(errorMessage, e);
+ }
+ }
+
+ /**
+ * Deletes a sub-tree in as few ZK round trips as possible: in one
transaction when it fits,
+ * otherwise znode by znode.
+ */
+ private void deleteSubtree(String path) {
+ try {
+ _zkClient.deleteRecursivelyAtomic(path);
+ } catch (ZkTransactionTooLargeException e) {
+ _zkClient.deleteRecursively(path);
+ }
+ }
+
+ private List<String> getChildrenOrEmpty(String path) {
+ try {
+ return _zkClient.getChildren(path);
+ } catch (ZkNoNodeException e) {
+ // Deleted concurrently, nothing left to delete under it.
+ return Collections.emptyList();
+ }
+ }
+
/**
* Please note that the purge function should only be called when there is
no new instance
* joining happening in the cluster. The reason is that current
implementation is not thread safe,
diff --git
a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java
b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java
index b60b909a2..65c827a79 100644
--- a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java
+++ b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java
@@ -80,9 +80,13 @@ import org.apache.helix.tools.StateModelConfigGenerator;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.exception.ZkClientException;
+import org.apache.helix.zookeeper.exception.ZkTransactionTooLargeException;
import org.apache.helix.zookeeper.zkclient.NetworkUtil;
import org.apache.helix.zookeeper.zkclient.exception.ZkException;
+import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;
import org.mockito.Mockito;
import org.testng.Assert;
@@ -1367,4 +1371,68 @@ public class TestZkHelixAdmin extends ZkUnitTestBase {
System.out.println("End test :" + TestHelper.getTestMethodName());
}
+
+ /*
+ * An instance that accumulated thousands of current/customized states
cannot be dropped in one
+ * ZK transaction (jute.maxbuffer). dropInstance must then delete the
instance's sub-trees
+ * non-atomically and still remove the instance path and config atomically.
+ */
+ @Test
+ public void testDropInstanceTooLargeForOneTransaction() {
+ System.out.println("Start test :" + TestHelper.getTestMethodName());
+ final String clusterName = getShortClassName() + "_" +
TestHelper.getTestMethodName();
+ final String instanceName = "localhost_12918";
+ HelixAdmin admin = new ZKHelixAdmin(_gZkClient);
+ admin.addCluster(clusterName, true);
+ admin.addInstance(clusterName, new InstanceConfig(instanceName));
+ String instancePath = PropertyPathBuilder.instance(clusterName,
instanceName);
+ String instanceConfigPath =
PropertyPathBuilder.instanceConfig(clusterName, instanceName);
+
+ // session1 holds 700 long-named current states that each hold one more
long-named znode, so
+ // even this single session does not fit in one transaction and has to be
deleted znode by
+ // znode. session2 holds 700 long-named current states, which fit in one
transaction once
+ // deleted on their own. Together they make the drop of the whole instance
(~3 MB) exceed
+ // jute.maxbuffer, while each getChildren() response stays below it.
+ String namePrefix = String.join("", Collections.nCopies(1000, "r"));
+ String session1Path =
+ PropertyPathBuilder.instanceCurrentState(clusterName, instanceName,
"session1");
+ String session2Path =
+ PropertyPathBuilder.instanceCurrentState(clusterName, instanceName,
"session2");
+ for (String sessionPath : Arrays.asList(session1Path, session2Path)) {
+ _gZkClient.createPersistent(sessionPath, true);
+ List<Op> creates = new ArrayList<>();
+ for (int i = 0; i < 700; i++) {
+ String resourcePath = sessionPath + "/" + namePrefix + i;
+ creates.add(Op.create(resourcePath, new byte[0],
ZooDefs.Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT));
+ if (sessionPath.equals(session1Path)) {
+ creates.add(Op.create(resourcePath + "/" + namePrefix, new byte[0],
+ ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
+ }
+ if (creates.size() >= 50 || i == 699) {
+ _gZkClient.multi(creates);
+ creates.clear();
+ }
+ }
+ }
+ // Make sure the test really exercises both fallbacks: neither the whole
instance nor session1
+ // alone fits in one transaction.
+ for (String path : Arrays.asList(instancePath, session1Path)) {
+ try {
+ _gZkClient.deleteRecursivelyAtomic(path);
+ Assert.fail(path + " should not fit in one transaction");
+ } catch (ZkTransactionTooLargeException expected) {
+ // expected
+ }
+ }
+ Assert.assertTrue(_gZkClient.exists(instancePath));
+
+ admin.dropInstance(clusterName, new InstanceConfig(instanceName));
+
+ Assert.assertFalse(_gZkClient.exists(instancePath));
+ Assert.assertFalse(_gZkClient.exists(instanceConfigPath));
+ Assert.assertTrue(admin.getInstancesInCluster(clusterName).isEmpty());
+ admin.dropCluster(clusterName);
+ System.out.println("End test :" + TestHelper.getTestMethodName());
+ }
}
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/exception/ZkTransactionTooLargeException.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/exception/ZkTransactionTooLargeException.java
new file mode 100644
index 000000000..a2f54d048
--- /dev/null
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/exception/ZkTransactionTooLargeException.java
@@ -0,0 +1,35 @@
+package org.apache.helix.zookeeper.exception;
+
+/*
+ * 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.
+ */
+
+/**
+ * Thrown when a ZooKeeper transaction (multi) would be larger than the packet
size limit
+ * ({@code jute.maxbuffer}) and is therefore rejected before being sent.
Nothing has been
+ * written to ZooKeeper when this exception is thrown.
+ */
+public class ZkTransactionTooLargeException extends ZkClientException {
+ public ZkTransactionTooLargeException(String message) {
+ super(message);
+ }
+
+ public ZkTransactionTooLargeException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
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 0aec48e8f..95ca167be 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
@@ -19,6 +19,7 @@ package org.apache.helix.zookeeper.zkclient;
* under the License.
*/
+import java.io.IOException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
@@ -45,11 +46,15 @@ import javax.security.auth.login.AppConfigurationEntry;
import javax.security.auth.login.Configuration;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Iterables;
+import com.google.common.io.ByteStreams;
+import com.google.common.io.CountingOutputStream;
import org.apache.helix.zookeeper.api.client.ChildrenSubscribeResult;
import org.apache.helix.zookeeper.constant.ZkSystemPropertyKeys;
import org.apache.helix.zookeeper.datamodel.SessionAwareZNRecord;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.exception.ZkClientException;
+import org.apache.helix.zookeeper.exception.ZkTransactionTooLargeException;
import org.apache.helix.zookeeper.util.GZipCompressionUtil;
import org.apache.helix.zookeeper.util.ZNRecordUtil;
import org.apache.helix.zookeeper.zkclient.annotation.PreFetchChangedData;
@@ -71,12 +76,14 @@ import
org.apache.helix.zookeeper.zkclient.serialize.PathBasedZkSerializer;
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.jute.BinaryOutputArchive;
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;
import org.apache.zookeeper.KeeperException.SessionExpiredException;
+import org.apache.zookeeper.MultiOperationRecord;
import org.apache.zookeeper.Op;
import org.apache.zookeeper.OpResult;
import org.apache.zookeeper.WatchedEvent;
@@ -88,6 +95,7 @@ import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.client.ZKClientConfig;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.proto.RequestHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -118,7 +126,9 @@ public class ZkClient implements Watcher {
private static AtomicLong UID = new AtomicLong(0);
public final long _uid;
- // ZNode write size limit in bytes:
+ // ZNode write size limit in bytes. This is the client JVM's jute.maxbuffer,
or ZNRecord.SIZE_LIMIT
+ // (which leaves a margin under ZooKeeper's 1 MB default) when it is not
set. The server is what
+ // rejects an oversized packet, so this assumes the server's jute.maxbuffer
is not smaller.
// TODO: use ZKConfig#JUTE_MAXBUFFER once bumping up ZK to 3.5.2+
private static final int WRITE_SIZE_LIMIT =
Integer.getInteger(ZkSystemPropertyKeys.JUTE_MAXBUFFER,
ZNRecord.SIZE_LIMIT);
@@ -1864,7 +1874,13 @@ public class ZkClient implements Watcher {
/**
* Delete the paths as well as all their children. This operation is atomic
and will either delete all nodes or none.
* This operation may fail if another agent is concurrently creating or
deleting nodes under any of the paths.
+ * <p>
+ * All deletes are sent to ZooKeeper in a single transaction, see {@link
#multi(Iterable)} for its
+ * size limit. If the paths hold too many znodes for the transaction to fit,
this method throws
+ * {@link ZkTransactionTooLargeException} without deleting anything; callers
that can accept a
+ * non-atomic delete should then fall back to {@link
#deleteRecursively(String)}.
* @param paths ZK paths to delete
+ * @throws ZkTransactionTooLargeException if the deletes do not fit in one
ZooKeeper transaction
*/
public void deleteRecursivelyAtomic(List<String> paths) {
List<Op> ops = new ArrayList<>();
@@ -1880,6 +1896,10 @@ public class ZkClient implements Watcher {
try {
opResults = multi(ops);
+ } catch (ZkTransactionTooLargeException e) {
+ // Nothing was sent to ZooKeeper; let the caller decide whether to
delete non-atomically.
+ throw new ZkTransactionTooLargeException(
+ "Cannot atomically delete paths " + paths + ": " + e.getMessage(),
e);
} catch (ZkInterruptedException e) {
// Outcome unknown (see awaitInFlightWrite); do not report it as a plain
failure.
throw e;
@@ -1933,6 +1953,40 @@ public class ZkClient implements Watcher {
return ops;
}
+ /**
+ * Fails fast if the transaction packet would be larger than the write size
limit. The server
+ * rejects such a packet by closing the connection without replying, which
+ * {@link #retryUntilConnected(Callable)} would treat as a connection loss
and answer by re-sending
+ * the same oversized packet until the operation retry timeout.
+ */
+ private void checkTransactionSizeLimit(Iterable<Op> ops) {
+ long transactionSize = getTransactionSize(ops);
+ if (transactionSize > WRITE_SIZE_LIMIT) {
+ throw new ZkTransactionTooLargeException("Transaction of " +
Iterables.size(ops)
+ + " operations is " + transactionSize + " bytes, which is greater
than write size limit "
+ + WRITE_SIZE_LIMIT + " bytes");
+ }
+ }
+
+ /**
+ * Returns the size in bytes of the request packet ZooKeeper sends for
{@code multi(ops)}, which is
+ * what the server compares against {@code jute.maxbuffer}. The packet is
serialized into a
+ * counting stream, so no copy of it is kept in memory.
+ */
+ @VisibleForTesting
+ public static long getTransactionSize(Iterable<Op> ops) {
+ CountingOutputStream counter = new
CountingOutputStream(ByteStreams.nullOutputStream());
+ BinaryOutputArchive archive = BinaryOutputArchive.getArchive(counter);
+ try {
+ new RequestHeader(0, ZooDefs.OpCode.multi).serialize(archive, "header");
+ new MultiOperationRecord(ops).serialize(archive, "multi");
+ } catch (IOException e) {
+ // The counting stream never throws; keep the checked exception from
leaking.
+ throw new ZkMarshallingError("Failed to serialize transaction", e);
+ }
+ return counter.getCount();
+ }
+
private void processDataOrChildChange(WatchedEvent event, long
notificationTime) {
final String path = event.getPath();
final boolean pathExists = event.getType() != EventType.NodeDeleted;
@@ -3066,6 +3120,14 @@ public class ZkClient implements Watcher {
return listeners;
}
+ /**
+ * Executes the operations as one ZooKeeper transaction.
+ * <p>
+ * The transaction is sent as a single packet, whose size is bounded by the
server's
+ * {@code jute.maxbuffer}. Its size is checked against the write size limit
before it is sent.
+ * @throws ZkTransactionTooLargeException if the transaction does not fit in
one packet; nothing
+ * has been sent to ZooKeeper in that case
+ */
public List<OpResult> multi(final Iterable<Op> ops) throws ZkException {
if (ops == null) {
throw new NullPointerException("ops must not be null.");
@@ -3076,6 +3138,8 @@ public class ZkClient implements Watcher {
return Collections.emptyList();
}
+ checkTransactionSizeLimit(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
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 948cdd88a..2e92ab6b1 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
@@ -19,9 +19,13 @@ package org.apache.helix.zookeeper.impl.client;
* under the License.
*/
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
import java.lang.management.ManagementFactory;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.UUID;
@@ -44,6 +48,7 @@ import
org.apache.helix.zookeeper.datamodel.SessionAwareZNRecord;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.helix.zookeeper.exception.ZkClientException;
+import org.apache.helix.zookeeper.exception.ZkTransactionTooLargeException;
import org.apache.helix.zookeeper.impl.TestHelper;
import org.apache.helix.zookeeper.impl.ZkTestBase;
import org.apache.helix.zookeeper.impl.ZkTestHelper;
@@ -59,8 +64,10 @@ import
org.apache.helix.zookeeper.zkclient.exception.ZkSessionMismatchedExceptio
import org.apache.helix.zookeeper.zkclient.exception.ZkTimeoutException;
import org.apache.helix.zookeeper.zkclient.metric.ZkClientMonitor;
import org.apache.helix.zookeeper.zkclient.metric.ZkClientPathMonitor;
+import org.apache.jute.BinaryOutputArchive;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.MultiOperationRecord;
import org.apache.zookeeper.Op;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
@@ -68,6 +75,7 @@ import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.proto.RequestHeader;
import org.testng.Assert;
import org.testng.AssertJUnit;
import org.testng.annotations.AfterClass;
@@ -1342,6 +1350,99 @@ public class TestRawZkClient extends ZkTestBase {
Assert.assertFalse(_zkClient.exists(newNode));
}
+ @Test
+ void testGetTransactionSize() throws IOException {
+ // The size is counted without materializing the packet, so it must equal
the length of the
+ // request ZooKeeper actually serializes: the request header followed by
the transaction record.
+ String path = "/bcd/\u00e9";
+ List<Op> ops = Arrays.asList(Op.delete("/a", -1), Op.delete(path, -1),
+ Op.create("/x", new byte[100], ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT),
+ Op.setData("/y", new byte[10], -1));
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ BinaryOutputArchive archive = BinaryOutputArchive.getArchive(bytes);
+ new RequestHeader(0, ZooDefs.OpCode.multi).serialize(archive, "header");
+ new MultiOperationRecord(ops).serialize(archive, "multi");
+
Assert.assertEquals(org.apache.helix.zookeeper.zkclient.ZkClient.getTransactionSize(ops),
+ bytes.size());
+
+ // Each additional delete adds at least the UTF-8 bytes of its path.
+ long oneDelete = org.apache.helix.zookeeper.zkclient.ZkClient
+ .getTransactionSize(Collections.singletonList(Op.delete("/a", -1)));
+ long twoDeletes = org.apache.helix.zookeeper.zkclient.ZkClient
+ .getTransactionSize(Arrays.asList(Op.delete("/a", -1), Op.delete(path,
-1)));
+ Assert.assertTrue(twoDeletes > oneDelete +
path.getBytes(StandardCharsets.UTF_8).length);
+ }
+
+ /*
+ * A multi larger than jute.maxbuffer is never answered by the server, which
just closes the
+ * connection, so ZkClient used to retry it until the operation retry
timeout. multi() and
+ * deleteRecursivelyAtomic() must instead fail fast without changing
anything.
+ */
+ @Test
+ void testDeleteRecursivelyAtomicTooLargeTransaction() {
+ System.out.println("Start test: " + TestHelper.getTestMethodName());
+ String root = "/" + TestHelper.getTestMethodName();
+ List<String> parents = Arrays.asList(root + "/parent1", root + "/parent2");
+ // Children are spread over two parents so that each getChildren()
response stays well below
+ // jute.maxbuffer while the delete transaction for the whole tree (~1.5
MB) exceeds it.
+ int childrenPerParent = 700;
+ String childPrefix = String.join("", Collections.nCopies(1000, "c"));
+ _zkClient.createPersistent(root);
+ for (String parent : parents) {
+ _zkClient.createPersistent(parent);
+ List<Op> creates = new ArrayList<>();
+ for (int i = 0; i < childrenPerParent; i++) {
+ creates.add(Op.create(parent + "/" + childPrefix + i, new byte[0],
+ ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
+ if (creates.size() == 50 || i == childrenPerParent - 1) {
+ _zkClient.multi(creates);
+ creates.clear();
+ }
+ }
+ Assert.assertEquals(_zkClient.getChildren(parent).size(),
childrenPerParent);
+ }
+
+ try {
+ _zkClient.deleteRecursivelyAtomic(root);
+ Assert.fail("Deleting a tree that does not fit in one transaction should
fail");
+ } catch (ZkTransactionTooLargeException expected) {
+ Assert.assertTrue(expected.getMessage().contains("greater than write
size limit"),
+ expected.getMessage());
+ }
+ // The multi-path variant builds the same transaction and must fail the
same way.
+ try {
+ _zkClient.deleteRecursivelyAtomic(parents);
+ Assert.fail("Deleting paths that do not fit in one transaction should
fail");
+ } catch (ZkTransactionTooLargeException expected) {
+ // expected
+ }
+ // So must a plain multi() of the same size.
+ List<Op> deletes = new ArrayList<>();
+ for (String parent : parents) {
+ for (String child : _zkClient.getChildren(parent)) {
+ deletes.add(Op.delete(parent + "/" + child, -1));
+ }
+ }
+ try {
+ _zkClient.multi(deletes);
+ Assert.fail("A multi that does not fit in one packet should fail");
+ } catch (ZkTransactionTooLargeException expected) {
+ // expected
+ }
+
+ // Nothing was deleted.
+ Assert.assertTrue(_zkClient.exists(root));
+ for (String parent : parents) {
+ Assert.assertEquals(_zkClient.getChildren(parent).size(),
childrenPerParent);
+ }
+ // A subtree that does fit still gets deleted atomically, and the
non-atomic delete still
+ // handles the whole tree.
+ _zkClient.deleteRecursivelyAtomic(parents.get(0));
+ Assert.assertFalse(_zkClient.exists(parents.get(0)));
+ _zkClient.deleteRecursively(root);
+ Assert.assertFalse(_zkClient.exists(root));
+ }
+
/*
* 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
