This is an automated email from the ASF dual-hosted git repository. dsmiley pushed a commit to branch branch_10x in repository https://gitbox.apache.org/repos/asf/solr.git
commit d1132fca39e83a7835efd1c2948742a37625e19c Author: NextBrickINC.com <[email protected]> AuthorDate: Wed Sep 16 07:50:56 2026 -0700 SOLR-18298: only recover after ZooKeeper session expiry (#4774) The migration to curator incidentally resulted in Solr's onReconnect triggering on reconnect. Funnily enough, that's not the desired behavior at this point. Solr's OnReconnect was designed to be triggered only when there is a reconnect *after* zk session expiration (so it is not literally "on reconnect"). This fix is the minimal way of reverting to the old behavior while keeping the naming. Co-authored-by: Shrey Narayan <[email protected]> Co-authored-by: Cursor <[email protected]> Co-authored-by: rayshrey <[email protected]> (cherry picked from commit 98a61bf53f16c0ea27225e31415f56d027076179) --- .../SOLR-18298-zk-reconnect-session-expiry.yml | 8 +++ .../java/org/apache/solr/cloud/ZkController.java | 9 ++++ .../org/apache/solr/cloud/ZkControllerTest.java | 51 ++++++++++++++++++ .../apache/solr/common/cloud/ZkStateReader.java | 42 +++++++++------ .../common/cloud/TestOnReconnectSessionExpiry.java | 60 ++++++++++++++++++++++ 5 files changed, 153 insertions(+), 17 deletions(-) diff --git a/changelog/unreleased/SOLR-18298-zk-reconnect-session-expiry.yml b/changelog/unreleased/SOLR-18298-zk-reconnect-session-expiry.yml new file mode 100644 index 00000000000..e9f6f4527c4 --- /dev/null +++ b/changelog/unreleased/SOLR-18298-zk-reconnect-session-expiry.yml @@ -0,0 +1,8 @@ +title: ZkController no longer treats every ZooKeeper reconnect as session expiration. Re-election and core re-registration now run only after Curator reports ConnectionState.LOST, restoring Solr 9 behavior during ZK rolling restarts. +type: fixed +authors: + - name: Shrey Narayan (NextBrick) + url: https://nextbrick.com +links: + - name: SOLR-18298 + url: https://issues.apache.org/jira/browse/SOLR-18298 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 051bdf753af..f006584edb2 100644 --- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java +++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java @@ -54,6 +54,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Predicate; import java.util.stream.Collectors; @@ -214,6 +215,7 @@ public class ZkController implements Closeable { new SolrNamedThreadFactory("zkConnectionListenerCallback")); private final OnReconnect onReconnect = this::onReconnect; private final OnDisconnect onDisconnect = this::onDisconnect; + private final AtomicBoolean zkSessionExpired = new AtomicBoolean(); private final String zkServerAddress; // example: 127.0.0.1:54062/solr @@ -410,6 +412,10 @@ public class ZkController implements Closeable { } private void onDisconnect(boolean sessionExpired) { + if (!sessionExpired) { + return; + } + zkSessionExpired.set(true); try { overseer.close(); } catch (Exception e) { @@ -439,6 +445,9 @@ public class ZkController implements Closeable { } private void onReconnect() { + if (!zkSessionExpired.compareAndSet(true, false)) { + return; + } // on reconnect, reload cloud info log.info("ZooKeeper session re-connected ... refreshing core states after session expiration."); clearZkCollectionTerms(); 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 53c9e3fd83e..0d993da70ac 100644 --- a/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java +++ b/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java @@ -36,6 +36,9 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; 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; @@ -50,6 +53,7 @@ import org.apache.solr.common.cloud.ZkNodeProps; import org.apache.solr.common.cloud.ZkStateReader; import org.apache.solr.common.params.CollectionParams; import org.apache.solr.common.util.ExecutorUtil; +import org.apache.solr.common.util.RetryUtil; import org.apache.solr.common.util.SolrNamedThreadFactory; import org.apache.solr.common.util.Utils; import org.apache.solr.core.CloudConfig; @@ -764,6 +768,53 @@ public class ZkControllerTest extends SolrCloudTestCase { } } + @Test + public void testReconnectRecoveryRequiresSessionExpiration() throws Exception { + try (TestingCluster zkCluster = new TestingCluster(3)) { + zkCluster.start(); + 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)) { + AtomicInteger recoveries = new AtomicInteger(); + zkController.addOnReconnectListener(recoveries::incrementAndGet); + CuratorZookeeperClient curatorClient = + zkController.getZkClient().getCuratorFramework().getZookeeperClient(); + + InstanceSpec connected = zkCluster.findConnectionInstance(curatorClient.getZooKeeper()); + assertNotNull(connected); + zkCluster.killServer(connected); + RetryUtil.retryUntil( + "Solr did not connect to another ZooKeeper server", + 30, + 200, + TimeUnit.MILLISECONDS, + () -> zkCluster.findConnectionInstance(curatorClient.getZooKeeper()), + current -> current != null && !current.equals(connected)); + assertEquals( + "A transient ZooKeeper reconnect must not trigger session-expiration recovery", + 0, + recoveries.get()); + + curatorClient.getZooKeeper().getTestable().injectSessionExpiration(); + RetryUtil.retryUntil( + "A reconnect after session expiration did not trigger recovery", + 30, + 200, + TimeUnit.MILLISECONDS, + () -> recoveries.get() == 1); + assertEquals(1, recoveries.get()); + } + } finally { + cc.shutdown(); + } + } finally { + // TestingCluster closes its quorum asynchronously; allow its worker threads to terminate. + Thread.sleep(3000); + } + } + private CoreContainer getCoreContainer() { return new MockCoreContainer(); } 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 b5eab171aa6..9d01f139bbb 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 @@ -390,6 +390,9 @@ public class ZkStateReader implements SolrCloseable { private final SolrZkClient zkClient; private final boolean closeClient; + private final AtomicBoolean zkSessionExpired = new AtomicBoolean(); + private final OnDisconnect onDisconnect = this::onDisconnect; + private final OnReconnect onReconnect = this::onReconnect; private volatile boolean closed = false; @@ -424,29 +427,34 @@ public class ZkStateReader implements SolrCloseable { .withConnTimeOut(zkClientConnectTimeout, TimeUnit.MILLISECONDS) .withUseDefaultCredsAndACLs(canUseZkACLs) .build(); - this.zkClient - .getCuratorFramework() - .getConnectionStateListenable() - .addListener( - (OnReconnect) - () -> { - // on reconnect, reload cloud info - try { - this.createClusterStateWatchersAndUpdate(); - } catch (InterruptedException e) { - // Restore the interrupted status - Thread.currentThread().interrupt(); - log.warn("Interrupted", e); - } catch (Throwable e) { - log.error("An error has occurred while updating the cluster state", e); - } - }); + this.zkClient.getCuratorFramework().getConnectionStateListenable().addListener(onReconnect); + this.zkClient.getCuratorFramework().getConnectionStateListenable().addListener(onDisconnect); this.closeClient = true; this.securityNodeWatcher = null; collectionPropertiesZkStateReader = new CollectionPropertiesZkStateReader(this); assert ObjectReleaseTracker.track(this); } + private void onDisconnect(boolean sessionExpired) { + if (sessionExpired) { + zkSessionExpired.set(true); + } + } + + private void onReconnect() { + if (!zkSessionExpired.compareAndSet(true, false)) { + return; + } + try { + createClusterStateWatchersAndUpdate(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("Interrupted", e); + } catch (Throwable e) { + log.error("An error has occurred while updating the cluster state", e); + } + } + /** * Forcibly refresh cluster state from ZK. Do this only to avoid race conditions because it's * expensive. diff --git a/solr/solrj-zookeeper/src/test/org/apache/solr/common/cloud/TestOnReconnectSessionExpiry.java b/solr/solrj-zookeeper/src/test/org/apache/solr/common/cloud/TestOnReconnectSessionExpiry.java new file mode 100644 index 00000000000..4e3741f863a --- /dev/null +++ b/solr/solrj-zookeeper/src/test/org/apache/solr/common/cloud/TestOnReconnectSessionExpiry.java @@ -0,0 +1,60 @@ +/* + * 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.AtomicInteger; +import org.apache.curator.framework.state.ConnectionState; +import org.apache.solr.SolrTestCase; +import org.junit.Test; + +/** Verifies the shared Curator listener adapters retain their general-purpose behavior. */ +public class TestOnReconnectSessionExpiry extends SolrTestCase { + + @Test + public void testReconnectFiresForEveryReconnectedEvent() { + AtomicInteger reconnects = new AtomicInteger(); + OnReconnect listener = () -> reconnects.incrementAndGet(); + + listener.stateChanged(null, ConnectionState.SUSPENDED); + listener.stateChanged(null, ConnectionState.RECONNECTED); + listener.stateChanged(null, ConnectionState.LOST); + listener.stateChanged(null, ConnectionState.RECONNECTED); + + listener.stateChanged(null, ConnectionState.RECONNECTED); + assertEquals(3, reconnects.get()); + } + + @Test + public void testDisconnectDistinguishesSuspensionFromSessionLoss() { + AtomicInteger suspensions = new AtomicInteger(); + AtomicInteger expirations = new AtomicInteger(); + OnDisconnect listener = + sessionExpired -> { + if (sessionExpired) { + expirations.incrementAndGet(); + } else { + suspensions.incrementAndGet(); + } + }; + + listener.stateChanged(null, ConnectionState.SUSPENDED); + listener.stateChanged(null, ConnectionState.LOST); + + assertEquals(1, suspensions.get()); + assertEquals(1, expirations.get()); + } +}
