You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@solr.apache.org by no...@apache.org on 2022/11/09 02:57:17 UTC
[solr] branch jira/solr16456_1 created (now 3ef938e1e2a)
This is an automated email from the ASF dual-hosted git repository.
noble pushed a change to branch jira/solr16456_1
in repository https://gitbox.apache.org/repos/asf/solr.git
at 3ef938e1e2a more ZkNodeProps eliminated
This branch includes the following new commits:
new 3ef938e1e2a more ZkNodeProps eliminated
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[solr] 01/01: more ZkNodeProps eliminated
Posted by no...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
noble pushed a commit to branch jira/solr16456_1
in repository https://gitbox.apache.org/repos/asf/solr.git
commit 3ef938e1e2a4fea8434566b09783c48556fbca0c
Author: Noble Paul <no...@gmail.com>
AuthorDate: Wed Nov 9 13:56:34 2022 +1100
more ZkNodeProps eliminated
---
.../src/java/org/apache/solr/cloud/Overseer.java | 6 +--
.../org/apache/solr/cloud/ZkDistributedQueue.java | 4 --
.../org/apache/solr/cloud/DeleteShardTest.java | 41 +++++-----------
.../apache/solr/cloud/OverseerTaskQueueTest.java | 20 ++++----
.../test/org/apache/solr/cloud/OverseerTest.java | 54 +++++++++++-----------
.../solr/cloud/TestRandomRequestDistribution.java | 3 +-
.../org/apache/solr/cloud/ZkControllerTest.java | 33 +++++++------
.../solr/client/solrj/cloud/DistributedQueue.java | 10 +++-
.../src/java/org/apache/solr/common/MapWriter.java | 8 ++++
9 files changed, 86 insertions(+), 93 deletions(-)
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 6736affd526..c7546a491b3 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Overseer.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
@@ -1220,8 +1220,8 @@ public class Overseer implements SolrCloseable {
*/
public void sendQuitToOverseer(String overseerId) throws KeeperException, InterruptedException {
getOverseerQuitNotificationQueue()
- .offer(
- new ZkNodeProps(
- Overseer.QUEUE_OPERATION, OverseerAction.QUIT.toLower(), ID, overseerId));
+ .offer(ew -> ew.put(Overseer.QUEUE_OPERATION, OverseerAction.QUIT.toLower())
+ .put(ID, overseerId));
+
}
}
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java b/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java
index 1332661019b..0c684e13a70 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java
@@ -309,10 +309,6 @@ public class ZkDistributedQueue implements DistributedQueue {
}
}
- public void offer(MapWriter mw) throws KeeperException, InterruptedException {
- offer(Utils.toJSON(mw));
- }
-
/**
* Inserts data into queue. If there are no other queue consumers, the offered element will be
* immediately visible when this method returns.
diff --git a/solr/core/src/test/org/apache/solr/cloud/DeleteShardTest.java b/solr/core/src/test/org/apache/solr/cloud/DeleteShardTest.java
index b763b58bad6..1eaa956e208 100644
--- a/solr/core/src/test/org/apache/solr/cloud/DeleteShardTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/DeleteShardTest.java
@@ -17,20 +17,18 @@
package org.apache.solr.cloud;
import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.CoreStatus;
import org.apache.solr.cloud.overseer.OverseerAction;
+import org.apache.solr.common.MapWriter;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.Slice.State;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.util.Utils;
import org.apache.solr.util.FileUtils;
import org.junit.After;
import org.junit.Before;
@@ -66,9 +64,7 @@ public class DeleteShardTest extends SolrCloudTestCase {
// Can't delete an ACTIVE shard
expectThrows(
Exception.class,
- () -> {
- CollectionAdminRequest.deleteShard(collection, "shard1").process(cluster.getSolrClient());
- });
+ () -> CollectionAdminRequest.deleteShard(collection, "shard1").process(cluster.getSolrClient()));
setSliceState(collection, "shard1", Slice.State.INACTIVE);
@@ -77,9 +73,7 @@ public class DeleteShardTest extends SolrCloudTestCase {
waitForState(
"Expected 'shard1' to be removed",
collection,
- (n, c) -> {
- return c.getSlice("shard1") == null;
- });
+ (n, c) -> c.getSlice("shard1") == null);
// Can delete a shard under construction
setSliceState(collection, "shard2", Slice.State.CONSTRUCTION);
@@ -87,27 +81,22 @@ public class DeleteShardTest extends SolrCloudTestCase {
waitForState(
"Expected 'shard2' to be removed",
collection,
- (n, c) -> {
- return c.getSlice("shard2") == null;
- });
+ (n, c) -> c.getSlice("shard2") == null);
}
protected void setSliceState(String collection, String slice, State state) throws Exception {
// TODO can this be encapsulated better somewhere?
- Map<String, Object> propMap = new HashMap<>();
- propMap.put(Overseer.QUEUE_OPERATION, OverseerAction.UPDATESHARDSTATE.toLower());
- propMap.put(slice, state.toString());
- propMap.put(ZkStateReader.COLLECTION_PROP, collection);
- ZkNodeProps m = new ZkNodeProps(propMap);
-
+ MapWriter m = ew -> ew.put(Overseer.QUEUE_OPERATION, OverseerAction.UPDATESHARDSTATE.toLower())
+ .put(slice, state.toString())
+ .put(ZkStateReader.COLLECTION_PROP, collection);
final Overseer overseer = cluster.getOpenOverseer();
if (overseer.getDistributedClusterStateUpdater().isDistributedStateUpdate()) {
overseer
.getDistributedClusterStateUpdater()
.doSingleStateUpdate(
DistributedClusterStateUpdater.MutatingCommand.SliceUpdateShardState,
- m,
+ new ZkNodeProps(m),
cluster.getOpenOverseer().getSolrCloudManager(),
cluster.getOpenOverseer().getZkStateReader());
} else {
@@ -118,15 +107,13 @@ public class DeleteShardTest extends SolrCloudTestCase {
.getZkController()
.getOverseer()
.getStateUpdateQueue();
- inQueue.offer(Utils.toJSON(m));
+ inQueue.offer(m);
}
waitForState(
"Expected shard " + slice + " to be in state " + state,
collection,
- (n, c) -> {
- return c.getSlice(slice).getState() == state;
- });
+ (n, c) -> c.getSlice(slice).getState() == state);
}
@Test
@@ -155,9 +142,7 @@ public class DeleteShardTest extends SolrCloudTestCase {
waitForState(
"Expected 'a' to be removed",
collection,
- (n, c) -> {
- return c.getSlice("a") == null;
- });
+ (n, c) -> c.getSlice("a") == null);
assertEquals(2, getCollectionState(collection).getActiveSlices().size());
assertFalse(
@@ -176,9 +161,7 @@ public class DeleteShardTest extends SolrCloudTestCase {
waitForState(
"Expected 'b' to be removed",
collection,
- (n, c) -> {
- return c.getSlice("b") == null;
- });
+ (n, c) -> c.getSlice("b") == null);
assertEquals(1, getCollectionState(collection).getActiveSlices().size());
assertTrue(
diff --git a/solr/core/src/test/org/apache/solr/cloud/OverseerTaskQueueTest.java b/solr/core/src/test/org/apache/solr/cloud/OverseerTaskQueueTest.java
index 64f1c5d4c29..c6676844b2d 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OverseerTaskQueueTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OverseerTaskQueueTest.java
@@ -17,10 +17,10 @@
package org.apache.solr.cloud;
import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.solr.cloud.api.collections.CollectionHandlingUtils;
+import org.apache.solr.common.MapWriter;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.CollectionAdminParams;
import org.apache.solr.common.params.CommonAdminParams;
@@ -47,13 +47,12 @@ public class OverseerTaskQueueTest extends DistributedQueueTest {
// Basic ops
// Put an expected Overseer task onto the queue
- final Map<String, Object> props = new HashMap<>();
- props.put(CommonParams.NAME, "coll1");
- props.put(CollectionAdminParams.COLL_CONF, "myconf");
- props.put(CollectionHandlingUtils.NUM_SLICES, 1);
- props.put(ZkStateReader.REPLICATION_FACTOR, 3);
- props.put(CommonAdminParams.ASYNC, requestId);
- tq.offer(Utils.toJSON(props));
+ MapWriter props = ew -> ew.put(CommonParams.NAME, "coll1")
+ .put(CollectionAdminParams.COLL_CONF, "myconf")
+ .put(CollectionHandlingUtils.NUM_SLICES, 1)
+ .put(ZkStateReader.REPLICATION_FACTOR, 3)
+ .put(CommonAdminParams.ASYNC, requestId);
+ tq.offer(props);
assertTrue(
"Task queue should contain task with requestid " + requestId,
@@ -71,8 +70,9 @@ public class OverseerTaskQueueTest extends DistributedQueueTest {
// containsTaskWithRequestId runs while the response is still in the queue.
String watchID = tq.createResponseNode();
String requestId2 = "baz";
- props.put(CommonAdminParams.ASYNC, requestId2);
- tq.createRequestNode(Utils.toJSON(props), watchID);
+
+ tq.createRequestNode(Utils.toJSON(props.append(ew -> ew.put(CommonAdminParams.ASYNC, requestId2))),
+ watchID);
// Set a SolrResponse as the response node by removing the QueueEvent, as done in
// OverseerTaskProcessor
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 0f67e6209fb..5df90af9028 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
@@ -58,6 +58,7 @@ import org.apache.solr.cloud.overseer.NodeMutator;
import org.apache.solr.cloud.overseer.OverseerAction;
import org.apache.solr.cloud.overseer.ZkWriteCommand;
import org.apache.solr.common.AlreadyClosedException;
+import org.apache.solr.common.MapWriter;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
@@ -72,7 +73,6 @@ import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.IOUtils;
import org.apache.solr.common.util.SolrNamedThreadFactory;
import org.apache.solr.common.util.TimeSource;
-import org.apache.solr.common.util.Utils;
import org.apache.solr.core.CloudConfig;
import org.apache.solr.core.ClusterSingletons;
import org.apache.solr.core.CoreContainer;
@@ -208,7 +208,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
// Look for "new Overseer" calls in this class.
assertFalse(overseer.getDistributedClusterStateUpdater().isDistributedStateUpdate());
ZkDistributedQueue q = overseer.getStateUpdateQueue();
- q.offer(Utils.toJSON(m));
+ q.offer(m);
}
public String publishState(
@@ -242,7 +242,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
collection);
assertFalse(overseer.getDistributedClusterStateUpdater().isDistributedStateUpdate());
ZkDistributedQueue q = overseer.getStateUpdateQueue();
- q.offer(Utils.toJSON(m));
+ q.offer(m);
return null;
} else {
ZkNodeProps m =
@@ -266,7 +266,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.NUM_SHARDS_PROP,
Integer.toString(numShards));
ZkDistributedQueue q = overseer.getStateUpdateQueue();
- q.offer(Utils.toJSON(m));
+ q.offer(m);
}
if (startElection && collection.length() > 0) {
@@ -460,7 +460,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
"createNodeSet",
"");
ZkDistributedQueue q = getOverseerZero().getStateUpdateQueue();
- q.offer(Utils.toJSON(m));
+ q.offer(m);
}
@Test
@@ -729,7 +729,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkDistributedQueue q = getOverseerZero().getStateUpdateQueue();
- q.offer(Utils.toJSON(m));
+ q.offer(m);
verifyReplicaStatus(
reader, commands.get(0).name, "shard1", "core_node1", Replica.State.DOWN);
@@ -813,7 +813,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.STATE_PROP,
Replica.State.RECOVERING.toString());
- q.offer(Utils.toJSON(m));
+ q.offer(m);
waitForCollections(reader, COLLECTION);
verifyReplicaStatus(reader, "collection1", "shard1", "core_node1", Replica.State.RECOVERING);
@@ -836,7 +836,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.STATE_PROP,
Replica.State.ACTIVE.toString());
- q.offer(Utils.toJSON(m));
+ q.offer(m);
verifyReplicaStatus(reader, "collection1", "shard1", "core_node1", Replica.State.ACTIVE);
@@ -1138,11 +1138,11 @@ public class OverseerTest extends SolrTestCaseJ4 {
"createNodeSet",
"");
ZkDistributedQueue workQueue = Overseer.getInternalWorkQueue(zkClient, new Stats());
- workQueue.offer(Utils.toJSON(badMessage));
+ workQueue.offer(badMessage);
overseerClient = electNewOverseer(server.getZkAddress());
ZkDistributedQueue q = getOpenOverseer().getStateUpdateQueue();
- q.offer(Utils.toJSON(badMessage));
+ q.offer(badMessage);
TimeOut timeOut = new TimeOut(10, TimeUnit.SECONDS, TimeSource.NANO_TIME);
while (!timeOut.hasTimedOut()) {
@@ -1226,7 +1226,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
"createNodeSet",
"");
ZkDistributedQueue q = getOpenOverseer().getStateUpdateQueue();
- q.offer(Utils.toJSON(m));
+ q.offer(m);
break;
} catch (SolrException | KeeperException | AlreadyClosedException e) {
log.error("error updating state", e);
@@ -1462,7 +1462,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.REPLICATION_FACTOR,
"1");
ZkDistributedQueue q = getOverseerZero().getStateUpdateQueue();
- q.offer(Utils.toJSON(m));
+ q.offer(m);
}
for (int i = 0, j = 0, k = 0; i < MAX_STATE_CHANGES; i++, j++, k++) {
@@ -1485,7 +1485,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.NUM_SHARDS_PROP,
"1");
ZkDistributedQueue q = getOverseerZero().getStateUpdateQueue();
- q.offer(Utils.toJSON(m));
+ q.offer(m);
if (j >= MAX_COLLECTIONS - 1) j = 0;
if (k >= MAX_CORES - 1) k = 0;
if (i > 0 && i % 100 == 0) log.info("Published {} items", i);
@@ -1585,7 +1585,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
"1",
"createNodeSet",
"");
- queue.offer(Utils.toJSON(m));
+ queue.offer(m);
m =
new ZkNodeProps(
Overseer.QUEUE_OPERATION,
@@ -1602,7 +1602,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
"",
ZkStateReader.STATE_PROP,
Replica.State.RECOVERING.toString());
- queue.offer(Utils.toJSON(m));
+ queue.offer(m);
m =
new ZkNodeProps(
Overseer.QUEUE_OPERATION,
@@ -1619,7 +1619,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
"",
ZkStateReader.STATE_PROP,
Replica.State.RECOVERING.toString());
- queue.offer(Utils.toJSON(m));
+ queue.offer(m);
overseerClient = electNewOverseer(server.getZkAddress());
@@ -1641,7 +1641,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
"",
ZkStateReader.STATE_PROP,
Replica.State.RECOVERING.toString());
- queue.offer(Utils.toJSON(m));
+ queue.offer(m);
reader.waitForState(
COLLECTION,
@@ -1705,7 +1705,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.STATE_PROP,
Replica.State.DOWN.toString());
- q.offer(Utils.toJSON(m));
+ q.offer(m);
waitForCollections(reader, "c1");
verifyReplicaStatus(reader, "c1", "shard1", "core_node1", Replica.State.DOWN);
@@ -1727,7 +1727,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.STATE_PROP,
Replica.State.RECOVERING.toString());
- q.offer(Utils.toJSON(m));
+ q.offer(m);
m =
new ZkNodeProps(
@@ -1746,7 +1746,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.STATE_PROP,
Replica.State.ACTIVE.toString());
- q.offer(Utils.toJSON(m));
+ q.offer(m);
final String testCollectionName = "test";
zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/" + testCollectionName, true);
@@ -1760,7 +1760,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
"1",
ZkStateReader.REPLICATION_FACTOR,
"1");
- q.offer(Utils.toJSON(m));
+ q.offer(m);
// Wait for the overseer to create state.json for the collection
waitForCollections(reader, testCollectionName);
@@ -1781,7 +1781,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
"x",
ZkStateReader.REPLICATION_FACTOR,
"1");
- q.offer(Utils.toJSON(m));
+ q.offer(m);
m =
new ZkNodeProps(
@@ -1799,7 +1799,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
"core1",
ZkStateReader.STATE_PROP,
Replica.State.DOWN.toString());
- q.offer(Utils.toJSON(m));
+ q.offer(m);
// Verify replica creation worked ok in spite of external update of state.json (although in
// theory such updates do not happen unless an old overseer is still updating ZK after a new
@@ -1977,7 +1977,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
numShards.toString(),
ZkStateReader.REPLICATION_FACTOR,
"1");
- q.offer(Utils.toJSON(m));
+ q.offer(m);
}
waitForCollections(zkStateReader, COLLECTION);
@@ -2004,7 +2004,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.STATE_PROP,
Replica.State.RECOVERING.toString());
- q.offer(Utils.toJSON(m));
+ q.offer(m);
}
}
// verify recovering
@@ -2037,7 +2037,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.STATE_PROP,
Replica.State.ACTIVE.toString());
- q.offer(Utils.toJSON(m));
+ q.offer(m);
}
}
// verify active
@@ -2062,7 +2062,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.CORE_NODE_NAME_PROP,
"core_node" + N);
- q.offer(Utils.toJSON(m));
+ q.offer(m);
{
String shard = "shard" + ss;
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestRandomRequestDistribution.java b/solr/core/src/test/org/apache/solr/cloud/TestRandomRequestDistribution.java
index 949f4b29e94..2d093cd7e16 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestRandomRequestDistribution.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestRandomRequestDistribution.java
@@ -39,7 +39,6 @@ import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.util.Utils;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.SolrCore;
import org.apache.solr.metrics.SolrMetricManager;
@@ -204,7 +203,7 @@ public class TestRandomRequestDistribution extends AbstractFullDistribZkTestBase
overseer.getZkStateReader());
} else {
ZkDistributedQueue q = overseer.getStateUpdateQueue();
- q.offer(Utils.toJSON(m));
+ q.offer(m);
}
verifyReplicaStatus(
diff --git a/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java b/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java
index 0ddf70c360f..06c93c11684 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java
@@ -24,12 +24,12 @@ import static org.mockito.Mockito.mock;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.common.MapWriter;
import org.apache.solr.common.cloud.ClusterProperties;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.SolrZkClient;
@@ -285,13 +285,13 @@ public class ZkControllerTest extends SolrTestCaseJ4 {
zkController.getOverseerJobQueue().offer(Utils.toJSON(m));
}
- HashMap<String, Object> propMap = new HashMap<>();
- propMap.put(Overseer.QUEUE_OPERATION, ADDREPLICA.toLower());
- propMap.put(COLLECTION_PROP, collectionName);
- propMap.put(SHARD_ID_PROP, "shard1");
- propMap.put(ZkStateReader.NODE_NAME_PROP, "non_existent_host:1_");
- propMap.put(ZkStateReader.CORE_NAME_PROP, collectionName);
- propMap.put(ZkStateReader.STATE_PROP, "active");
+ MapWriter propMap = ew -> ew.put(Overseer.QUEUE_OPERATION, ADDREPLICA.toLower())
+ .put(COLLECTION_PROP, collectionName)
+ .put(SHARD_ID_PROP, "shard1")
+ .put(ZkStateReader.NODE_NAME_PROP, "non_existent_host:1_")
+ .put(ZkStateReader.CORE_NAME_PROP, collectionName)
+ .put(ZkStateReader.STATE_PROP, "active");
+
if (zkController.getDistributedClusterStateUpdater().isDistributedStateUpdate()) {
zkController
.getDistributedClusterStateUpdater()
@@ -301,16 +301,15 @@ public class ZkControllerTest extends SolrTestCaseJ4 {
zkController.getSolrCloudManager(),
zkController.getZkStateReader());
} else {
- zkController.getOverseerJobQueue().offer(Utils.toJSON(propMap));
+ zkController.getOverseerJobQueue().offer(propMap);
}
- propMap = new HashMap<>();
- propMap.put(Overseer.QUEUE_OPERATION, ADDREPLICA.toLower());
- propMap.put(COLLECTION_PROP, collectionName);
- propMap.put(SHARD_ID_PROP, "shard1");
- propMap.put(ZkStateReader.NODE_NAME_PROP, "non_existent_host:2_");
- propMap.put(ZkStateReader.CORE_NAME_PROP, collectionName);
- propMap.put(ZkStateReader.STATE_PROP, "down");
+ propMap = ew -> ew.put(Overseer.QUEUE_OPERATION, ADDREPLICA.toLower())
+ .put(COLLECTION_PROP, collectionName)
+ .put(SHARD_ID_PROP, "shard1")
+ .put(ZkStateReader.NODE_NAME_PROP, "non_existent_host:2_")
+ .put(ZkStateReader.CORE_NAME_PROP, collectionName)
+ .put(ZkStateReader.STATE_PROP, "down");
if (zkController.getDistributedClusterStateUpdater().isDistributedStateUpdate()) {
zkController
.getDistributedClusterStateUpdater()
@@ -320,7 +319,7 @@ public class ZkControllerTest extends SolrTestCaseJ4 {
zkController.getSolrCloudManager(),
zkController.getZkStateReader());
} else {
- zkController.getOverseerJobQueue().offer(Utils.toJSON(propMap));
+ zkController.getOverseerJobQueue().offer(propMap);
}
zkController.getZkStateReader().forciblyRefreshAllClusterStateSlow();
diff --git a/solr/solrj-zookeeper/src/java/org/apache/solr/client/solrj/cloud/DistributedQueue.java b/solr/solrj-zookeeper/src/java/org/apache/solr/client/solrj/cloud/DistributedQueue.java
index 097f013a3c0..a47641bb9af 100644
--- a/solr/solrj-zookeeper/src/java/org/apache/solr/client/solrj/cloud/DistributedQueue.java
+++ b/solr/solrj-zookeeper/src/java/org/apache/solr/client/solrj/cloud/DistributedQueue.java
@@ -19,7 +19,11 @@ package org.apache.solr.client.solrj.cloud;
import java.util.Collection;
import java.util.Map;
import java.util.function.Predicate;
+
+import org.apache.solr.common.MapWriter;
import org.apache.solr.common.util.Pair;
+import org.apache.solr.common.util.Utils;
+import org.apache.zookeeper.KeeperException;
/** Distributed queue component. Methods largely follow those in {@link java.util.Queue}. */
public interface DistributedQueue {
@@ -35,7 +39,11 @@ public interface DistributedQueue {
byte[] take() throws Exception;
- void offer(byte[] data) throws Exception;
+ void offer(byte[] data) throws KeeperException, InterruptedException;
+
+ default void offer(MapWriter mw) throws KeeperException, InterruptedException {
+ offer(Utils.toJSON(mw));
+ }
/** Retrieve statistics about the queue size, operations and their timings. */
Map<String, Object> getStats();
diff --git a/solr/solrj/src/java/org/apache/solr/common/MapWriter.java b/solr/solrj/src/java/org/apache/solr/common/MapWriter.java
index b9005057c38..62c456d38fb 100644
--- a/solr/solrj/src/java/org/apache/solr/common/MapWriter.java
+++ b/solr/solrj/src/java/org/apache/solr/common/MapWriter.java
@@ -83,6 +83,14 @@ public interface MapWriter extends MapSerializable, NavigableObject {
void writeMap(EntryWriter ew) throws IOException;
+ default MapWriter append(MapWriter another) {
+ MapWriter m = this;
+ return ew -> {
+ m.writeMap(ew);
+ another.writeMap(ew);
+ };
+ }
+
/**
* An interface to push one entry at a time to the output. The order of the keys is not defined,
* but we assume they are distinct -- don't call {@code put} more than once for the same key.