SOLR-6213: Run leader election rejoins on their own thread

changelog/unreleased/SOLR-6213-rejoin-thread.yml

@@ -0,0 +1,12 @@
+title: >
+  Leader election rejoins after a failed sync or leader publish now run on their own thread instead
+  of nesting on the caller's thread, so repeated rejoins no longer grow the stack toward
+  StackOverflowError.
+  Closing and cancelling a replica's election waits for an in-flight rejoin to finish so it
+  cannot leave a stale election participant behind.
+type: fixed
+authors:
+  - name: Nick Shanin
+links:
+  - name: SOLR-6213
+    url: https://issues.apache.org/jira/browse/SOLR-6213

solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java

@@ -18,7 +18,9 @@
 
 import java.lang.invoke.MethodHandles;
 import java.util.EnumSet;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import org.apache.lucene.search.MatchAllDocsQuery;
 import org.apache.solr.cloud.api.collections.CollectionHandlingUtils;
@@ -33,6 +35,8 @@
 import org.apache.solr.common.cloud.ZkCoreNodeProps;
 import org.apache.solr.common.cloud.ZkNodeProps;
 import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
 import org.apache.solr.core.CoreContainer;
 import org.apache.solr.core.SolrCore;
 import org.apache.solr.logging.MDCLoggingContext;
@@ -56,6 +60,13 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
 
   private volatile boolean isClosed = false;
 
