diff --git a/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java b/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java index f864d919c937..7df36a9c459f 100644 --- a/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java +++ b/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java @@ -18,8 +18,13 @@ import java.io.Closeable; import java.lang.invoke.MethodHandles; +import java.util.List; +import org.apache.curator.framework.api.transaction.CuratorTransactionResult; +import org.apache.curator.framework.api.transaction.OperationType; import org.apache.solr.common.cloud.SolrZkClient; +import org.apache.solr.common.cloud.ZkMaintenanceUtils; import org.apache.solr.common.cloud.ZkNodeProps; +import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.NoNodeException; import org.slf4j.Logger; @@ -31,9 +36,21 @@ public abstract class ElectionContext implements Closeable { final ZkNodeProps leaderProps; final String id; final String leaderPath; + + /** Parent of {@link #leaderPath}; derived once since {@link #leaderPath} is final. */ + final String leaderParentPath; + volatile String leaderSeqPath; private SolrZkClient zkClient; + /** + * Version of {@link #leaderPath}'s parent captured when this context registered as leader (see + * {@link #registerLeaderNode}); {@link #deleteLeaderNode} uses it on cancel to remove only our + * own registration (ABA-safe). Null until/after registration. Thread-safety is provided by each + * subclass's own election lock — there is no base-level lock guarding this field. + */ + protected Integer leaderZkNodeParentVersion; + public ElectionContext( final String coreNodeName, final String electionPath, @@ -44,6 +61,7 @@ public ElectionContext( this.id = coreNodeName; this.electionPath = electionPath; this.leaderPath = leaderPath; + this.leaderParentPath = ZkMaintenanceUtils.getZkParent(leaderPath); this.leaderProps = leaderProps; this.zkClient = zkClient; } @@ -81,4 +99,48 @@ public void joinedElectionFired() {} public ElectionContext copy() { throw new UnsupportedOperationException("copy"); } + + /** + * Registers {@link #leaderPath} as an ephemeral leader node in a single multi transaction that + * also bumps the version of the leader node's parent (via a {@code setData}), capturing it into + * {@link #leaderZkNodeParentVersion} so {@link #deleteLeaderNode()} can later remove only our + * own registration, ABA-safe. The transaction also sanity-checks that {@link #leaderSeqPath} + * still exists. + * + *

Does the ZooKeeper work only; callers own any retry, locking, and error handling around it + * (which legitimately differ between overseer and shard leader election). + */ + protected void registerLeaderNode(byte[] leaderData) + throws KeeperException, InterruptedException { + List results = + zkClient.multi( + op -> op.check().withVersion(-1).forPath(leaderSeqPath), + op -> op.create().withMode(CreateMode.EPHEMERAL).forPath(leaderPath, leaderData), + op -> op.setData().withVersion(-1).forPath(leaderParentPath, null)); + leaderZkNodeParentVersion = + results.stream() + .filter( + CuratorTransactionResult.ofTypeAndPath(OperationType.SET_DATA, leaderParentPath)) + .findFirst() + .orElseThrow( + () -> + new RuntimeException( + "Could not set data for parent path in ZK: " + leaderParentPath)) + .getResultStat() + .getVersion(); + } + + /** + * Deletes {@link #leaderPath} guarded by {@link #leaderZkNodeParentVersion}, so it only removes a + * registration whose parent version still matches — i.e. our own, never a newer lineage's/host's. + * Must only be called when {@link #leaderZkNodeParentVersion} is non-null. Callers own locking, + * any exception handling (e.g. treating {@link KeeperException.BadVersionException}/{@link + * NoNodeException} as "not ours / already gone"), and clearing {@link #leaderZkNodeParentVersion} + * afterward. + */ + protected void deleteLeaderNode() throws KeeperException, InterruptedException { + zkClient.multi( + op -> op.check().withVersion(leaderZkNodeParentVersion).forPath(leaderParentPath), + op -> op.delete().withVersion(-1).forPath(leaderPath)); + } } 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 e3b1651eeeef..0f390123a7b0 100644 --- a/solr/core/src/java/org/apache/solr/cloud/Overseer.java +++ b/solr/core/src/java/org/apache/solr/cloud/Overseer.java @@ -25,7 +25,6 @@ import java.util.ArrayDeque; import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; import org.apache.solr.client.solrj.cloud.SolrCloudManager; @@ -66,7 +65,6 @@ import org.apache.solr.update.UpdateShardHandler; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; -import org.apache.zookeeper.data.Stat; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -182,6 +180,9 @@ private class ClusterStateUpdater implements SolrInfoBean, Runnable, Closeable { private final Compressor compressor; private boolean isClosed = false; + // Set when this overseer is told to step down via an explicit QUIT (roles handoff). Read in + // run()'s finally to decide whether to spawn the OverseerExitThread to rejoin the election. + private volatile boolean quitReceived = false; public ClusterStateUpdater( final ZkStateReader reader, @@ -234,6 +235,7 @@ public void run() { if (log.isInfoEnabled()) { log.info("Starting to work on the main queue : {}", LeaderElector.getNodeName(myId)); } + boolean crashed = false; try { ZkStateWriter zkStateWriter = null; ClusterState clusterState = null; @@ -391,14 +393,29 @@ public void run() { refreshClusterState = true; // it might have been a bad version error } } + } catch (Throwable t) { + // The main loop terminated abnormally -- not a clean close, not a session-expiry return, + // not + // a QUIT. Rejoin below so we recover instead of leaving a dead overseer still holding the + // /overseer_elect/leader znode with nothing behind it. + crashed = true; + log.error("Overseer main loop terminated unexpectedly", t); } finally { if (log.isInfoEnabled()) { log.info("Overseer Loop exiting : {}", LeaderElector.getNodeName(myId)); } - // do this in a separate thread because any wait is interrupted in this main thread - Thread checkLeaderThread = new Thread(this::checkIfIamStillLeader, "OverseerExitThread"); - checkLeaderThread.setDaemon(true); - checkLeaderThread.start(); + // Only spawn the exit thread to rejoin the election when nobody else will: an explicit QUIT + // (roles handoff) or an unexpected crash. On a clean close or a ZK session-expiry + // reconnect, + // the ZkController reconnect handler owns re-election, so spawning here would just race it + // and + // risk two competing overseer lineages. + if (quitReceived || crashed) { + // do this in a separate thread because any wait is interrupted in this main thread + Thread checkLeaderThread = new Thread(this::checkIfIamStillLeader, "OverseerExitThread"); + checkLeaderThread.setDaemon(true); + checkLeaderThread.start(); + } } } @@ -455,44 +472,15 @@ private void checkIfIamStillLeader() { && (zkController.getCoreContainer().isShutDown() || zkController.isClosed())) { return; // shutting down no need to go further } - Stat stat = new Stat(); - final String path = OVERSEER_ELECT + "/leader"; - byte[] data; - try { - data = zkClient.getData(path, null, stat); - } catch (IllegalStateException | KeeperException.NoNodeException e) { - return; - } catch (Exception e) { - log.warn("Error communicating with ZooKeeper", e); - return; - } + // We only reach here after a QUIT (roles handoff) or an unexpected crash, i.e. cases where no + // Zk reconnect handler will re-drive the election. The rejoin below cancels our context, + // which is what removes our leader registration. try { - Map m = (Map) Utils.fromJSON(data); - String id = (String) m.get(ID); - if (overseerCollectionConfigSetProcessor.getId().equals(id)) { - try { - log.warn( - "I (id={}) am exiting, but I'm still the leader", - overseerCollectionConfigSetProcessor.getId()); - zkClient.delete(path, stat.getVersion()); - } catch (KeeperException.BadVersionException e) { - // no problem ignore it some other Overseer has already taken over - } catch (Exception e) { - log.error("Could not delete my leader node {}", path, e); - } - - } else { - log.info("somebody else (id={}) has already taken up the overseer position", id); - } - } finally { - // if I am not shutting down, Then I need to rejoin election - try { - if (zkController != null && !zkController.getCoreContainer().isShutDown()) { - zkController.rejoinOverseerElection(null, false); - } - } catch (Exception e) { - log.warn("Unable to rejoinElection ", e); + if (zkController != null && !zkController.getCoreContainer().isShutDown()) { + zkController.rejoinOverseerElection(null, false); } + } catch (Exception e) { + log.warn("Unable to rejoinElection ", e); } } @@ -572,6 +560,7 @@ private List processMessage( if (log.isInfoEnabled()) { log.info("Quit command received {} {}", message, LeaderElector.getNodeName(myId)); } + quitReceived = true; IOUtils.closeQuietly(overseerCollectionConfigSetProcessor); IOUtils.closeQuietly(this); } else { diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerElectionContext.java b/solr/core/src/java/org/apache/solr/cloud/OverseerElectionContext.java index 0dfc5d087d2f..b99dac1949d9 100644 --- a/solr/core/src/java/org/apache/solr/cloud/OverseerElectionContext.java +++ b/solr/core/src/java/org/apache/solr/cloud/OverseerElectionContext.java @@ -26,7 +26,6 @@ import org.apache.solr.common.cloud.ZkMaintenanceUtils; import org.apache.solr.common.cloud.ZkNodeProps; import org.apache.solr.common.util.Utils; -import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -61,11 +60,21 @@ void runLeaderProcess(boolean weAreReplacement) throws KeeperException, Interrup final String id = leaderSeqPath.substring(leaderSeqPath.lastIndexOf('/') + 1); ZkNodeProps myProps = new ZkNodeProps(ID, id); - zkClient.makePath(leaderPath, Utils.toJSON(myProps), CreateMode.EPHEMERAL); - + // Register and start under the same lock close() takes, so a close() cannot land between them + // and leave a leader znode with no overseer behind it. Registration also captures the parent + // version so cancelElection() only deletes our own. Mirrors ShardLeaderElectionContextBase. synchronized (this) { - if (!this.isClosed && !overseer.getZkController().getCoreContainer().isShutDown()) { + boolean shutDown = overseer.getZkController().getCoreContainer().isShutDown(); + if (!this.isClosed && !shutDown) { + registerLeaderNode(Utils.toJSON(myProps)); + log.info("Created overseer leader registration {} -> {}", leaderPath, id); overseer.start(id); + } else { + log.info( + "Not registering as overseer leader for {}: isClosed={}, shutDown={}", + leaderPath, + this.isClosed, + shutDown); } } } @@ -73,6 +82,19 @@ void runLeaderProcess(boolean weAreReplacement) throws KeeperException, Interrup @Override public void cancelElection() throws InterruptedException, KeeperException { super.cancelElection(); + // Delete only our own registration, guarded by the parent version captured at registration, so + // we can never remove a newer lineage's (ABA-safe). Mirrors ShardLeaderElectionContextBase. + synchronized (this) { + if (leaderZkNodeParentVersion != null) { + try { + deleteLeaderNode(); + } catch (KeeperException.BadVersionException | KeeperException.NoNodeException e) { + // A newer lineage already re-registered (parent version bumped) or the node is already + // gone -- either way the leader znode is not ours to remove. + } + leaderZkNodeParentVersion = null; + } + } overseer.close(); } 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..7b9f55352349 100644 --- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java +++ b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java @@ -164,7 +164,8 @@ void runLeaderProcess(boolean weAreReplacement) throws KeeperException, Interrup if (shouldAbort()) { // Solr is shutting down or the ZooKeeper session expired while waiting for replicas. If the - // later, we cannot be sure we are still the leader, so we should bail out. The OnReconnect + // later, we cannot be sure we are still the leader, so we should bail out. The + // OnExpiredReconnection // handler will re-register the cores and handle a new leadership election. return; } diff --git a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java index 8e9ceeaba221..51a1fd3e4dc7 100644 --- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java +++ b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java @@ -18,11 +18,8 @@ package org.apache.solr.cloud; import java.lang.invoke.MethodHandles; -import java.util.List; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import org.apache.curator.framework.api.transaction.CuratorTransactionResult; -import org.apache.curator.framework.api.transaction.OperationType; import org.apache.solr.cloud.overseer.OverseerAction; import org.apache.solr.common.SolrException; import org.apache.solr.common.SolrException.ErrorCode; @@ -51,7 +48,6 @@ class ShardLeaderElectionContextBase extends ElectionContext { protected LeaderElector leaderElector; protected ZkStateReader zkStateReader; protected ZkController zkController; - protected Integer leaderZkNodeParentVersion; // Prevents a race between cancelling and becoming leader. private final Object lock = new Object(); @@ -76,11 +72,11 @@ public ShardLeaderElectionContextBase( this.shardId = shardId; this.collection = collection; - String parent = ZkMaintenanceUtils.getZkParent(leaderPath); // only if /collections/{collection} exists already do we succeed in creating this path - log.info("make sure parent is created {}", parent); + log.info("make sure parent is created {}", leaderParentPath); try { - ZkMaintenanceUtils.ensureExists(parent, (byte[]) null, CreateMode.PERSISTENT, zkClient, 2); + ZkMaintenanceUtils.ensureExists( + leaderParentPath, (byte[]) null, CreateMode.PERSISTENT, zkClient, 2); } catch (KeeperException e) { throw new RuntimeException(e); } catch (InterruptedException e) { @@ -104,10 +100,7 @@ public void cancelElection() throws InterruptedException, KeeperException { "Removing leader registration node on cancel: {} {}", leaderPath, leaderZkNodeParentVersion); - String parent = ZkMaintenanceUtils.getZkParent(leaderPath); - zkClient.multi( - op -> op.check().withVersion(leaderZkNodeParentVersion).forPath(parent), - op -> op.delete().withVersion(-1).forPath(leaderPath)); + deleteLeaderNode(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw e; @@ -126,7 +119,6 @@ public void cancelElection() throws InterruptedException, KeeperException { void runLeaderProcess(boolean weAreReplacement) throws KeeperException, InterruptedException { // register as leader - if an ephemeral is already there, wait to see if it goes away - String parent = ZkMaintenanceUtils.getZkParent(leaderPath); try { RetryUtil.retryOnException( NodeExistsException.class, @@ -138,30 +130,7 @@ void runLeaderProcess(boolean weAreReplacement) throws KeeperException, Interrup "Creating leader registration node {} after winning as {}", leaderPath, leaderSeqPath); - - // We use a multi operation to get the parent nodes version, which will - // be used to make sure we only remove our own leader registration node. - // The setData call used to get the parent version is also the trigger to - // increment the version. We also do a sanity check that our leaderSeqPath exists. - List results = - zkClient.multi( - op -> op.check().withVersion(-1).forPath(leaderSeqPath), - op -> - op.create() - .withMode(CreateMode.EPHEMERAL) - .forPath(leaderPath, Utils.toJSON(leaderProps)), - op -> op.setData().withVersion(-1).forPath(parent, null)); - leaderZkNodeParentVersion = - results.stream() - .filter( - CuratorTransactionResult.ofTypeAndPath(OperationType.SET_DATA, parent)) - .findFirst() - .orElseThrow( - () -> - new RuntimeException( - "Could not set data for parent path in ZK: " + parent)) - .getResultStat() - .getVersion(); + registerLeaderNode(Utils.toJSON(leaderProps)); } }); } catch (NoNodeException 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 a3702c1f1e31..bd3a9bb50f9d 100644 --- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java +++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java @@ -27,6 +27,7 @@ import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDROLE; import static org.apache.zookeeper.ZooDefs.Ids.OPEN_ACL_UNSAFE; +import com.google.common.annotations.VisibleForTesting; import io.opentelemetry.api.internal.StringUtils; import java.io.Closeable; import java.io.IOException; @@ -58,6 +59,7 @@ import java.util.function.Predicate; import java.util.stream.Collectors; import org.apache.curator.framework.api.ACLProvider; +import org.apache.curator.framework.state.ConnectionStateListener; import org.apache.solr.client.api.util.SolrVersion; import org.apache.solr.client.solrj.SolrClient; import org.apache.solr.client.solrj.cloud.SolrCloudManager; @@ -84,13 +86,12 @@ import org.apache.solr.common.cloud.DocCollectionWatcher; import org.apache.solr.common.cloud.LiveNodesListener; import org.apache.solr.common.cloud.NodesSysPropsCacher; -import org.apache.solr.common.cloud.OnDisconnect; -import org.apache.solr.common.cloud.OnReconnect; import org.apache.solr.common.cloud.PerReplicaStates; import org.apache.solr.common.cloud.PerReplicaStatesOps; import org.apache.solr.common.cloud.Replica; import org.apache.solr.common.cloud.SecurityAwareZkACLProvider; import org.apache.solr.common.cloud.Slice; +import org.apache.solr.common.cloud.SolrCuratorEvent; import org.apache.solr.common.cloud.SolrZkClient; import org.apache.solr.common.cloud.ZkACLProvider; import org.apache.solr.common.cloud.ZkCoreNodeProps; @@ -212,8 +213,10 @@ public String toString() { private final ExecutorService zkConnectionListenerCallbackExecutor = ExecutorUtil.newMDCAwareSingleThreadExecutor( new SolrNamedThreadFactory("zkConnectionListenerCallback")); - private final OnReconnect onReconnect = this::onReconnect; - private final OnDisconnect onDisconnect = this::onDisconnect; + private final ConnectionStateListener onExpiredReconnection = + SolrCuratorEvent.EXPIRED_RECONNECTION.of(this::onExpiredReconnection); + private final ConnectionStateListener onSessionExpiration = + SolrCuratorEvent.SESSION_EXPIRATION.of(this::onSessionExpiration); private final String zkServerAddress; // example: 127.0.0.1:54062/solr @@ -253,7 +256,8 @@ public String toString() { // keeps track of a list of objects that need to know a new ZooKeeper session was created after // expiration occurred ref is held as a HashSet since we clone the set before notifying to avoid // synchronizing too long - private final HashSet reconnectListeners = new HashSet<>(); + private final HashSet expiredReconnectionListeners = + new HashSet<>(); private class RegisterCoreAsync implements Callable { @@ -322,7 +326,7 @@ public ZkController( new DefaultZkCredentialsProvider()); zkCredentialsProvider.setZkCredentialsInjector(zkCredentialsInjector); - addOnReconnectListener(getConfigDirListener()); + addExpiredReconnectionListener(getConfigDirListener()); final var compressor = loadPluginOrDefault( @@ -342,11 +346,11 @@ public ZkController( zkClient .getCuratorFramework() .getConnectionStateListenable() - .addListener(onReconnect, zkConnectionListenerCallbackExecutor); + .addListener(onExpiredReconnection, zkConnectionListenerCallbackExecutor); zkClient .getCuratorFramework() .getConnectionStateListenable() - .addListener(onDisconnect, zkConnectionListenerCallbackExecutor); + .addListener(onSessionExpiration, zkConnectionListenerCallbackExecutor); // Refuse to start if ZK has a non-empty /clusterstate.json or a /solr.xml file checkNoOldClusterstate(zkClient); @@ -401,7 +405,7 @@ public ZkController( assert ObjectReleaseTracker.track(this); } - private void onDisconnect(boolean sessionExpired) { + private void onSessionExpiration() { try { overseer.close(); } catch (Exception e) { @@ -411,7 +415,7 @@ private void onDisconnect(boolean sessionExpired) { // Close outstanding leader elections List descriptors = cc.getCoreDescriptors(); for (CoreDescriptor descriptor : descriptors) { - closeExistingElectionContext(descriptor, sessionExpired); + closeExistingElectionContext(descriptor); } // Mark all cores as not leader @@ -430,7 +434,7 @@ private T loadPluginOrDefault( return cc.getResourceLoader().newInstance(concretePluginClassName, basePluginType); } - private void onReconnect() { + private void onExpiredReconnection() { // on reconnect, reload cloud info log.info("ZooKeeper session re-connected ... refreshing core states after session expiration."); clearZkCollectionTerms(); @@ -507,34 +511,34 @@ private void onReconnect() { } // notify any other objects that need to know when the session was re-connected - HashSet clonedListeners; - synchronized (reconnectListeners) { - clonedListeners = new HashSet<>(reconnectListeners); + HashSet clonedListeners; + synchronized (expiredReconnectionListeners) { + clonedListeners = new HashSet<>(expiredReconnectionListeners); } - // the OnReconnect operation can be expensive per listener, so do that async in + // the ExpiredReconnection operation can be expensive per listener, so do that async in // the background - for (OnReconnect listener : clonedListeners) { + for (SolrCuratorEvent.EventAction listener : clonedListeners) { try { if (executorService != null) { executorService.execute( () -> { try { - listener.onReconnect(); + listener.respond(); } catch (Throwable exc) { // not much we can do here other than warn in the log log.warn( - "Error when notifying OnReconnect listener {} after session re-connected.", + "Error when notifying ExpiredReconnection listener {} after session re-connected.", listener, exc); } }); } else { - listener.onReconnect(); + listener.respond(); } } catch (Throwable exc) { // not much we can do here other than warn in the log log.warn( - "Error when notifying OnReconnect listener {} after session re-connected.", + "Error when notifying ExpiredReconnection listener {} after session re-connected.", listener, exc); } @@ -755,7 +759,7 @@ public void waitForPendingTasksToComplete() { } } - private ContextKey closeExistingElectionContext(CoreDescriptor cd, boolean sessionExpired) { + private ContextKey closeExistingElectionContext(CoreDescriptor cd) { // look for old context - if we find it, cancel it String collection = cd.getCloudDescriptor().getCollectionName(); final String coreNodeName = cd.getCloudDescriptor().getCoreNodeName(); @@ -765,11 +769,7 @@ private ContextKey closeExistingElectionContext(CoreDescriptor cd, boolean sessi if (prevContext != null) { prevContext.close(); - // Only remove the election contexts if the session expired, otherwise the ephemeral nodes - // will still exist - if (sessionExpired) { - electionContexts.remove(contextKey); - } + electionContexts.remove(contextKey); } return contextKey; @@ -779,8 +779,14 @@ public void preClose() { this.isClosed = true; try { // We do not want to react to connection state changes after we have started to close - zkClient.getCuratorFramework().getConnectionStateListenable().removeListener(onReconnect); - zkClient.getCuratorFramework().getConnectionStateListenable().removeListener(onDisconnect); + zkClient + .getCuratorFramework() + .getConnectionStateListenable() + .removeListener(onExpiredReconnection); + zkClient + .getCuratorFramework() + .getConnectionStateListenable() + .removeListener(onSessionExpiration); ExecutorUtil.shutdownNowAndAwaitTermination(zkConnectionListenerCallbackExecutor); } catch (Exception e) { log.warn( @@ -2593,41 +2599,43 @@ public void throwErrorIfReplicaReplaced(CoreDescriptor desc) { * expiration occurs; in most cases, listeners will be components that have watchers that need to * be re-created. */ - public void addOnReconnectListener(OnReconnect listener) { + public void addExpiredReconnectionListener(SolrCuratorEvent.EventAction listener) { if (listener != null) { - synchronized (reconnectListeners) { - reconnectListeners.add(listener); - log.debug("Added new OnReconnect listener {}", listener); + synchronized (expiredReconnectionListeners) { + expiredReconnectionListeners.add(listener); + log.debug("Added new ExpiredReconnection listener {}", listener); } } } /** - * Removed a previously registered OnReconnect listener, such as when a core is removed or + * Removed a previously registered expired-reconnect listener, such as when a core is removed or * reloaded. */ - public void removeOnReconnectListener(OnReconnect listener) { + public void removeExpiredReconnectionListener(SolrCuratorEvent.EventAction listener) { if (listener != null) { boolean wasRemoved; - synchronized (reconnectListeners) { - wasRemoved = reconnectListeners.remove(listener); + synchronized (expiredReconnectionListeners) { + wasRemoved = expiredReconnectionListeners.remove(listener); } if (wasRemoved) { - log.debug("Removed OnReconnect listener {}", listener); + log.debug("Removed ExpiredReconnection listener {}", listener); } else { log.warn( - "Was asked to remove OnReconnect listener {}, but remove operation " + "Was asked to remove ExpiredReconnection listener {}, but remove operation " + "did not find it in the list of registered listeners.", listener); } } } + @VisibleForTesting @SuppressWarnings({"unchecked"}) - Set getCurrentOnReconnectListeners() { - HashSet clonedListeners; - synchronized (reconnectListeners) { - clonedListeners = (HashSet) reconnectListeners.clone(); + Set getCurrentExpiredReconnectionListeners() { + HashSet clonedListeners; + synchronized (expiredReconnectionListeners) { + clonedListeners = + (HashSet) expiredReconnectionListeners.clone(); } return clonedListeners; } @@ -2878,7 +2886,7 @@ private void setConfWatcher(String zkDir, Watcher watcher, Stat stat) { } } - public OnReconnect getConfigDirListener() { + private SolrCuratorEvent.EventAction getConfigDirListener() { return () -> { synchronized (confDirectoryListeners) { for (String s : confDirectoryListeners.keySet()) { diff --git a/solr/core/src/java/org/apache/solr/handler/admin/ZookeeperInfoHandler.java b/solr/core/src/java/org/apache/solr/handler/admin/ZookeeperInfoHandler.java index c2e6d29ac5c4..5d270e43a7ec 100644 --- a/solr/core/src/java/org/apache/solr/handler/admin/ZookeeperInfoHandler.java +++ b/solr/core/src/java/org/apache/solr/handler/admin/ZookeeperInfoHandler.java @@ -44,9 +44,9 @@ import org.apache.solr.common.cloud.ClusterState; import org.apache.solr.common.cloud.DocCollection; import org.apache.solr.common.cloud.DocCollection.CollectionStateProps; -import org.apache.solr.common.cloud.OnReconnect; import org.apache.solr.common.cloud.Replica; import org.apache.solr.common.cloud.Slice.SliceStateProps; +import org.apache.solr.common.cloud.SolrCuratorEvent; import org.apache.solr.common.cloud.SolrZkClient; import org.apache.solr.common.cloud.ZkStateReader; import org.apache.solr.common.params.MapSolrParams; @@ -250,7 +250,8 @@ public String toString() { * data, this object watches the /collections znode, which will change if a collection is added or * removed. */ - static final class PagedCollectionSupport implements Watcher, Comparator, OnReconnect { + static final class PagedCollectionSupport + implements Watcher, Comparator, SolrCuratorEvent.EventAction { // this is the full merged list of collections from ZooKeeper private List cachedCollections; @@ -335,7 +336,7 @@ public int compare(String left, String right) { /** Called after a ZooKeeper session expiration occurs */ @Override - public void onReconnect() { + public void respond() { // we need to re-establish the watcher on the collections list after session expires synchronized (this) { cachedCollections = null; @@ -378,7 +379,7 @@ private void ensurePagingSupportInitialized() { ZkController zkController = cores.getZkController(); if (zkController != null) { // Get notified when the ZK session expires (so we can clear cached collections) - zkController.addOnReconnectListener(pagingSupport); + zkController.addExpiredReconnectionListener(pagingSupport); } } } diff --git a/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java b/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java index 21af5c3c6567..01285719645c 100644 --- a/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java +++ b/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java @@ -21,7 +21,7 @@ import java.util.concurrent.TimeUnit; import org.apache.solr.cloud.ZkSolrResourceLoader; import org.apache.solr.common.SolrException.ErrorCode; -import org.apache.solr.common.cloud.OnReconnect; +import org.apache.solr.common.cloud.SolrCuratorEvent; import org.apache.solr.common.cloud.SolrZkClient; import org.apache.solr.common.cloud.ZooKeeperException; import org.apache.solr.core.CloseHook; @@ -38,7 +38,7 @@ * Keeps a ManagedIndexSchema up-to-date when changes are made to the serialized managed schema in * ZooKeeper */ -public class ZkIndexSchemaReader implements OnReconnect { +public class ZkIndexSchemaReader implements SolrCuratorEvent.EventAction { private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); private final ManagedIndexSchemaFactory managedIndexSchemaFactory; private final SolrZkClient zkClient; @@ -66,10 +66,10 @@ public void preClose(SolrCore core) { if (cc.isZooKeeperAware()) { if (log.isDebugEnabled()) { log.debug( - "Removing ZkIndexSchemaReader OnReconnect listener as core {} is shutting down.", + "Removing ZkIndexSchemaReader OnExpirationReconnection listener as core {} is shutting down.", core.getName()); } - cc.getZkController().removeOnReconnectListener(ZkIndexSchemaReader.this); + cc.getZkController().removeExpiredReconnectionListener(ZkIndexSchemaReader.this); } } @@ -84,7 +84,7 @@ public void postClose(SolrCore core) { this.schemaWatcher = createSchemaWatcher(); - zkLoader.getZkController().addOnReconnectListener(this); + zkLoader.getZkController().addExpiredReconnectionListener(this); } public Object getSchemaUpdateLock() { @@ -225,7 +225,7 @@ void updateSchema(Watcher watcher, int expectedZkVersion) * the current schema from ZooKeeper. */ @Override - public void onReconnect() { + public void respond() { try { // setup a new watcher to get notified when the managed schema changes schemaWatcher = createSchemaWatcher(); diff --git a/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java b/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java index 003a04bda613..68ee32868e8e 100644 --- a/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java +++ b/solr/core/src/test/org/apache/solr/cloud/DistributedQueueTest.java @@ -27,7 +27,7 @@ import java.util.function.Predicate; import org.apache.solr.SolrTestCaseJ4; import org.apache.solr.client.solrj.cloud.DistributedQueue; -import org.apache.solr.common.cloud.OnDisconnect; +import org.apache.solr.common.cloud.SolrCuratorEvent; import org.apache.solr.common.cloud.SolrZkClient; import org.apache.solr.common.util.ExecutorUtil; import org.apache.solr.common.util.SolrNamedThreadFactory; @@ -295,13 +295,7 @@ private void forceSessionExpire() throws InterruptedException, TimeoutException zkClient .getCuratorFramework() .getConnectionStateListenable() - .addListener( - (OnDisconnect) - ((sessionExpired) -> { - if (sessionExpired) { - hasDisconnected.countDown(); - } - })); + .addListener(SolrCuratorEvent.SESSION_EXPIRATION.of(hasDisconnected::countDown)); long sessionId = zkClient.getZkSessionId(); zkServer.expire(sessionId); hasDisconnected.await(10, TimeUnit.SECONDS); diff --git a/solr/core/src/test/org/apache/solr/cloud/LeaderElectionTest.java b/solr/core/src/test/org/apache/solr/cloud/LeaderElectionTest.java index e95c1b411116..ed3268b6dc68 100644 --- a/solr/core/src/test/org/apache/solr/cloud/LeaderElectionTest.java +++ b/solr/core/src/test/org/apache/solr/cloud/LeaderElectionTest.java @@ -28,7 +28,7 @@ import java.util.concurrent.TimeUnit; import org.apache.curator.test.KillSession; import org.apache.solr.SolrTestCaseJ4; -import org.apache.solr.common.cloud.OnReconnect; +import org.apache.solr.common.cloud.SolrCuratorEvent; import org.apache.solr.common.cloud.SolrZkClient; import org.apache.solr.common.cloud.ZkCoreNodeProps; import org.apache.solr.common.cloud.ZkNodeProps; @@ -106,15 +106,18 @@ class ElectorSetup { ZkController zkController; LeaderElector elector; - public ElectorSetup(OnReconnect onReconnect) { + public ElectorSetup(SolrCuratorEvent.EventAction onExpiredReconnection) { zkClient = new SolrZkClient.Builder() .withUrl(server.getZkAddress()) .withTimeout(TIMEOUT, TimeUnit.MILLISECONDS) .withConnTimeOut(TIMEOUT, TimeUnit.MILLISECONDS) .build(); - if (onReconnect != null) { - zkClient.getCuratorFramework().getConnectionStateListenable().addListener(onReconnect); + if (onExpiredReconnection != null) { + zkClient + .getCuratorFramework() + .getConnectionStateListenable() + .addListener(SolrCuratorEvent.EXPIRED_RECONNECTION.of(onExpiredReconnection)); } zkStateReader = new ZkStateReader(zkClient); elector = new LeaderElector(zkClient); diff --git a/solr/core/src/test/org/apache/solr/cloud/OverseerElectionReconnectTest.java b/solr/core/src/test/org/apache/solr/cloud/OverseerElectionReconnectTest.java new file mode 100644 index 000000000000..b35f83dcb290 --- /dev/null +++ b/solr/core/src/test/org/apache/solr/cloud/OverseerElectionReconnectTest.java @@ -0,0 +1,287 @@ +/* + * 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 com.carrotsearch.randomizedtesting.annotations.ThreadLeakLingering; +import java.net.URI; +import java.nio.file.Path; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.solr.SolrTestCaseJ4; +import org.apache.solr.common.cloud.SolrZkClient; +import org.apache.solr.common.util.TimeSource; +import org.apache.solr.core.CloudConfig; +import org.apache.solr.core.CoreContainer; +import org.apache.solr.util.SocketProxy; +import org.apache.solr.util.TimeOut; +import org.jspecify.annotations.NonNull; +import org.junit.Test; + +/** + * Regression test that overseer election recovers correctly when a ZooKeeper session expiry + * coincides with the reconnect that re-drives election. Specifically in the case where a departing + * overseer lineage could race the reconnecting one and leave a "zombie" /overseer_elect/leader + * znode with no running overseer behind it. + */ +// This test deliberately strands ZooKeeper connections (Curator abandons one per expiry), and a +// discarded ClientCnxn.SendThread sleeps briefly inside its socket cleanup on the way out. Give +// those threads a moment to finish rather than reporting them as leaks. +@ThreadLeakLingering(linger = 2000) +public class OverseerElectionReconnectTest extends SolrTestCaseJ4 { + + private static final String SOLRXML = ""; + + /** + * ZooKeeper's fixed reconnect interval: ClientCnxn passes 1000 to hostProvider.next(), and with a + * single-server connect string that sleep runs before every retry. It is not configurable, and + * the constants below are chosen relative to it. + */ + private static final int ZK_RECONNECT_INTERVAL_MS = 1000; + + /** + * Short enough that a real expiry is reachable in a test, but deliberately larger than and offset + * from the reconnect interval -- do not round this to a multiple of it. + * + *

Curator abandons a session on its own timer, at detectionLag + this, while the old + * connection retries on a fixed ladder at detectionLag + N * interval. The detection lag appears + * in both and cancels, so a timeout that is a multiple of the interval puts those two events on + * the same millisecond. Whoever wins then decides whether the session expires at all -- and if + * the old connection wins it resumes the session, nothing expires, and this test silently + * exercises nothing. Offsetting by half an interval puts Curator's decision midway between two + * rungs, so it wins reliably rather than on a coin flip. + */ + private static final int SESSION_TIMEOUT_MS = ZK_RECONNECT_INTERVAL_MS + 500; + + /** + * ZooKeeper's session-expiry bucket width, so the server's reap of an expired session's ephemeral + * nodes can trail Curator's client-side expiry by up to this much. That lag is the whole reason + * the departing OverseerExitThread still finds the old leader znode instead of taking its NoNode + * early-out, so this must comfortably exceed the client's detection latency (~100ms). At 100 the + * reap always wins and the race is never reached. + */ + private static final int TICK_MS = 750; + + private static final int MAX_CYCLES = TEST_NIGHTLY ? 25 : 2; + + /** + * Recovery cannot start until Curator gives up, which is one session timeout after the cut, so + * this has to scale with the timeout -- a fixed value would silently time out on every cycle if + * the timeout were raised, leaving the test doing a single cycle and still passing. + */ + private static final int RECOVERY_WAIT_SECONDS = SESSION_TIMEOUT_MS / 1000 + 10; + + /** + * Reproduction of the residual overseer zombie race that survives PR #4577 (which stops + * onReconnect/onDisconnect from firing on same-session blips). Here we drive real session + * expiries and try to make the expiry coincide with the reconnect. + * + *

The race needs two things to line up. + * + *

First, the departing OverseerExitThread has to miss its early-out. onExpiredReconnection + * cancels the previous election context, which calls overseer.close(); the updater loop exits and + * its finally block spawns the OET. The OET runs checkIfIamStillLeader, which returns immediately + * if /overseer_elect/leader is already gone — so it is only dangerous while the old session's + * ephemeral leader znode is still visible. That happens because Curator does not wait for the + * server to declare the session dead; it injects the expiration on its own timer and starts a + * fresh session, while the server only reaps ephemerals on a tickTime-wide bucket. The reap can + * therefore trail the client's expiry by up to a tick, which is why the tick is set coarse here. + * + *

Second, the rejoin has to land in the registration window. Having found the stale node, the + * OET deletes it and calls rejoinOverseerElection, which picks up the elector's current context — + * by now the reconnect thread's brand-new one — and closes it while that thread is still joining. + * If the close lands between creating the leader znode and starting the overseer, the znode is + * left with no updater behind it and cancelElection() will not clean it up, so every later + * election fails with NodeExists. + */ + @Test + public void testOverseerWedgesOnExpiryRacingReconnect() throws Exception { + Path zkDir = createTempDir("zkData"); + Path ccDir = createTempDir("testOverseerExpiryRace-solr"); + + ZkTestServer zkServer = buildZkTestServer(zkDir); + try { + zkServer.run(); + + SocketProxy zkProxy = new SocketProxy(); + zkProxy.open(URI.create("http://127.0.0.1:" + zkServer.getPort())); + String proxiedZkAddress = "127.0.0.1:" + zkProxy.getListenPort() + "/solr"; + try { + // The persistent client's session timeout comes from CloudConfig.getZkClientTimeout(), not + // from the ZkController constructor arg (which only governs the bootstrap connect), so it + // has to be set on both. + CloudConfig cloudConfig = + new CloudConfig.CloudConfigBuilder("127.0.0.1", 8984) + .setZkClientTimeout(SESSION_TIMEOUT_MS) + .setLeaderConflictResolveWait(180000) + .setLeaderVoteWait(180000) + .build(); + + CoreContainer cc = createCoreContainer(ccDir, SOLRXML); + try (ZkController zkController = + new ZkController(cc, proxiedZkAddress, SESSION_TIMEOUT_MS, cloudConfig); + SolrZkClient probe = + new SolrZkClient.Builder() + .withUrl(zkServer.getZkAddress()) + .withTimeout(30000, TimeUnit.MILLISECONDS) + .build()) { + + assertNotNull("Overseer leader should be elected", waitForOverseerLeader(zkServer, 30)); + assertTrue( + "Overseer should be healthy before the storm", + waitForHealthyOverseer(zkController, probe, 30)); + assertEquals( + "Session timeout must be negotiated verbatim; check tickTime and the min/max clamping", + SESSION_TIMEOUT_MS, + zkController.getZkClient().getZkSessionTimeout()); + long sessionBefore = zkController.getZkClient().getZkSessionId(); + + for (int i = 0; i < MAX_CYCLES; i++) { + zkProxy.close(); + // Deliberate fault injection, not a poll: hold ZK unreachable long enough to expire the + // session. There is no condition to wait on, so waitFor/RetryUtil do not apply. + // + // Unpadded on purpose. The cut has to outlast the old connection's retry (detection lag + // + the reconnect interval) or that retry resumes the session and nothing ever expires. + // It also has to end before Curator's replacement connection reaches out (detection lag + // + the session timeout), or that misses the port, waits another full interval, and the + // server reaps the old leader znode before the race can happen. Reopening exactly at + // the session timeout clears the second bound for any positive detection lag. + Thread.sleep(SESSION_TIMEOUT_MS); + zkProxy.reopen(); + if (!waitForHealthyOverseer(zkController, probe, RECOVERY_WAIT_SECONDS)) { + break; + } + } + + boolean healthy = waitForHealthyOverseer(zkController, probe, RECOVERY_WAIT_SECONDS * 4); + assertTrue( + "Overseer wedged after an expiry coinciding with a reconnect (zombie leader /" + + " NodeExists spin): leader znode id=" + + leaderId(probe) + + ", running updater id=" + + updaterId(zkController), + healthy); + // Guards against this test silently becoming a no-op: if the outage stops severing the + // session, every cycle is just a same-session blip and none of the above exercises the + // race. + assertNotEquals( + "No session ever expired, so this test exercised nothing -- check SESSION_TIMEOUT_MS" + + " against the reconnect interval and the min/max clamping", + sessionBefore, + zkController.getZkClient().getZkSessionId()); + } finally { + cc.shutdown(); + } + } finally { + zkProxy.close(); + } + } finally { + zkServer.shutdown(); + } + } + + private static @NonNull ZkTestServer buildZkTestServer(Path zkDir) throws Exception { + ZkTestServer zkServer = new ZkTestServer(zkDir); + // The coarse tick is load-bearing: the server reaps an expired session's ephemerals on a + // tickTime-wide bucket, so it can lag the client's (Curator-injected) expiry by up to tickTime, + // and only that lag lets the departing OverseerExitThread still see the old leader znode. Pin + // the min/max bounds so the session timeout is negotiated verbatim; ZkTestServer's own defaults + // ([3000, 90000]) would otherwise clamp it up. + zkServer.setTheTickTime(TICK_MS); + zkServer.setMinSessionTimeout(SESSION_TIMEOUT_MS); + zkServer.setMaxSessionTimeout(SESSION_TIMEOUT_MS); + return zkServer; + } + + private boolean waitForHealthyOverseer(ZkController zkController, SolrZkClient probe, int seconds) + throws InterruptedException { + try { + new TimeOut(seconds, TimeUnit.SECONDS, TimeSource.NANO_TIME) + .waitFor("overseer did not become healthy", () -> isHealthyOverseer(zkController, probe)); + return true; + } catch (TimeoutException e) { + return false; + } + } + + /** Healthy == connected, a live updater thread, and its id equals the leader znode's id. */ + private boolean isHealthyOverseer(ZkController zkController, SolrZkClient probe) { + try { + if (!zkController.getZkClient().isConnected()) return false; + Overseer overseer = zkController.getOverseer(); + if (overseer == null || overseer.isClosed()) return false; + Overseer.OverseerThread updater = overseer.getUpdaterThread(); + if (updater == null || updater.isClosed() || !updater.isAlive()) return false; + String runningId = updaterId(zkController); + String leaderId = OverseerTaskProcessor.getLeaderId(probe); + return leaderId != null && leaderId.equals(runningId); + } catch (Exception e) { + return false; + } + } + + /** The id stored in the leader znode, or null if it cannot be read. */ + private String leaderId(SolrZkClient probe) { + try { + return OverseerTaskProcessor.getLeaderId(probe); + } catch (Exception e) { + return null; + } + } + + /** The id of the currently running updater, parsed from its thread name, or null. */ + private String updaterId(ZkController zkController) { + Overseer overseer = zkController.getOverseer(); + if (overseer == null) return null; + Overseer.OverseerThread updater = overseer.getUpdaterThread(); + if (updater == null) return null; + String prefix = "OverseerStateUpdate-"; + String name = updater.getName(); + return name.startsWith(prefix) ? name.substring(prefix.length()) : name; + } + + private String waitForOverseerLeader(ZkTestServer zkServer, int timeoutSeconds) throws Exception { + AtomicReference leader = new AtomicReference<>(); + try (SolrZkClient zc = + new SolrZkClient.Builder() + .withUrl(zkServer.getZkAddress()) + .withTimeout(30000, TimeUnit.MILLISECONDS) + .build()) { + try { + new TimeOut(timeoutSeconds, TimeUnit.SECONDS, TimeSource.NANO_TIME) + .waitFor( + "overseer leader was not elected", + () -> { + try { + String leaderNode = OverseerCollectionConfigSetProcessor.getLeaderNode(zc); + if (leaderNode != null && !leaderNode.trim().isEmpty()) { + leader.set(leaderNode); + return true; + } + } catch (Exception e) { + // Leader not yet elected + } + return false; + }); + } catch (TimeoutException e) { + // leave leader null + } + } + return leader.get(); + } +} diff --git a/solr/core/src/test/org/apache/solr/cloud/TestOnReconnectListenerSupport.java b/solr/core/src/test/org/apache/solr/cloud/TestExpiredReconnectionListenerSupport.java similarity index 76% rename from solr/core/src/test/org/apache/solr/cloud/TestOnReconnectListenerSupport.java rename to solr/core/src/test/org/apache/solr/cloud/TestExpiredReconnectionListenerSupport.java index e5cefb2b2310..4e7e11e7223e 100644 --- a/solr/core/src/test/org/apache/solr/cloud/TestOnReconnectListenerSupport.java +++ b/solr/core/src/test/org/apache/solr/cloud/TestExpiredReconnectionListenerSupport.java @@ -24,8 +24,8 @@ import java.util.concurrent.TimeUnit; import org.apache.solr.SolrTestCaseJ4.SuppressSSL; import org.apache.solr.client.solrj.request.CollectionAdminRequest; -import org.apache.solr.common.cloud.OnReconnect; import org.apache.solr.common.cloud.Replica; +import org.apache.solr.common.cloud.SolrCuratorEvent; import org.apache.solr.core.CoreContainer; import org.apache.solr.core.SolrCore; import org.apache.solr.embedded.JettySolrRunner; @@ -36,11 +36,11 @@ import org.slf4j.LoggerFactory; @SuppressSSL(bugUrl = "https://issues.apache.org/jira/browse/SOLR-5776") -public class TestOnReconnectListenerSupport extends AbstractFullDistribZkTestBase { +public class TestExpiredReconnectionListenerSupport extends AbstractFullDistribZkTestBase { private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); - public TestOnReconnectListenerSupport() { + public TestExpiredReconnectionListenerSupport() { super(); sliceCount = 2; fixShardCount(3); @@ -80,11 +80,12 @@ public void test() throws Exception { leaderCoreId = leaderCore.getName() + ":" + leaderCore.getStartNanoTime(); } - // verify the ZkIndexSchemaReader is a registered OnReconnect listener - Set listeners = zkController.getCurrentOnReconnectListeners(); - assertNotNull("ZkController returned null OnReconnect listeners", listeners); + // verify the ZkIndexSchemaReader is a registered OnExpirationReconnection listener + Set listeners = + zkController.getCurrentExpiredReconnectionListeners(); + assertNotNull("ZkController returned null OnExpirationReconnection listeners", listeners); ZkIndexSchemaReader expectedListener = null; - for (OnReconnect listener : listeners) { + for (SolrCuratorEvent.EventAction listener : listeners) { if (listener instanceof ZkIndexSchemaReader reader) { if (leaderCoreId.equals(reader.getUniqueCoreId())) { expectedListener = reader; @@ -95,7 +96,7 @@ public void test() throws Exception { assertNotNull( "ZkIndexSchemaReader for core " + leaderCoreName - + " not registered as an OnReconnect listener and should be", + + " not registered as an OnExpirationReconnection listener and should be", expectedListener); // reload the collection @@ -106,7 +107,8 @@ public void test() throws Exception { + "' failed to reload within a reasonable amount of time!", wasReloaded); - // after reload, the new core should be registered as an OnReconnect listener and the old should + // after reload, the new core should be registered as an OnExpirationReconnection listener and + // the old should // not be String reloadedLeaderCoreId; try (SolrCore leaderCore = cores.getCore(leaderCoreName)) { @@ -116,17 +118,17 @@ public void test() throws Exception { // they shouldn't be equal after reload assertNotEquals(leaderCoreId, reloadedLeaderCoreId); - listeners = zkController.getCurrentOnReconnectListeners(); - assertNotNull("ZkController returned null OnReconnect listeners", listeners); + listeners = zkController.getCurrentExpiredReconnectionListeners(); + assertNotNull("ZkController returned null OnExpirationReconnection listeners", listeners); expectedListener = null; // reset - for (OnReconnect listener : listeners) { + for (SolrCuratorEvent.EventAction listener : listeners) { if (listener instanceof ZkIndexSchemaReader reader) { if (leaderCoreId.equals(reader.getUniqueCoreId())) { fail( "Previous core " + leaderCoreId - + " should no longer be a registered OnReconnect listener! Current listeners: " + + " should no longer be a registered OnExpirationReconnection listener! Current listeners: " + listeners); } else if (reloadedLeaderCoreId.equals(reader.getUniqueCoreId())) { expectedListener = reader; @@ -138,7 +140,7 @@ public void test() throws Exception { assertNotNull( "ZkIndexSchemaReader for core " + reloadedLeaderCoreId - + " not registered as an OnReconnect listener and should be", + + " not registered as an OnExpirationReconnection listener and should be", expectedListener); // try to clean up @@ -149,18 +151,18 @@ public void test() throws Exception { log.warn("Could not delete collection {} after test completed", testCollectionName); } - listeners = zkController.getCurrentOnReconnectListeners(); - for (OnReconnect listener : listeners) { + listeners = zkController.getCurrentExpiredReconnectionListeners(); + for (SolrCuratorEvent.EventAction listener : listeners) { if (listener instanceof ZkIndexSchemaReader reader) { if (reloadedLeaderCoreId.equals(reader.getUniqueCoreId())) { fail( "Previous core " + reloadedLeaderCoreId - + " should no longer be a registered OnReconnect listener after collection delete!"); + + " should no longer be a registered OnExpirationReconnection listener after collection delete!"); } } } - log.info("TestOnReconnectListenerSupport succeeded ... shutting down now!"); + log.info("TestOnExpirationReconnectionListenerSupport succeeded ... shutting down now!"); } } 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 5c7cd7442d3f..8c19adbec04b 100644 --- a/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java +++ b/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java @@ -34,8 +34,12 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import org.apache.curator.CuratorZookeeperClient; +import org.apache.curator.test.InstanceSpec; +import org.apache.curator.test.TestingCluster; import org.apache.solr.SolrTestCaseJ4; import org.apache.solr.client.api.util.SolrVersion; import org.apache.solr.client.solrj.jetty.HttpJettySolrClient; @@ -766,6 +770,123 @@ public void testOverseerEnabledClusterPropertyTrue() throws Exception { } } + @Test + public void testZkReconnectionEvents() throws Exception { + // Do not use MiniSolrCloudCluster + + // Create a zookeeper cluster with 3 nodes + try (TestingCluster zkCluster = new TestingCluster(3)) { + zkCluster.start(); + // Now create a ZkController - it should respect the cluster property and have overseer + // enabled + CoreContainer cc = getCoreContainer(); + try { + CloudConfig cloudConfig = new CloudConfig.CloudConfigBuilder("127.0.0.1", 8983).build(); + try (ZkController zkController = + new ZkController(cc, zkCluster.getConnectString(), TIMEOUT, cloudConfig)) { + AtomicBoolean invoked = new AtomicBoolean(Boolean.FALSE); + zkController.addExpiredReconnectionListener(() -> invoked.set(true)); + CuratorZookeeperClient zkClient = + zkController.getZkClient().getCuratorFramework().getZookeeperClient(); + zkClient.getZooKeeper().getTestable().injectSessionExpiration(); + // Wait 10 seconds to make sure Solr receives the ExpiredReconnection event and invokes + // listeners + Thread.sleep(10000); + assertTrue( + "Reconnected to ZK cluster after session expiration should have triggered the invocation of method onExpiredReconnection", + invoked.get()); + invoked.set(false); + + // Kill the connected server to force Solr to reconnected to another solr server + InstanceSpec connectedIns = zkCluster.findConnectionInstance(zkClient.getZooKeeper()); + zkCluster.killServer(connectedIns); + // Wait 3 seconds to let solr connectes to another server. + Thread.sleep(3000); + InstanceSpec newConnectedIns = zkCluster.findConnectionInstance(zkClient.getZooKeeper()); + assertNotEquals(connectedIns, newConnectedIns); + // Wait 10 seconds to make sure the event is received by zkController + Thread.sleep(10000); + assertFalse( + "Reconnected to ZK cluster before session expiration should NOT trigger the invocation of method onExpiredReconnection", + invoked.get()); + } + } finally { + cc.shutdown(); + } + } finally { + // Closing zookeeper cluster is asynchronous, we need some time to let it finish. Otherwise we + // may encounter + // Thread Leak + Thread.sleep(3000); + } + } + + @Test + public void testZkDisconnectionEvents() throws Exception { + // Do not use MiniSolrCloudCluster + + // Create a zookeeper cluster with 3 nodes + try (TestingCluster zkCluster = new TestingCluster(3)) { + zkCluster.start(); + // Now create a ZkController - it should respect the cluster property and have overseer + // enabled + CoreContainer cc = getCoreContainer(); + try { + CloudConfig cloudConfig = new CloudConfig.CloudConfigBuilder("127.0.0.1", 8983).build(); + MockClusterSingleton mockClusterSingleton = new MockClusterSingleton(); + cc.getClusterSingletons() + .getSingletons() + .put(mockClusterSingleton.getName(), mockClusterSingleton); + try (ZkController zkController = + new ZkController(cc, zkCluster.getConnectString(), TIMEOUT, cloudConfig)) { + // During initialization of ZkController, mockClusterSingleton.stop is invoked and thus we + // need to reset it here. + mockClusterSingleton.reset(); + assertFalse(mockClusterSingleton.isStopped()); + CuratorZookeeperClient zkClient = + zkController.getZkClient().getCuratorFramework().getZookeeperClient(); + // Kill the connected server to force Solr to reconnected to another solr server + InstanceSpec connectedIns = zkCluster.findConnectionInstance(zkClient.getZooKeeper()); + zkCluster.killServer(connectedIns); + // Wait 3 seconds to let solr connectes to another server. + Thread.sleep(3000); + InstanceSpec newConnectedIns = zkCluster.findConnectionInstance(zkClient.getZooKeeper()); + assertNotEquals(connectedIns, newConnectedIns); + // Wait 10 seconds to make sure the event is received by zkController + Thread.sleep(10000); + assertFalse( + "Reconnected to ZK cluster before session expiration should NOT trigger the invocation of method onSessionExpiration", + mockClusterSingleton.isStopped()); + + mockClusterSingleton.reset(); + assertFalse(mockClusterSingleton.isStopped()); + AtomicBoolean invoked = new AtomicBoolean(Boolean.FALSE); + zkController.addExpiredReconnectionListener(() -> invoked.set(true)); + // Stop the cluster to prevent invoking zkController.onExpiredReconnection, + // which also stops overseer and thus invokes mockClusterSingleton.stop. + zkCluster.stop(); + // Even if the cluster is stopped, the session won't be expired at once. We still need to + // manually expire it. + zkClient.getZooKeeper().getTestable().injectSessionExpiration(); + // Wait 10 seconds to make sure Solr receives the ExpiredReconnection event and invokes + // listeners + Thread.sleep(3000); + assertFalse("ExpiredReconnection should not be triggered", invoked.get()); + assertTrue( + "Session expiration should have triggered the invocation of method onSessionExpiration", + mockClusterSingleton.isStopped()); + } + } finally { + cc.shutdown(); + } + } finally { + // Closing zookeeper cluster is asynchronous, we need some time to let it finish. Otherwise we + // may encounter + // Thread Leak + Thread.sleep(3000); + } + } + private CoreContainer getCoreContainer() { return new MockCoreContainer(); } @@ -815,4 +936,34 @@ public SolrMetricManager getMetricManager() { return metricManager; } } + + private static class MockClusterSingleton implements ClusterSingleton { + protected volatile boolean isStopped = false; + + @Override + public String getName() { + return this.getClass().getName(); + } + + @Override + public void start() throws Exception {} + + @Override + public State getState() { + return null; + } + + @Override + public void stop() { + this.isStopped = true; + } + + public boolean isStopped() { + return isStopped; + } + + public void reset() { + isStopped = false; + } + } } diff --git a/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/OnDisconnect.java b/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/OnDisconnect.java deleted file mode 100644 index 9535a59cef55..000000000000 --- a/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/OnDisconnect.java +++ /dev/null @@ -1,32 +0,0 @@ -/* - * 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.common.cloud; - -import org.apache.curator.framework.CuratorFramework; -import org.apache.curator.framework.state.ConnectionState; -import org.apache.curator.framework.state.ConnectionStateListener; - -public interface OnDisconnect extends ConnectionStateListener { - void onDisconnect(boolean sessionExpired); - - @Override - default void stateChanged(CuratorFramework client, ConnectionState newState) { - if (newState == ConnectionState.LOST || newState == ConnectionState.SUSPENDED) { - onDisconnect(newState == ConnectionState.LOST); - } - } -} diff --git a/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/OnReconnect.java b/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/OnReconnect.java deleted file mode 100644 index 8d54312d3e0f..000000000000 --- a/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/OnReconnect.java +++ /dev/null @@ -1,39 +0,0 @@ -/* - * 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.common.cloud; - -import org.apache.curator.framework.CuratorFramework; -import org.apache.curator.framework.state.ConnectionState; -import org.apache.curator.framework.state.ConnectionStateListener; - -/** - * Implementations are expected to implement a correct hashCode and equals method needed to uniquely - * identify the listener as listeners are managed in a Set. In addition, your listener - * implementation should call - * org.apache.solr.cloud.ZkController#removeOnReconnectListener(OnReconnect) when it no longer needs - * to be notified of ZK reconnection events. - */ -public interface OnReconnect extends ConnectionStateListener { - void onReconnect(); - - @Override - default void stateChanged(CuratorFramework client, ConnectionState newState) { - if (ConnectionState.RECONNECTED.equals(newState)) { - onReconnect(); - } - } -} diff --git a/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/SolrCuratorEvent.java b/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/SolrCuratorEvent.java new file mode 100644 index 000000000000..8cffc9a56214 --- /dev/null +++ b/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/SolrCuratorEvent.java @@ -0,0 +1,62 @@ +/* + * 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.common.cloud; + +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.state.ConnectionState; +import org.apache.curator.framework.state.ConnectionStateListener; + +public enum SolrCuratorEvent { + // Only triggered after resume from expiration + EXPIRED_RECONNECTION { + @Override + public ConnectionStateListener of(EventAction action) { + return new ConnectionStateListener() { + private final AtomicBoolean isExpired = new AtomicBoolean(false); + + @Override + public void stateChanged(CuratorFramework client, ConnectionState newState) { + if (newState == ConnectionState.LOST) { + isExpired.set(true); + } else if (newState == ConnectionState.RECONNECTED) { + if (isExpired.compareAndSet(true, false)) { + action.respond(); + } + } + } + }; + } + }, + + SESSION_EXPIRATION { + @Override + public ConnectionStateListener of(EventAction action) { + return (client, newState) -> { + if (newState == ConnectionState.LOST) { + action.respond(); + } + }; + } + }; + + public abstract ConnectionStateListener of(EventAction action); + + public interface EventAction { + void respond(); + } +} diff --git a/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/ZkStateReader.java b/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/ZkStateReader.java index 4f0c3bb38366..ff609508573f 100644 --- a/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/ZkStateReader.java +++ b/solr/solrj-zookeeper/src/java/org/apache/solr/common/cloud/ZkStateReader.java @@ -428,7 +428,7 @@ public ZkStateReader( .getCuratorFramework() .getConnectionStateListenable() .addListener( - (OnReconnect) + SolrCuratorEvent.EXPIRED_RECONNECTION.of( () -> { // on reconnect, reload cloud info try { @@ -440,7 +440,7 @@ public ZkStateReader( } catch (Throwable e) { log.error("An error has occurred while updating the cluster state", e); } - }); + })); this.closeClient = true; this.securityNodeWatcher = null; collectionPropertiesZkStateReader = new CollectionPropertiesZkStateReader(this); diff --git a/solr/test-framework/src/java/org/apache/solr/cloud/ZkTestServer.java b/solr/test-framework/src/java/org/apache/solr/cloud/ZkTestServer.java index 8be5bec00db0..4f453eceb788 100644 --- a/solr/test-framework/src/java/org/apache/solr/cloud/ZkTestServer.java +++ b/solr/test-framework/src/java/org/apache/solr/cloud/ZkTestServer.java @@ -606,6 +606,19 @@ public void setTheTickTime(int theTickTime) { this.theTickTime = theTickTime; } + /** + * Lower bound (ms) the server will negotiate for client session timeouts. Defaults to 3000; lower + * it to allow tests that need fast session expiry. Must be set before {@link #run()}. + */ + public void setMinSessionTimeout(int minSessionTimeout) { + this.minSessionTimeout = minSessionTimeout; + } + + /** Upper bound (ms) the server will negotiate for client session timeouts. Defaults to 90000. */ + public void setMaxSessionTimeout(int maxSessionTimeout) { + this.maxSessionTimeout = maxSessionTimeout; + } + public Path getZkDir() { return zkDir; }