This is an automated email from the ASF dual-hosted git repository. dsmiley pushed a commit to branch branch_10x in repository https://gitbox.apache.org/repos/asf/solr.git
commit 3a5f14d54970a94a307057bb81eddeab73884c52 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) (cherry picked from commit ad55842492387d0f0bb313c5619774cd2ac47510) --- .../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 112b301b62c..051bdf753af 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 -> {
