This is an automated email from the ASF dual-hosted git repository.
tflobbe 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 40e554c292e SOLR-16630: TestCoordinatorRole.testNRTRestart waits for
the first add to complete before stopping the PULL node (#5106)
40e554c292e is described below
commit 40e554c292e713c14afea63ff0cfd5d670ef9dba
Author: Nick <[email protected]>
AuthorDate: Tue Oct 27 19:18:19 2026 -0600
SOLR-16630: TestCoordinatorRole.testNRTRestart waits for the first add to
complete before stopping the PULL node (#5106)
Co-authored-by: Tomas Fernandez Lobbe <[email protected]>
---
.../apache/solr/search/TestCoordinatorRole.java | 264 ++++++++++++++++-----
1 file changed, 207 insertions(+), 57 deletions(-)
diff --git a/solr/core/src/test/org/apache/solr/search/TestCoordinatorRole.java
b/solr/core/src/test/org/apache/solr/search/TestCoordinatorRole.java
index 3019cf48fd6..9988fb0efc4 100644
--- a/solr/core/src/test/org/apache/solr/search/TestCoordinatorRole.java
+++ b/solr/core/src/test/org/apache/solr/search/TestCoordinatorRole.java
@@ -27,10 +27,13 @@ import java.util.EnumSet;
import java.util.List;
import java.util.Objects;
import java.util.Random;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import org.apache.solr.client.solrj.SolrClient;
@@ -232,31 +235,58 @@ public class TestCoordinatorRole extends
SolrCloudTestCase {
// Now , kill NRT jetty
JettySolrRunner nrtJettyF = nrtJetty;
JettySolrRunner pullJettyF = pullJetty;
+ // captured while the jetty is running: JettySolrRunner.getNodeName()
returns null once
+ // the jetty is stopped, and the state waits below evaluate it while NRT
is down
+ final String nrtNodeName = nrtJetty.getNodeName();
Random r = random();
- final long establishBaselineMs = r.nextInt(1000);
- final long nrtDowntimeMs = r.nextInt(10000);
- // NOTE: for `pullServiceTimeMs`, it can't be super-short. This is just
to simplify our
- // indexing code,
- // based on the fact that our indexing is based on a PULL-node client.
- final long pullServiceTimeMs = 1000 + (long) r.nextInt(9000);
+ // NOTE: indexing goes through the PULL-node client, so PULL is only
stopped after the add
+ // has completed (see addDone).
+ CountDownLatch addDone = new CountDownLatch(1);
+ // counted down by the main thread right before the phase1 NRT queries,
so the NRT jetty
+ // stops as soon as those queries are under way instead of after a fixed
sleep
+ CountDownLatch stopNrtJetty = new CountDownLatch(1);
+ // counted down by the main thread once it has observed queries failing
over to the PULL
+ // replica, so the NRT jetty restarts once the outage has been exercised
instead of after
+ // a fixed downtime
+ CountDownLatch outageExercised = new CountDownLatch(1);
+ // the coordinator node's own state reader: the state its routing
decisions are made from
+ ZkStateReader zkStateReader =
qaJetty.getCoreContainer().getZkController().getZkStateReader();
Future<?> jettyManipulationFuture =
executor.submit(
() -> {
// we manipulate the jetty instances in a separate thread to
more closely mimic
// the behavior we'd see irl.
try {
- Thread.sleep(establishBaselineMs);
+ if (!stopNrtJetty.await(2, TimeUnit.MINUTES)) {
+ log.warn("phase1 NRT queries did not start in time;
stopping NRT anyway");
+ }
log.info("stopping NRT jetty ...");
nrtJettyF.stop();
log.info("NRT jetty stopped.");
- Thread.sleep(nrtDowntimeMs); // let NRT be down for a while
+ // keep NRT down until the cluster state shows the node as
down and the main
+ // thread has exercised the outage by observing the failover
to PULL, instead
+ // of holding it down for a fixed time
+ try {
+ zkStateReader.waitForState(
+ COLL,
+ 2,
+ TimeUnit.MINUTES,
+ (liveNodes, collectionState) ->
!liveNodes.contains(nrtNodeName));
+ } catch (TimeoutException e) {
+ log.warn("cluster state did not show the NRT node down in
time", e);
+ }
+ if (!outageExercised.await(2, TimeUnit.MINUTES)) {
+ log.warn("failover to PULL was not observed in time;
restarting NRT anyway");
+ }
log.info("restarting NRT jetty ...");
nrtJettyF.start(true);
log.info("NRT jetty restarted.");
// once NRT is back up, we expect PULL to continue serving
until the TTL on ZK
// state used for query request routing has expired (60s).
But here we force a
- // return to NRT by stopping the PULL replica after a brief
delay ...
- Thread.sleep(pullServiceTimeMs);
+ // return to NRT by stopping the PULL replica once the NRT
add has completed ...
+ if (!addDone.await(2, TimeUnit.MINUTES)) {
+ log.warn("NRT add did not succeed in time; stopping PULL
anyway");
+ }
log.info("stopping PULL jetty ...");
pullJettyF.stop();
log.info("PULL jetty stopped.");
@@ -268,31 +298,38 @@ public class TestCoordinatorRole extends
SolrCloudTestCase {
long start = new Date().getTime();
long individualRequestStart = start;
int count = 0;
- while (nrtCore.equals(
- hostCore =
- getHostCoreName(
- COLL,
- qaJettyBase,
- p -> p.add(ShardParams.SHARDS_PREFERENCE,
"replica.type:NRT")))) {
- count++;
- Thread.sleep(100);
+ stopNrtJetty.countDown();
+ // Run the phase1 queries until one is served by PULL. Each retry waits
for the cluster
+ // state to change (the NRT node leaving the live nodes is the change
that flips the
+ // routing) instead of sleeping between requests; the wait itself is
bounded, so a quiet
+ // cluster still retries at the old pace, and the whole phase is bounded
as well.
+ long phase1Deadline =
+ new Date().getTime() + TimeUnit.MILLISECONDS.convert(2,
TimeUnit.MINUTES);
+ hostCore = null;
+ do {
individualRequestStart = new Date().getTime();
- }
+ hostCore =
+ getHostCoreName(
+ COLL, qaJettyBase, p -> p.add(ShardParams.SHARDS_PREFERENCE,
"replica.type:NRT"));
+ count++;
+ if (!pullCore.equals(hostCore)) {
+ waitForClusterStateChange(zkStateReader, COLL);
+ }
+ } while (!pullCore.equals(hostCore) && new Date().getTime() <
phase1Deadline);
+ // the outage has been exercised (or the phase bound ran out, which the
assert below
+ // reports): the NRT jetty may restart now
+ outageExercised.countDown();
long now = new Date().getTime();
log.info(
- "phase1 NRT queries count={}, overall_duration={},
baseline_expected_overall_duration={}, switch-to-pull_duration={}",
+ "phase1 NRT queries count={}, overall_duration={},
switch-to-pull_duration={}",
count,
now - start,
- establishBaselineMs,
now - individualRequestStart);
- // default tolerance of 500ms below should suffice. Failover to PULL for
this case should be
- // very fast, because our QA-based client already knows both replicas
are active, the index
- // is stable, so the moment the client finds NRT is down it should be
able to failover
- // immediately and transparently to PULL.
+ // Failover to PULL for this case should be very fast, because our
QA-based client already
+ // knows both replicas are active, the index is stable, so the moment
the client finds NRT
+ // is down it should be able to failover immediately and transparently
to PULL.
assertEquals(
- "when we break out of the NRT query loop, should be b/c routed to
PULL",
- pullCore,
- hostCore);
+ "once the NRT node is down, queries should be routed to PULL",
pullCore, hostCore);
SolrInputDocument d = new SolrInputDocument();
d.addField("id", "345");
d.addField("desc_s", "Another Document");
@@ -301,53 +338,82 @@ public class TestCoordinatorRole extends
SolrCloudTestCase {
count = 0;
start = new Date().getTime();
individualRequestStart = start;
+ long addDeadline = start + TimeUnit.MILLISECONDS.convert(2,
TimeUnit.MINUTES);
for (; ; ) {
try {
client.add(COLL, d);
client.commit(COLL);
+ addDone.countDown();
break;
} catch (SolrException ex) {
- // we expect these until nrtJetty is back up.
+ // we expect these until nrtJetty is back up. Wait for the cluster
state to show the
+ // NRT replica active again, or at least to move on, before
retrying, instead of
+ // sleeping between attempts.
count++;
- Thread.sleep(100);
+ if (new Date().getTime() > addDeadline) {
+ throw ex;
+ }
+ DocCollection stateBefore = zkStateReader.getCollection(COLL);
+ int znodeVersionBefore = stateBefore == null ? -1 :
stateBefore.getZNodeVersion();
+ int childVersionBefore = stateBefore == null ? -1 :
stateBefore.getChildNodesVersion();
+ try {
+ zkStateReader.waitForState(
+ COLL,
+ 30,
+ TimeUnit.SECONDS,
+ (liveNodes, collectionState) ->
+ collectionState != null
+ && ((liveNodes.contains(nrtNodeName)
+ &&
collectionState.getReplicasOnNode(nrtNodeName).stream()
+ .allMatch(rep -> rep.getState() ==
Replica.State.ACTIVE))
+ || collectionState.getZNodeVersion() !=
znodeVersionBefore
+ || collectionState.getChildNodesVersion() !=
childVersionBefore));
+ } catch (TimeoutException e) {
+ throw ex;
+ }
}
individualRequestStart = new Date().getTime();
}
now = new Date().getTime();
log.info(
- "successfully added another doc; duration: {}, overall_duration={},
baseline_expected_overall_duration={}, exception_count={}",
+ "successfully added another doc; duration: {}, overall_duration={},
exception_count={}",
now - individualRequestStart,
now - start,
- nrtDowntimeMs,
count);
// NRT replica is back up, registered as available with Zk, and
availability info has been
// pulled down by our PULL-replica-based `client`, forwarded indexing
command to NRT,
- // index/commit completed. All of this accounts for the 3000ms tolerance
allowed for below.
- // This is not a strict value, and if it causes failures regularly we
should feel free to
- // increase the tolerance; but it's meant to provide a stable baseline
from which to detect
- // regressions.
+ // index/commit completed.
count = 0;
start = new Date().getTime();
individualRequestStart = start;
- while (pullCore.equals(
- hostCore =
- getHostCoreName(
- COLL,
- qaJettyBase,
- p -> {
- p.set(CommonParams.Q, "id:345");
- p.add(ShardParams.SHARDS_PREFERENCE, "replica.type:NRT");
- }))) {
- count++;
- Thread.sleep(100);
+ // The added doc is served by the NRT replica again once the manipulator
has stopped
+ // PULL (it does so right after the add) and the coordinator's routing
flips back. Each
+ // retry waits for the cluster state to change instead of sleeping
between requests; the
+ // wait itself is bounded, so a quiet cluster still retries at the old
pace, and the
+ // whole phase is bounded as well.
+ long phase3Deadline =
+ new Date().getTime() + TimeUnit.MILLISECONDS.convert(2,
TimeUnit.MINUTES);
+ hostCore = null;
+ do {
individualRequestStart = new Date().getTime();
- }
+ hostCore =
+ getHostCoreName(
+ COLL,
+ qaJettyBase,
+ p -> {
+ p.set(CommonParams.Q, "id:345");
+ p.add(ShardParams.SHARDS_PREFERENCE, "replica.type:NRT");
+ });
+ count++;
+ if (!nrtCore.equals(hostCore)) {
+ waitForClusterStateChange(zkStateReader, COLL);
+ }
+ } while (!nrtCore.equals(hostCore) && new Date().getTime() <
phase3Deadline);
now = new Date().getTime();
log.info(
- "query retries between NRT index-ready and query-ready: {};
overall_duration={}; baseline_expected_overall_duration={};
failover-request_duration={}",
+ "queries between NRT index-ready and query-ready: {};
overall_duration={}; failover-request_duration={}",
count,
now - start,
- pullServiceTimeMs,
now - individualRequestStart);
assertEquals(nrtCore, hostCore);
// allow any exceptions to propagate
@@ -369,6 +435,11 @@ public class TestCoordinatorRole extends SolrCloudTestCase
{
new Date().getTime() + TimeUnit.MILLISECONDS.convert(runMinutes,
TimeUnit.MINUTES);
JettySolrRunner[] jettys = new JettySolrRunner[] {pullJettyF, nrtJettyF};
Random threadRandom = new Random(r.nextInt());
+ // the toggle thread publishes one latch per phase here; the main thread
counts it down
+ // once per query it completes, so each phase lasts for a counted number
of queries
+ // instead of a duration slept through
+ AtomicReference<CountDownLatch> phaseLatch = new AtomicReference<>();
+ boolean failoverObserved = false;
Future<Integer> f =
executor.submit(
() -> {
@@ -377,14 +448,19 @@ public class TestCoordinatorRole extends
SolrCloudTestCase {
int idx = iteration++ % jettys.length;
JettySolrRunner toManipulate = jettys[idx];
try {
- int serveTogetherTime = threadRandom.nextInt(7000);
- int downTime = threadRandom.nextInt(7000);
- log.info("serving together for {}ms", serveTogetherTime);
- Thread.sleep(serveTogetherTime);
+ // The old phase duration was random from 0 to 7000ms,
around 0 to 70 queries
+ // at the old ~100ms query pace. These phase lengths are
now expressed as
+ // completed query counts.
+ int serveTogetherQueries = threadRandom.nextInt(70);
+ int downQueries = threadRandom.nextInt(70);
+ log.info("serving together for {} queries",
serveTogetherQueries);
+ if (!awaitQueryCount(phaseLatch, serveTogetherQueries,
finishTimeMs, done)) {
+ break;
+ }
log.info("stopping {} ...", idx);
toManipulate.stop();
log.info("stopped {}.", idx);
- Thread.sleep(downTime);
+ awaitQueryCount(phaseLatch, downQueries, finishTimeMs,
done);
log.info("restarting {} ...", idx);
toManipulate.start(true);
log.info("restarted {}.", idx);
@@ -403,6 +479,7 @@ public class TestCoordinatorRole extends SolrCloudTestCase {
});
count = 0;
start = new Date().getTime();
+ long toggleEndMs;
try {
do {
if (pullCore.equals(
@@ -413,10 +490,18 @@ public class TestCoordinatorRole extends
SolrCloudTestCase {
p.set(CommonParams.Q, "id:345");
p.add(ShardParams.SHARDS_PREFERENCE, "replica.type:NRT");
}))) {
+ failoverObserved = true;
done.set(true);
}
count++;
- Thread.sleep(100);
+ CountDownLatch latch = phaseLatch.get();
+ if (latch != null) {
+ latch.countDown();
+ } else {
+ // no phase is counting queries right now (a jetty is stopping or
starting); pace
+ // this loop on cluster state changes instead of sleeping between
requests
+ waitForClusterStateChange(zkStateReader, COLL);
+ }
} while (!done.get());
} finally {
final String result;
@@ -428,6 +513,7 @@ public class TestCoordinatorRole extends SolrCloudTestCase {
result = "Failure";
}
Integer toggleCount = f.get();
+ toggleEndMs = new Date().getTime();
long secondsDuration =
TimeUnit.SECONDS.convert(new Date().getTime() - start,
TimeUnit.MILLISECONDS);
log.info(
@@ -437,6 +523,11 @@ public class TestCoordinatorRole extends SolrCloudTestCase
{
toggleCount,
count);
}
+ if (failoverObserved) {
+ assertTrue(
+ "toggle thread ran until the end of the toggle window after the
failover was observed",
+ toggleEndMs < finishTimeMs);
+ }
} finally {
try {
ExecutorUtil.shutdownAndAwaitTermination(executor);
@@ -462,6 +553,8 @@ public class TestCoordinatorRole extends SolrCloudTestCase {
p.accept(q);
SolrDocumentList docs = null;
SolrClient solrClient = cluster.getJetty(qaNode).getSolrClient();
+ ZkStateReader zkStateReader =
+
cluster.getJetty(qaNode).getCoreContainer().getZkController().getZkStateReader();
for (int i = 0; i < 100; i++) {
try {
QueryResponse queryResponse = solrClient.query(COLL, q);
@@ -483,13 +576,70 @@ public class TestCoordinatorRole extends
SolrCloudTestCase {
fail("initial error time threshold exceeded");
}
}
- Thread.sleep(100);
+ // pace the retry on cluster state changes instead of sleeping: the
transient errors and
+ // empty results above clear once the state moves on, and the wait
returns as soon as it
+ // does (bounded, so a quiet cluster still retries at the old pace)
+ waitForClusterStateChange(zkStateReader, COLL);
}
assertTrue(found);
return (String) docs.get(0).getFieldValue("_core_");
}
+ /**
+ * Blocks until the collection's cluster state changes, or the bounded wait
runs out. Used to pace
+ * retries on the event they are waiting for instead of on a fixed sleep; a
timeout is not an
+ * error, the caller simply retries.
+ */
+ private void waitForClusterStateChange(ZkStateReader zkStateReader, String
collection)
+ throws Exception {
+ DocCollection stateBefore = zkStateReader.getCollection(collection);
+ int znodeVersionBefore = stateBefore == null ? -1 :
stateBefore.getZNodeVersion();
+ int childVersionBefore = stateBefore == null ? -1 :
stateBefore.getChildNodesVersion();
+ try {
+ zkStateReader.waitForState(
+ collection,
+ 100,
+ TimeUnit.MILLISECONDS,
+ (liveNodes, collectionState) ->
+ collectionState == null
+ || collectionState.getZNodeVersion() != znodeVersionBefore
+ || collectionState.getChildNodesVersion() !=
childVersionBefore);
+ } catch (TimeoutException e) {
+ // no state change within the bound; the caller retries anyway
+ }
+ }
+
+ /**
+ * Publishes a latch that the main thread counts down once per completed
query, and blocks until
+ * {@code queries} queries have completed, the toggle window ends, or {@code
done} is set (the
+ * main thread stops querying once done, so the count would never be
reached). Returns false when
+ * the count was not reached.
+ */
+ private boolean awaitQueryCount(
+ AtomicReference<CountDownLatch> phaseLatch,
+ int queries,
+ long finishTimeMs,
+ AtomicBoolean done)
+ throws InterruptedException {
+ CountDownLatch latch = new CountDownLatch(queries);
+ phaseLatch.set(latch);
+ try {
+ while (!done.get()) {
+ long remainingMs = finishTimeMs - new Date().getTime();
+ if (remainingMs <= 0) {
+ return false;
+ }
+ if (latch.await(Math.min(remainingMs, 100), TimeUnit.MILLISECONDS)) {
+ return true;
+ }
+ }
+ return false;
+ } finally {
+ phaseLatch.compareAndSet(latch, null);
+ }
+ }
+
public void testConcurrentAccess() throws Exception {
final int DATA_NODE_COUNT = 2;
final int COORDINATOR_NODE_COUNT = 4;
