diff --git a/changelog/unreleased/SOLR-6213-rejoin-thread.yml b/changelog/unreleased/SOLR-6213-rejoin-thread.yml new file mode 100644 index 000000000000..69e0ffa105d0 --- /dev/null +++ b/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 diff --git a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java index 7512867a794d..f98335216182 100644 --- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java +++ b/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); + } + } } } 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 5db9f8d7256a..19037e39832e 100644 --- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java +++ b/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(); } } diff --git a/solr/core/src/test/org/apache/solr/cloud/ShardLeaderElectionRejoinTest.java b/solr/core/src/test/org/apache/solr/cloud/ShardLeaderElectionRejoinTest.java new file mode 100644 index 000000000000..f5f4dbfb9c7d --- /dev/null +++ b/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 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; + } +}