This is an automated email from the ASF dual-hosted git repository.
sigram pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/solr.git
The following commit(s) were added to refs/heads/main by this push:
new ad558424923 SOLR-17894: Support collection state.json compression when
Overseer is disabled. (#4905)
ad558424923 is described below
commit ad55842492387d0f0bb313c5619774cd2ac47510
Author: Andrzej Białecki <[email protected]>
AuthorDate: Sat Sep 12 10:29:42 2026 +0200
SOLR-17894: Support collection state.json compression when Overseer is
disabled. (#4905)
---
.../solr-17894-distrib-update-compression.yml | 7 +
.../solr/cloud/DistributedClusterStateUpdater.java | 74 +++++--
.../src/java/org/apache/solr/cloud/Overseer.java | 11 +-
.../java/org/apache/solr/cloud/ZkController.java | 12 +-
.../cloud/DistributedClusterStateUpdaterTest.java | 229 +++++++++++++++++++++
.../test/org/apache/solr/cloud/OverseerTest.java | 2 +-
6 files changed, 310 insertions(+), 25 deletions(-)
diff --git a/changelog/unreleased/solr-17894-distrib-update-compression.yml
b/changelog/unreleased/solr-17894-distrib-update-compression.yml
new file mode 100644
index 00000000000..8b5d9fd2534
--- /dev/null
+++ b/changelog/unreleased/solr-17894-distrib-update-compression.yml
@@ -0,0 +1,7 @@
+title: Support collection state.json compression when Overseer is disabled.
+type: changed
+authors:
+ - name: Andrzej Bialecki
+links:
+ - name: SOLR-17894
+ url: https://issues.apache.org/jira/browse/SOLR-17894
diff --git
a/solr/core/src/java/org/apache/solr/cloud/DistributedClusterStateUpdater.java
b/solr/core/src/java/org/apache/solr/cloud/DistributedClusterStateUpdater.java
index 9a8e249ca3c..478b0fdc9ad 100644
---
a/solr/core/src/java/org/apache/solr/cloud/DistributedClusterStateUpdater.java
+++
b/solr/core/src/java/org/apache/solr/cloud/DistributedClusterStateUpdater.java
@@ -45,6 +45,7 @@ import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.util.Compressor;
import org.apache.solr.common.util.Pair;
import org.apache.solr.common.util.Utils;
import org.apache.zookeeper.CreateMode;
@@ -67,6 +68,9 @@ public class DistributedClusterStateUpdater {
*/
private final boolean useDistributedStateUpdate;
+ private final int minStateByteLenForCompression;
+ private final Compressor compressor;
+
/**
* Builds an instance with the specified behavior regarding distribution of
state updates,
* allowing to know distributed updates are not enabled (parameter {@code
@@ -76,9 +80,16 @@ public class DistributedClusterStateUpdater {
* @param useDistributedStateUpdate when this parameter is {@code false},
only method expected to
* ever be called on this instance is {@link #isDistributedStateUpdate},
and it will return
* {@code false}.
+ * @param minStateByteLenForCompression the minimum size of a state.json
file that should be
+ * compressed before being written to Zookeeper.
+ * @param compressor the compressor to use for compressing state.json files
before writing them to
+ * Zookeeper.
*/
- public DistributedClusterStateUpdater(boolean useDistributedStateUpdate) {
+ public DistributedClusterStateUpdater(
+ boolean useDistributedStateUpdate, int minStateByteLenForCompression,
Compressor compressor) {
this.useDistributedStateUpdate = useDistributedStateUpdate;
+ this.minStateByteLenForCompression = minStateByteLenForCompression;
+ this.compressor = compressor;
}
/**
@@ -93,7 +104,8 @@ public class DistributedClusterStateUpdater {
throw new IllegalStateException(
"Not expecting to create instances of StateChangeRecorder when not
using distributed state update");
}
- return new StateChangeRecorder(collectionName, isCollectionCreation);
+ return new StateChangeRecorder(
+ collectionName, isCollectionCreation, minStateByteLenForCompression,
compressor);
}
/** Syntactic sugar to allow a single change to the cluster state to be made
in a single call. */
@@ -109,7 +121,11 @@ public class DistributedClusterStateUpdater {
}
String collectionName = command.getCollectionName(message);
final StateChangeRecorder scr =
- new StateChangeRecorder(collectionName,
command.isCollectionCreation());
+ new StateChangeRecorder(
+ collectionName,
+ command.isCollectionCreation(),
+ minStateByteLenForCompression,
+ compressor);
scr.record(command, message);
scr.executeStateUpdates(scm, zkStateReader);
}
@@ -119,7 +135,8 @@ public class DistributedClusterStateUpdater {
throw new IllegalStateException(
"Not expecting to execute executeNodeDownStateUpdate when not using
distributed state update");
}
- CollectionNodeDownChangeCalculator.executeNodeDownStateUpdate(nodeName,
zkStateReader);
+ CollectionNodeDownChangeCalculator.executeNodeDownStateUpdate(
+ nodeName, zkStateReader, minStateByteLenForCompression, compressor);
}
/**
@@ -349,16 +366,29 @@ public class DistributedClusterStateUpdater {
private final ZkStateReader zkStateReader;
private final StateChangeCalculator updater;
-
- static void applyUpdate(ZkStateReader zkStateReader, StateChangeCalculator
updater)
+ private final int minStateByteLenForCompression;
+ private final Compressor compressor;
+
+ static void applyUpdate(
+ ZkStateReader zkStateReader,
+ StateChangeCalculator updater,
+ int minStateByteLenForCompression,
+ Compressor compressor)
throws KeeperException, InterruptedException {
- ZkUpdateApplicator zua = new ZkUpdateApplicator(zkStateReader, updater);
+ ZkUpdateApplicator zua =
+ new ZkUpdateApplicator(zkStateReader, updater,
minStateByteLenForCompression, compressor);
zua.applyUpdate();
}
- private ZkUpdateApplicator(ZkStateReader zkStateReader,
StateChangeCalculator updater) {
+ private ZkUpdateApplicator(
+ ZkStateReader zkStateReader,
+ StateChangeCalculator updater,
+ int minStateByteLenForCompression,
+ Compressor compressor) {
this.zkStateReader = zkStateReader;
this.updater = updater;
+ this.minStateByteLenForCompression = minStateByteLenForCompression;
+ this.compressor = compressor;
}
/**
@@ -546,6 +576,11 @@ public class DistributedClusterStateUpdater {
// Collection update or creation
DocCollection collection =
updatedState.getCollection(updater.getCollectionName());
byte[] stateJson = Utils.toJSON(Map.of(updater.getCollectionName(),
collection));
+ if (minStateByteLenForCompression > -1
+ && stateJson.length > minStateByteLenForCompression) {
+ // When compressing state.json, we expect at least a 10:1
compression ratio.
+ stateJson = compressor.compressBytes(stateJson, stateJson.length /
10);
+ }
if (updater.isCollectionCreation()) {
// The state.json file does not exist yet (more precisely it is
assumed not to exist)
@@ -627,7 +662,14 @@ public class DistributedClusterStateUpdater {
*/
boolean creationCommandRecorded = false;
- private StateChangeRecorder(String collectionName, boolean
isCollectionCreation) {
+ final int minStateByteLenForCompression;
+ final Compressor compressor;
+
+ private StateChangeRecorder(
+ String collectionName,
+ boolean isCollectionCreation,
+ int minStateByteLenForCompression,
+ Compressor compressor) {
if (collectionName == null) {
final String err =
"Internal bug. collectionName=null (isCollectionCreation=" +
isCollectionCreation + ")";
@@ -637,6 +679,8 @@ public class DistributedClusterStateUpdater {
mutations = new ArrayList<>();
this.collectionName = collectionName;
this.isCollectionCreation = isCollectionCreation;
+ this.minStateByteLenForCompression = minStateByteLenForCompression;
+ this.compressor = compressor;
}
/**
@@ -825,7 +869,8 @@ public class DistributedClusterStateUpdater {
RecordedMutationsPlayer mutationPlayer =
new RecordedMutationsPlayer(scm, collectionName,
isCollectionCreation, mutations);
- ZkUpdateApplicator.applyUpdate(zkStateReader, mutationPlayer);
+ ZkUpdateApplicator.applyUpdate(
+ zkStateReader, mutationPlayer, minStateByteLenForCompression,
compressor);
// TODO update stats here for the various commands executed successfully
or not?
// This would replace the stats about cluster state updates that the
Collection API currently
@@ -859,7 +904,11 @@ public class DistributedClusterStateUpdater {
* Entry point to mark all replicas of all collections present on a single
node as being DOWN
* (because the node is down)
*/
- public static void executeNodeDownStateUpdate(String nodeName,
ZkStateReader zkStateReader) {
+ public static void executeNodeDownStateUpdate(
+ String nodeName,
+ ZkStateReader zkStateReader,
+ int minStateByteLenForCompression,
+ Compressor compressor) {
// This code does a version of what NodeMutator.downNode() is doing. We
can't assume we have a
// cache of the collections, so we're going to read all of them from ZK,
fetch the state.json
// for each and if it has any replicas on the failed node, do an update
(conditional of
@@ -886,7 +935,8 @@ public class DistributedClusterStateUpdater {
for (String collectionName : collectionNames) {
CollectionNodeDownChangeCalculator collectionUpdater =
new CollectionNodeDownChangeCalculator(collectionName, nodeName);
- ZkUpdateApplicator.applyUpdate(zkStateReader, collectionUpdater);
+ ZkUpdateApplicator.applyUpdate(
+ zkStateReader, collectionUpdater, minStateByteLenForCompression,
compressor);
}
} catch (Exception e) {
if (e instanceof InterruptedException) {
diff --git a/solr/core/src/java/org/apache/solr/cloud/Overseer.java
b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
index 9d35afb7161..019dd57b3e1 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Overseer.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
@@ -52,9 +52,7 @@ import org.apache.solr.common.util.Compressor;
import org.apache.solr.common.util.IOUtils;
import org.apache.solr.common.util.ObjectReleaseTracker;
import org.apache.solr.common.util.Pair;
-import org.apache.solr.common.util.StrUtils;
import org.apache.solr.common.util.Utils;
-import org.apache.solr.common.util.ZLibCompressor;
import org.apache.solr.core.CloudConfig;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.SolrInfoBean;
@@ -734,14 +732,7 @@ public class Overseer implements SolrCloseable {
createOverseerNode(reader.getZkClient());
// launch cluster state updater thread
ThreadGroup tg = new ThreadGroup("Overseer state updater.");
- String stateCompressionProviderClass = config.getStateCompressorClass();
- Compressor compressor =
- StrUtils.isNullOrEmpty(stateCompressionProviderClass)
- ? new ZLibCompressor()
- : zkController
- .getCoreContainer()
- .getResourceLoader()
- .newInstance(stateCompressionProviderClass, Compressor.class);
+ Compressor compressor = zkController.getCompressor();
updaterThread =
new OverseerThread(
tg,
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
index 808bd0d24f9..8a260c1d85d 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -225,6 +225,8 @@ public class ZkController implements Closeable {
private final CloudConfig cloudConfig;
private final NodesSysPropsCacher sysPropsCacher;
+ private final Compressor compressor;
+
private final DistributedClusterStateUpdater distributedClusterStateUpdater;
private final Optional<DistributedCollectionConfigSetCommandRunner>
distributedCommandRunner;
@@ -324,7 +326,7 @@ public class ZkController implements Closeable {
addOnReconnectListener(getConfigDirListener());
- final var compressor =
+ compressor =
loadPluginOrDefault(
Compressor.class, cloudConfig.getStateCompressorClass(), new
ZLibCompressor());
@@ -382,7 +384,9 @@ public class ZkController implements Closeable {
"The Overseer is disabled. Cluster commands & state updates will
happen on any/all nodes.");
}
// These "distributed" things replace the Overseer when that's disabled
- this.distributedClusterStateUpdater = new
DistributedClusterStateUpdater(!overseerEnabled);
+ this.distributedClusterStateUpdater =
+ new DistributedClusterStateUpdater(
+ !overseerEnabled, cloudConfig.getMinStateByteLenForCompression(),
compressor);
this.distributedCommandRunner =
!overseerEnabled
? Optional.of(new DistributedCollectionConfigSetCommandRunner(cc,
zkClient))
@@ -401,6 +405,10 @@ public class ZkController implements Closeable {
assert ObjectReleaseTracker.track(this);
}
+ public Compressor getCompressor() {
+ return compressor;
+ }
+
private void onDisconnect(boolean sessionExpired) {
try {
overseer.close();
diff --git
a/solr/core/src/test/org/apache/solr/cloud/DistributedClusterStateUpdaterTest.java
b/solr/core/src/test/org/apache/solr/cloud/DistributedClusterStateUpdaterTest.java
new file mode 100644
index 00000000000..39b44220064
--- /dev/null
+++
b/solr/core/src/test/org/apache/solr/cloud/DistributedClusterStateUpdaterTest.java
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.cloud;
+
+import java.util.Map;
+import org.apache.lucene.util.IOUtils;
+import org.apache.solr.SolrTestCase;
+import org.apache.solr.client.solrj.cloud.SolrCloudManager;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.impl.SolrClientCloudManager;
+import org.apache.solr.client.solrj.impl.ZkClientClusterStateProvider;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.SolrZkClient;
+import org.apache.solr.common.cloud.ZkNodeProps;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.util.Compressor;
+import org.apache.solr.common.util.Utils;
+import org.apache.solr.common.util.ZLibCompressor;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+/**
+ * Verifies that {@link DistributedClusterStateUpdater} (used when the
Overseer is disabled)
+ * compresses {@code state.json} above the configured size threshold, the same
way {@link
+ * org.apache.solr.cloud.overseer.ZkStateWriter} does for the Overseer path.
See {@code
+ * ZkStateWriterTest#testSingleExternalCollectionCompressedState} for the
Overseer equivalent.
+ */
+public class DistributedClusterStateUpdaterTest extends SolrTestCase {
+
+ private static final int MIN_STATE_BYTE_LEN_FOR_COMPRESSION = 10_000;
+
+ private static ZkTestServer server;
+ private static SolrZkClient zkClient;
+
+ @BeforeClass
+ public static void startZkServer() throws Exception {
+ server = new
ZkTestServer(createTempDir("DistributedClusterStateUpdaterTest"));
+ server.run();
+ zkClient = new
SolrZkClient.Builder().withUrl(server.getZkAddress()).build();
+ ZkController.createClusterZkNodes(zkClient);
+ }
+
+ @AfterClass
+ public static void stopZkServer() throws Exception {
+ IOUtils.closeWhileHandlingException(zkClient);
+ if (server != null) {
+ server.shutdown();
+ }
+ }
+
+ public void testDistributedStateUpdateCompressesLargeStateJson() throws
Exception {
+ CloudSolrClient cloudSolrClient = null;
+
+ // default compressor impl
+ Compressor compressor = new ZLibCompressor();
+
+ try (ZkStateReader reader = new ZkStateReader(zkClient)) {
+ reader.createClusterStateWatchersAndUpdate();
+
+ cloudSolrClient =
+ new CloudSolrClient.Builder(new
ZkClientClusterStateProvider(reader)).build();
+ SolrCloudManager scm = new SolrClientCloudManager(cloudSolrClient, null);
+
+ DistributedClusterStateUpdater updater =
+ new DistributedClusterStateUpdater(true,
MIN_STATE_BYTE_LEN_FOR_COMPRESSION, compressor);
+
+ // small collection: state.json stays under the threshold and is not
compressed.
+ String smallCollection = "small";
+ zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/" +
smallCollection, true);
+ updater.doSingleStateUpdate(
+
DistributedClusterStateUpdater.MutatingCommand.ClusterCreateCollection,
+ new ZkNodeProps(CommonParams.NAME, smallCollection,
ZkStateReader.NUM_SHARDS_PROP, "1"),
+ scm,
+ reader);
+
+ byte[] smallStateJson =
+ zkClient
+ .getCuratorFramework()
+ .getData()
+ // make sure the Curator doesn't decompress it automatically
+ .undecompressed()
+ .forPath(ZkStateReader.COLLECTIONS_ZKNODE + "/" +
smallCollection + "/state.json");
+
+ assertFalse(
+ "small state.json should not be compressed",
+ compressor.isCompressedBytes(smallStateJson));
+ Map<?, ?> smallMap = (Map<?, ?>) Utils.fromJSON(smallStateJson);
+ assertNotNull(smallMap.get(smallCollection));
+
+ // large collection: enough replicas added in one batch to push
state.json past the
+ // compression threshold.
+ String bigCollection = "big";
+ zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/" +
bigCollection, true);
+ updater.doSingleStateUpdate(
+
DistributedClusterStateUpdater.MutatingCommand.ClusterCreateCollection,
+ new ZkNodeProps(CommonParams.NAME, bigCollection,
ZkStateReader.NUM_SHARDS_PROP, "1"),
+ scm,
+ reader);
+
+ DistributedClusterStateUpdater.StateChangeRecorder recorder =
+ updater.createStateChangeRecorder(bigCollection, false);
+ for (int i = 0; i < 300; i++) {
+ recorder.record(
+ DistributedClusterStateUpdater.MutatingCommand.SliceAddReplica,
+ new ZkNodeProps(
+ ZkStateReader.COLLECTION_PROP,
+ bigCollection,
+ ZkStateReader.SHARD_ID_PROP,
+ "shard1",
+ ZkStateReader.CORE_NODE_NAME_PROP,
+ "core_node" + i,
+ ZkStateReader.CORE_NAME_PROP,
+ "core_node" + i,
+ ZkStateReader.NODE_NAME_PROP,
+ "127.0.0.1:8983_solr",
+ ZkStateReader.STATE_PROP,
+ Replica.State.ACTIVE.toString(),
+ ZkStateReader.REPLICA_TYPE,
+ Replica.Type.NRT.toString()));
+ }
+ recorder.executeStateUpdates(scm, reader);
+
+ byte[] bigStateJsonRaw =
+ zkClient
+ .getCuratorFramework()
+ .getData()
+ .undecompressed()
+ .forPath(ZkStateReader.COLLECTIONS_ZKNODE + "/" + bigCollection
+ "/state.json");
+ assertTrue(
+ "big state.json should have been compressed by
DistributedClusterStateUpdater",
+ compressor.isCompressedBytes(bigStateJsonRaw));
+
+ Map<?, ?> bigMap = (Map<?, ?>)
Utils.fromJSON(compressor.decompressBytes(bigStateJsonRaw));
+ assertNotNull(bigMap.get(bigCollection));
+
+ // reading through the normal (automatically decompressing) path yields
the same, uncompressed
+ // JSON.
+ byte[] bigStateJsonDecoded =
+ zkClient.getData(
+ ZkStateReader.COLLECTIONS_ZKNODE + "/" + bigCollection +
"/state.json", null, null);
+ assertFalse(compressor.isCompressedBytes(bigStateJsonDecoded));
+ bigMap = (Map<?, ?>) Utils.fromJSON(bigStateJsonDecoded);
+ assertNotNull(bigMap.get(bigCollection));
+ } finally {
+ IOUtils.closeWhileHandlingException(cloudSolrClient);
+ }
+ }
+
+ public void testDistributedStateUpdateDisabledCompression() throws Exception
{
+ CloudSolrClient cloudSolrClient = null;
+
+ Compressor compressor = new ZLibCompressor();
+
+ try (ZkStateReader reader = new ZkStateReader(zkClient)) {
+ reader.createClusterStateWatchersAndUpdate();
+
+ cloudSolrClient =
+ new CloudSolrClient.Builder(new
ZkClientClusterStateProvider(reader)).build();
+ SolrCloudManager scm = new SolrClientCloudManager(cloudSolrClient, null);
+
+ // negative threshold: compression must never kick in, no matter the
state.json size.
+ DistributedClusterStateUpdater updater =
+ new DistributedClusterStateUpdater(true, -1, compressor);
+
+ String bigCollection = "bigNoCompression";
+ zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/" +
bigCollection, true);
+ updater.doSingleStateUpdate(
+
DistributedClusterStateUpdater.MutatingCommand.ClusterCreateCollection,
+ new ZkNodeProps(CommonParams.NAME, bigCollection,
ZkStateReader.NUM_SHARDS_PROP, "1"),
+ scm,
+ reader);
+
+ DistributedClusterStateUpdater.StateChangeRecorder recorder =
+ updater.createStateChangeRecorder(bigCollection, false);
+ for (int i = 0; i < 300; i++) {
+ recorder.record(
+ DistributedClusterStateUpdater.MutatingCommand.SliceAddReplica,
+ new ZkNodeProps(
+ ZkStateReader.COLLECTION_PROP,
+ bigCollection,
+ ZkStateReader.SHARD_ID_PROP,
+ "shard1",
+ ZkStateReader.CORE_NODE_NAME_PROP,
+ "core_node" + i,
+ ZkStateReader.CORE_NAME_PROP,
+ "core_node" + i,
+ ZkStateReader.NODE_NAME_PROP,
+ "127.0.0.1:8983_solr",
+ ZkStateReader.STATE_PROP,
+ Replica.State.ACTIVE.toString(),
+ ZkStateReader.REPLICA_TYPE,
+ Replica.Type.NRT.toString()));
+ }
+ recorder.executeStateUpdates(scm, reader);
+
+ byte[] bigStateJsonRaw =
+ zkClient
+ .getCuratorFramework()
+ .getData()
+ // make sure the Curator doesn't decompress it automatically
+ .undecompressed()
+ .forPath(ZkStateReader.COLLECTIONS_ZKNODE + "/" + bigCollection
+ "/state.json");
+ assertFalse(
+ "state.json must stay uncompressed when
minStateByteLenForCompression is negative,"
+ + " regardless of size",
+ compressor.isCompressedBytes(bigStateJsonRaw));
+
+ Map<?, ?> bigMap = (Map<?, ?>) Utils.fromJSON(bigStateJsonRaw);
+ assertNotNull(bigMap.get(bigCollection));
+ } finally {
+ IOUtils.closeWhileHandlingException(cloudSolrClient);
+ }
+ }
+}
diff --git a/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
b/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
index 15ae9733f7a..c29244590fe 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
@@ -1776,7 +1776,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
when(zkController.getZkClient()).thenReturn(zkClient);
when(zkController.getZkStateReader()).thenReturn(reader);
when(zkController.getDistributedClusterStateUpdater())
- .thenReturn(new DistributedClusterStateUpdater(false));
+ .thenReturn(new DistributedClusterStateUpdater(false, -1, null));
// primitive support for CC.runAsync
doAnswer(
invocable -> {