+  // Cancellation must see the election node created by an in-flight rejoin.
+  private final Object rejoinLock = new Object();
+
+  private final ExecutorService rejoinExecutor =
+      ExecutorUtil.newMDCAwareSingleLazyThreadExecutor(
+          new SolrNamedThreadFactory("leaderRejoin"), 10, TimeUnit.SECONDS);
+
   public ShardLeaderElectionContext(
       LeaderElector leaderElector,
       final String shardId,
@@ -74,6 +85,7 @@ public ShardLeaderElectionContext(
   public void close() {
     super.close();
     this.isClosed = true;
+    rejoinExecutor.shutdown();
     syncStrategy.close();
   }
 
@@ -87,14 +99,16 @@ private boolean shouldAbort() {
 
   @Override
   public void cancelElection() throws InterruptedException, KeeperException {
-    String coreName = leaderProps.getStr(ZkStateReader.CORE_NAME_PROP);
-    try (SolrCore core = cc.getCore(coreName)) {
-      if (core != null) {
-        core.getCoreDescriptor().getCloudDescriptor().setLeader(false);
+    synchronized (rejoinLock) {
+      String coreName = leaderProps.getStr(ZkStateReader.CORE_NAME_PROP);
+      try (SolrCore core = cc.getCore(coreName)) {
+        if (core != null) {
+          core.getCoreDescriptor().getCloudDescriptor().setLeader(false);
+        }
       }
-    }
 
-    super.cancelElection();
+      super.cancelElection();
+    }
   }
 
   @Override
@@ -185,7 +199,7 @@ void runLeaderProcess(boolean weAreReplacement) throws KeeperException, Interrup
         if (zkShardTerms.registered(coreNodeName) && !zkShardTerms.canBecomeLeader(coreNodeName)) {
           if (!waitForEligibleBecomeLeaderAfterTimeout(
               zkShardTerms, coreNodeName, leaderVoteWait)) {
-            rejoinLeaderElection(core);
+            rejoinLeaderElection();
             return;
           } else {
             // only log an error if this replica win the election
@@ -242,7 +256,7 @@ void runLeaderProcess(boolean weAreReplacement) throws KeeperException, Interrup
           }
         }
         if (!success) {
-          rejoinLeaderElection(core);
+          rejoinLeaderElection();
           return;
         }
       }
@@ -315,7 +329,7 @@ void runLeaderProcess(boolean weAreReplacement) throws KeeperException, Interrup
 
             // we could not publish ourselves as leader - try and rejoin election
             try {
-              rejoinLeaderElection(core);
+              rejoinLeaderElection();
             } catch (SessionExpiredException exc) {
               throw new SolrException(
                   ErrorCode.SERVER_ERROR,
@@ -480,9 +494,9 @@ private boolean waitForReplicasToComeUp(int timeoutms) throws InterruptedExcepti
     return false;
   }
 
-  private void rejoinLeaderElection(SolrCore core) throws InterruptedException, KeeperException {
+  void rejoinLeaderElection() throws InterruptedException, KeeperException {
     // remove our ephemeral and re join the election
-    if (cc.isShutDown()) {
+    if (cc.isShutDown() || isClosed) {
       log.debug("Not rejoining election because CoreContainer is closed");
       return;
     }
@@ -491,6 +505,31 @@ private void rejoinLeaderElection(SolrCore core) throws InterruptedException, Ke
 
     cancelElection();
 
-    leaderElector.joinElection(this, true);
+    try {
+      rejoinExecutor.execute(this::joinElectionAfterRejoin);
+    } catch (RejectedExecutionException e) {
+      log.debug("Not rejoining election because the context is closed");
+    }
+  }
+
+  private void joinElectionAfterRejoin() {
+    synchronized (rejoinLock) {
+      if (isClosed) {
+        return;
+      }
+      try {
+        try {
+          leaderElector.joinElection(this, true);
+        } finally {
+          if (isClosed) {
+            cancelElection();
+          }
+        }
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      } catch (KeeperException | RuntimeException e) {
+        log.error("Failed to rejoin the election", e);
+      }
+    }
   }
 }

solr/core/src/java/org/apache/solr/cloud/ZkController.java

@@ -2043,6 +2043,7 @@ public void unregister(String coreName, CoreDescriptor cd, boolean removeCoreFro
       ElectionContext context = electionContexts.remove(new ContextKey(collection, coreNodeName));
 
       if (context != null) {
+        context.close();
         context.cancelElection();
       }
     }

solr/core/src/test/org/apache/solr/cloud/ShardLeaderElectionRejoinTest.java

@@ -0,0 +1,238 @@
+/*
+ * 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 static org.apache.solr.SolrTestCaseJ4.assumeWorkingMockito;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+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.AtomicInteger;
+import org.apache.solr.SolrTestCase;
+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.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.core.CoreContainer;
+import org.apache.zookeeper.KeeperException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/** Exercises rejoin scheduling and cancellation with real ZooKeeper election participants. */
+public class ShardLeaderElectionRejoinTest extends SolrTestCase {
+  private static final int WAIT_SECONDS = 30;
+  private static final int REJOINS = 20;
+  private static final String ELECTION_PATH = "/collections/rejoin/leader_elect/shard1/election";
+
+  private ZkTestServer server;
+  private SolrZkClient zkClient;
+  private ZkStateReader zkStateReader;
+  private CoreContainer cc;
+  private ZkController zkController;
+
+  @Before
+  public void startZooKeeper() throws Exception {
+    assumeWorkingMockito();
+    server = new ZkTestServer(createTempDir("rejoin-zk"));
+    server.run();
+    zkClient =
+        new SolrZkClient.Builder()
+            .withUrl(server.getZkAddress())
+            .withTimeout(WAIT_SECONDS, TimeUnit.SECONDS)
+            .build();
+    zkStateReader = new ZkStateReader(zkClient);
+    zkClient.makePath("/collections/rejoin", true);
+    cc = mock(CoreContainer.class, RETURNS_DEEP_STUBS);
+    when(cc.getCore("core")).thenReturn(null);
+    zkController = mock(ZkController.class);
+    when(zkController.getZkClient()).thenReturn(zkClient);
+    when(zkController.getZkStateReader()).thenReturn(zkStateReader);
+  }
+
+  @After
+  public void stopZooKeeper() throws Exception {
+    try {
+      if (zkStateReader != null) {
+        zkStateReader.close();
+      }
+    } finally {
+      try {
+        if (zkClient != null) {
+          zkClient.close();
+        }
+      } finally {
+        if (server != null) {
+          server.shutdown();
+        }
+      }
+    }
+  }
+
+  @Test
+  public void testRepeatedRejoinDoesNotGrowStack() throws Exception {
+    List<Integer> depths = new CopyOnWriteArrayList<>();
+    CountDownLatch finished = new CountDownLatch(1);
+    LeaderElector elector =
+        new LeaderElector(zkClient) {
+          @Override
+          protected void runIamLeaderProcess(ElectionContext context, boolean replacement)
+              throws KeeperException, InterruptedException {
+            depths.add(Thread.currentThread().getStackTrace().length);
+            if (depths.size() < REJOINS) {
+              ((ShardLeaderElectionContext) context).rejoinLeaderElection();
+            } else {
+              finished.countDown();
+            }
+          }
+        };
+    ShardLeaderElectionContext context = newContext(elector);
+    try {
+      context.rejoinLeaderElection();
+      assertTrue("Rejoins did not finish", finished.await(WAIT_SECONDS, TimeUnit.SECONDS));
+      assertEquals(REJOINS, depths.size());
+      int min = depths.stream().mapToInt(Integer::intValue).min().orElseThrow();
+      int max = depths.stream().mapToInt(Integer::intValue).max().orElseThrow();
+      assertTrue("Stack grew by " + (max - min) + " frames", max - min < REJOINS);
+    } finally {
+      context.close();
+      context.cancelElection();
+    }
+    assertTrue(zkClient.getChildren(ELECTION_PATH, null).isEmpty());
+  }
+
+  @Test
+  public void testCancellationWaitsForRunningRejoin() throws Exception {
+    CountDownLatch joining = new CountDownLatch(1);
+    CountDownLatch allowJoin = new CountDownLatch(1);
+    CountDownLatch joined = new CountDownLatch(1);
+    CountDownLatch cancelling = new CountDownLatch(1);
+    LeaderElector elector =
+        new LeaderElector(zkClient) {
+          @Override
+          public int joinElection(ElectionContext context, boolean replacement)
+              throws KeeperException, InterruptedException {
+            joining.countDown();
+            if (!allowJoin.await(WAIT_SECONDS, TimeUnit.SECONDS)) {
+              throw new IllegalStateException("Timed out waiting to create election node");
+            }
+            try {
+              return super.joinElection(context, replacement);
+            } finally {
+              joined.countDown();
+            }
+          }
+
+          @Override
+          protected void runIamLeaderProcess(ElectionContext context, boolean replacement) {}
+        };
+    ShardLeaderElectionContext context = newContext(elector);
+    ExecutorService canceller =
+        ExecutorUtil.newMDCAwareSingleThreadExecutor(new SolrNamedThreadFactory("cancelElection"));
+    try {
+      context.rejoinLeaderElection();
+      assertTrue("Rejoin did not start", joining.await(WAIT_SECONDS, TimeUnit.SECONDS));
+      context.close();
+      Future<?> cancelled =
+          canceller.submit(
+              () -> {
+                cancelling.countDown();
+                context.cancelElection();
+                return null;
+              });
+      assertTrue(cancelling.await(WAIT_SECONDS, TimeUnit.SECONDS));
+      try {
+        cancelled.get(1, TimeUnit.SECONDS);
+        fail("Cancellation returned before the running rejoin could create its node");
+      } catch (TimeoutException expected) {
+        // Cancellation must remain blocked until the running join releases the lifecycle lock.
+      } finally {
+        allowJoin.countDown();
+      }
+      cancelled.get(WAIT_SECONDS, TimeUnit.SECONDS);
+      assertTrue(joined.await(WAIT_SECONDS, TimeUnit.SECONDS));
+      assertTrue(
+          "Closed context left an election node",
+          zkClient.getChildren(ELECTION_PATH, null).isEmpty());
+    } finally {
+      allowJoin.countDown();
+      context.close();
+      ExecutorUtil.shutdownAndAwaitTermination(canceller);
+      if (joining.getCount() == 0) {
+        assertTrue(joined.await(WAIT_SECONDS, TimeUnit.SECONDS));
+      }
+      context.cancelElection();
+    }
+
+    AtomicInteger elected = new AtomicInteger();
+    LeaderElector next =
+        new LeaderElector(zkClient) {
+          @Override
+          protected void runIamLeaderProcess(ElectionContext context, boolean replacement) {
+            elected.incrementAndGet();
+          }
+        };
+    ShardLeaderElectionContext nextContext = newContext(next);
+    try {
+      next.joinElection(nextContext, true);
+      assertEquals("A stale participant blocked the next candidate", 1, elected.get());
+    } finally {
+      nextContext.close();
+      nextContext.cancelElection();
+    }
+  }
+
+  @Test
+  public void testClosedContextDoesNotRejoin() throws Exception {
+    AtomicInteger joins = new AtomicInteger();
+    LeaderElector elector =
+        new LeaderElector(zkClient) {
+          @Override
+          public int joinElection(ElectionContext context, boolean replacement) {
+            joins.incrementAndGet();
+            return 0;
+          }
+        };
+    ShardLeaderElectionContext context = newContext(elector);
+    context.close();
+    context.rejoinLeaderElection();
+    assertEquals(0, joins.get());
+    assertTrue(zkClient.getChildren(ELECTION_PATH, null).isEmpty());
+  }
+
+  private ShardLeaderElectionContext newContext(LeaderElector elector) throws Exception {
+    ShardLeaderElectionContext context =
+        new ShardLeaderElectionContext(
+            elector,
+            "shard1",
+            "rejoin",
+            "core_node1",
+            new ZkNodeProps(ZkStateReader.CORE_NAME_PROP, "core"),
+            zkController,
+            cc);
+    elector.setup(context);
+    return context;
+  }
+}