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 3019cf48fd60..9988fb0efc4c 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.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 void testNRTRestart() throws Exception { // 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 void testNRTRestart() throws Exception { 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 void testNRTRestart() throws Exception { 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 void testNRTRestart() throws Exception { 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 phaseLatch = new AtomicReference<>(); + boolean failoverObserved = false; Future f = executor.submit( () -> { @@ -377,14 +448,19 @@ public void testNRTRestart() throws Exception { 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 void testNRTRestart() throws Exception { }); count = 0; start = new Date().getTime(); + long toggleEndMs; try { do { if (pullCore.equals( @@ -413,10 +490,18 @@ public void testNRTRestart() throws Exception { 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 void testNRTRestart() throws Exception { 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 void testNRTRestart() throws Exception { 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 @@ private String getHostCoreName(String COLL, String qaNode, Consumer p 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 @@ private String getHostCoreName(String COLL, String qaNode, Consumer p 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 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;