You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@solr.apache.org by ho...@apache.org on 2026/03/24 16:13:32 UTC
(solr) branch branch_10x updated: SOLR-18080: Initiate Leader election for ShardTerms (#4069)
This is an automated email from the ASF dual-hosted git repository.
houston pushed a commit to branch branch_10x
in repository https://gitbox.apache.org/repos/asf/solr.git
The following commit(s) were added to refs/heads/branch_10x by this push:
new 9878ad54b03 SOLR-18080: Initiate Leader election for ShardTerms (#4069)
9878ad54b03 is described below
commit 9878ad54b03ceb2b56e20961a1e66beffdd436db
Author: Houston Putman <ho...@apache.org>
AuthorDate: Tue Mar 24 09:05:24 2026 -0700
SOLR-18080: Initiate Leader election for ShardTerms (#4069)
(cherry picked from commit a364ac71b8392251bc2b10abfb6ff5b3b927135c)
---
...olr-18080-shard-term-induce-leader-election.yml | 9 +
.../solr/cloud/RecoveringCoreTermWatcher.java | 59 +++++-
.../solr/cloud/ShardLeaderElectionContext.java | 38 +---
.../solr/cloud/ShardLeaderElectionContextBase.java | 21 ++-
.../src/java/org/apache/solr/cloud/ShardTerms.java | 22 +++
.../java/org/apache/solr/cloud/SyncStrategy.java | 89 ++++++---
.../java/org/apache/solr/cloud/ZkShardTerms.java | 29 ++-
.../java/org/apache/solr/handler/IndexFetcher.java | 81 ++++----
.../solr/handler/admin/RequestSyncShardOp.java | 4 +-
.../apache/solr/update/DefaultSolrCoreState.java | 2 +-
.../org/apache/solr/update/SolrCmdDistributor.java | 5 +
.../processor/DistributedZkUpdateProcessor.java | 32 ++--
.../solr/cloud/LeaderVoteWaitTimeoutTest.java | 9 +
.../test/org/apache/solr/cloud/ShardTermsTest.java | 24 +++
.../solr/cloud/ZkShardTermsRecoveryTest.java | 207 +++++++++++++++++++++
.../org/apache/solr/cloud/ZkShardTermsTest.java | 48 +++++
.../solr/opentelemetry/TestDistributedTracing.java | 14 +-
.../apache/solr/common/cloud/ZkStateReader.java | 31 ++-
.../apache/solr/common/cloud/ZkCoreNodeProps.java | 4 +
19 files changed, 572 insertions(+), 156 deletions(-)
diff --git a/changelog/unreleased/solr-18080-shard-term-induce-leader-election.yml b/changelog/unreleased/solr-18080-shard-term-induce-leader-election.yml
new file mode 100644
index 00000000000..49c3f8f4db3
--- /dev/null
+++ b/changelog/unreleased/solr-18080-shard-term-induce-leader-election.yml
@@ -0,0 +1,9 @@
+# See https://github.com/apache/solr/blob/main/dev-docs/changelog.adoc
+title: ShardTerms can now induce a leader election if needed
+type: other # added, changed, fixed, deprecated, removed, dependency_update, security, other
+authors:
+ - name: Houston Putman
+ nick: HoustonPutman
+links:
+ - name: SOLR-18080
+ url: https://issues.apache.org/jira/browse/SOLR-18080
diff --git a/solr/core/src/java/org/apache/solr/cloud/RecoveringCoreTermWatcher.java b/solr/core/src/java/org/apache/solr/cloud/RecoveringCoreTermWatcher.java
index 40b340b283c..515110b9586 100644
--- a/solr/core/src/java/org/apache/solr/cloud/RecoveringCoreTermWatcher.java
+++ b/solr/core/src/java/org/apache/solr/cloud/RecoveringCoreTermWatcher.java
@@ -19,6 +19,7 @@ package org.apache.solr.cloud;
import java.lang.invoke.MethodHandles;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.solr.common.cloud.Replica;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.CoreDescriptor;
import org.apache.solr.core.SolrCore;
@@ -52,15 +53,55 @@ public class RecoveringCoreTermWatcher implements ZkShardTerms.CoreTermWatcher {
if (solrCore.getCoreDescriptor() == null
|| solrCore.getCoreDescriptor().getCloudDescriptor() == null) return true;
String coreNodeName = solrCore.getCoreDescriptor().getCloudDescriptor().getCoreNodeName();
- if (terms.haveHighestTermValue(coreNodeName)) return true;
- if (lastTermDoRecovery.get() < terms.getTerm(coreNodeName)) {
- log.info(
- "Start recovery on {} because core's term is less than leader's term", coreNodeName);
- lastTermDoRecovery.set(terms.getTerm(coreNodeName));
- solrCore
- .getUpdateHandler()
- .getSolrCoreState()
- .doRecovery(solrCore.getCoreContainer(), solrCore.getCoreDescriptor());
+
+ // If we have the highest term, there is nothing to do
+ if (terms.haveHighestTermValue(coreNodeName)) {
+ return true;
+ }
+
+ long lastRecoveryTerm;
+ long newTerm;
+ synchronized (lastTermDoRecovery) {
+ lastRecoveryTerm = lastTermDoRecovery.get();
+ newTerm = terms.getTerm(coreNodeName);
+ if (lastRecoveryTerm < newTerm) {
+ lastTermDoRecovery.set(newTerm);
+ }
+ }
+
+ if (coreDescriptor.getCloudDescriptor().isLeader()) {
+ log.warn(
+ "Removing {} leader as leader, since its term is no longer the highest. This will initiate recovery",
+ coreNodeName);
+ coreContainer.getZkController().giveupLeadership(coreDescriptor);
+ } else if (lastRecoveryTerm < newTerm) {
+ CloudDescriptor cloudDescriptor = solrCore.getCoreDescriptor().getCloudDescriptor();
+ Replica leaderReplica =
+ solrCore
+ .getCoreContainer()
+ .getZkController()
+ .getClusterState()
+ .getCollection(cloudDescriptor.getCollectionName())
+ .getSlice(cloudDescriptor.getShardId())
+ .getLeader();
+
+ // Only recover if the leader replica still has the highest term.
+ // If not, then the leader-election process will take care of recovery.
+ if (leaderReplica != null && terms.canBecomeLeader(leaderReplica.getName())) {
+ log.info(
+ "Start recovery on {} because core's term is less than leader's term", coreNodeName);
+ solrCore
+ .getUpdateHandler()
+ .getSolrCoreState()
+ .doRecovery(solrCore.getCoreContainer(), solrCore.getCoreDescriptor());
+ } else {
+ if (log.isInfoEnabled()) {
+ log.info(
+ "Defer recovery on {} because leader-election will happen soon, old leader: {}",
+ coreNodeName,
+ leaderReplica == null ? null : leaderReplica.getName());
+ }
+ }
}
} catch (Exception e) {
if (log.isInfoEnabled()) {
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 fc77cee2e20..7512867a794 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java
@@ -213,41 +213,13 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
}
}
- PeerSync.PeerSyncResult result = null;
boolean success = false;
try {
- result = syncStrategy.sync(zkController, core, leaderProps, weAreReplacement);
+ PeerSync.PeerSyncResult result =
+ syncStrategy.sync(zkController, core, leaderProps, weAreReplacement, true);
success = result.isSuccess();
} catch (Exception e) {
log.error("Exception while trying to sync", e);
- result = PeerSync.PeerSyncResult.failure();
- }
-
- UpdateLog ulog = core.getUpdateHandler().getUpdateLog();
-
- if (!success) {
- boolean hasRecentUpdates = false;
- if (ulog != null) {
- // TODO: we could optimize this if necessary
- try (UpdateLog.RecentUpdates recentUpdates = ulog.getRecentUpdates()) {
- hasRecentUpdates = !recentUpdates.getVersions(1).isEmpty();
- }
- }
-
- if (!hasRecentUpdates) {
- // we failed sync, but we have no versions - we can't sync in that case
- // - we were active
- // before, so become leader anyway if no one else has any versions either
- if (result.getOtherHasVersions().orElse(false)) {
- log.info(
- "We failed sync, but we have no versions - we can't sync in that case. But others have some versions, so we should not become leader");
- success = false;
- } else {
- log.info(
- "We failed sync, but we have no versions - we can't sync in that case - we were active before, so become leader anyway");
- success = true;
- }
- }
}
// solrcloud_debug
@@ -258,7 +230,7 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
try {
if (log.isDebugEnabled()) {
log.debug(
- "{} synched {}",
+ "{} synced {}",
core.getCoreContainer().getZkController().getNodeName(),
searcher.count(new MatchAllDocsQuery()));
}
@@ -515,12 +487,10 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
return;
}
- log.info("There may be a better leader candidate than us - going back into recovery");
+ log.info("There may be a better leader candidate than us - rejoining the election");
cancelElection();
- core.getUpdateHandler().getSolrCoreState().doRecovery(cc, core.getCoreDescriptor());
-
leaderElector.joinElection(this, true);
}
}
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 b9173384595..8e9ceeaba22 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java
@@ -19,6 +19,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;
@@ -182,8 +184,7 @@ class ShardLeaderElectionContextBase extends ElectionContext {
.getClusterState()
.getCollection(collection)
.getSlice(shardId)
- .getReplicas()
- .size()
+ .getNumLeaderReplicas()
< 2) {
Replica leader = zkStateReader.getLeader(collection, shardId);
if (leader != null
@@ -239,6 +240,22 @@ class ShardLeaderElectionContextBase extends ElectionContext {
prs)
.persist(coll.getZNode(), zkStateReader.getZkClient());
}
+ try {
+ zkStateReader.waitForState(
+ collection,
+ 10,
+ TimeUnit.SECONDS,
+ dc ->
+ dc.getLeader(shardId) != null
+ && dc.getLeader(shardId)
+ .getName()
+ .equals(leaderProps.get(ZkStateReader.CORE_NODE_NAME_PROP)));
+ } catch (TimeoutException e) {
+ throw new SolrException(
+ ErrorCode.SERVER_ERROR,
+ "Cluster state does not reflect leader change after issuing command",
+ e);
+ }
}
}
diff --git a/solr/core/src/java/org/apache/solr/cloud/ShardTerms.java b/solr/core/src/java/org/apache/solr/cloud/ShardTerms.java
index 323342c8353..dcc687f237f 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ShardTerms.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ShardTerms.java
@@ -124,6 +124,28 @@ public class ShardTerms implements MapWriter {
return replicasNeedingRecovery.contains(key);
}
+ public ShardTerms setHighestTerms(Set<String> highestTermKeys) {
+ long newMaxTerm = maxTerm + 1;
+ boolean keyFound = false;
+ HashMap<String, Long> newValues = new HashMap<>(values);
+ long nextHighestTerm = -1;
+ for (String key : values.keySet()) {
+ if (highestTermKeys.contains(key)) {
+ newValues.put(key, newMaxTerm);
+ keyFound = true;
+ } else {
+ nextHighestTerm = Math.max(nextHighestTerm, values.get(key));
+ }
+ }
+ // We only want to update if increasing the maxTerm makes an impact.
+ // If the nextHighestTerm is already < maxTerm, then upping the maxTerm doesn't do anything.
+ if (nextHighestTerm == maxTerm && keyFound) {
+ return new ShardTerms(newValues, version);
+ } else {
+ return null;
+ }
+ }
+
/**
* Return a new {@link ShardTerms} in which the highest terms are not zero
*
diff --git a/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java b/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java
index 1b2789cb3a0..cfbad7cd7e1 100644
--- a/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java
+++ b/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java
@@ -28,6 +28,7 @@ import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.jetty.HttpJettySolrClient;
import org.apache.solr.client.solrj.request.CoreAdminRequest.RequestRecovery;
+import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.params.CoreAdminParams.CoreAdminAction;
@@ -40,6 +41,7 @@ import org.apache.solr.handler.component.ShardHandler;
import org.apache.solr.handler.component.ShardRequest;
import org.apache.solr.handler.component.ShardResponse;
import org.apache.solr.update.PeerSync;
+import org.apache.solr.update.UpdateLog;
import org.apache.solr.update.UpdateShardHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,14 +81,15 @@ public class SyncStrategy {
public PeerSync.PeerSyncResult sync(
ZkController zkController, SolrCore core, ZkNodeProps leaderProps) {
- return sync(zkController, core, leaderProps, false);
+ return sync(zkController, core, leaderProps, false, false);
}
public PeerSync.PeerSyncResult sync(
ZkController zkController,
SolrCore core,
ZkNodeProps leaderProps,
- boolean peerSyncOnlyWithActive) {
+ boolean peerSyncOnlyWithActive,
+ boolean ignoreNoVersionsFailure) {
if (SKIP_AUTO_RECOVERY) {
return PeerSync.PeerSyncResult.success();
}
@@ -102,19 +105,16 @@ public class SyncStrategy {
log.info("Sync replicas to {}", ZkCoreNodeProps.getCoreUrl(leaderProps));
}
- if (core.getUpdateHandler().getUpdateLog() == null) {
- log.error("No UpdateLog found - cannot sync");
- return PeerSync.PeerSyncResult.failure();
- }
-
- return syncReplicas(zkController, core, leaderProps, peerSyncOnlyWithActive);
+ return syncReplicas(
+ zkController, core, leaderProps, peerSyncOnlyWithActive, ignoreNoVersionsFailure);
}
private PeerSync.PeerSyncResult syncReplicas(
ZkController zkController,
SolrCore core,
ZkNodeProps leaderProps,
- boolean peerSyncOnlyWithActive) {
+ boolean peerSyncOnlyWithActive,
+ boolean ignoreNoVersionsFailure) {
if (isClosed) {
log.info("We have been closed, won't sync with replicas");
return PeerSync.PeerSyncResult.failure();
@@ -129,9 +129,32 @@ public class SyncStrategy {
// first sync ourselves - we are the potential leader after all
try {
- result =
- syncWithReplicas(
- zkController, core, leaderProps, collection, shardId, peerSyncOnlyWithActive);
+ if (core.getUpdateHandler().getUpdateLog() == null) {
+ log.error("No UpdateLog found - cannot sync");
+ result = PeerSync.PeerSyncResult.failure();
+ } else {
+ result =
+ syncWithReplicas(
+ zkController, core, leaderProps, collection, shardId, peerSyncOnlyWithActive);
+ }
+
+ if (!result.isSuccess() && ignoreNoVersionsFailure) {
+ UpdateLog ulog = core.getUpdateHandler().getUpdateLog();
+ boolean hasRecentUpdates = false;
+ if (ulog != null) {
+ // TODO: we could optimize this if necessary
+ try (UpdateLog.RecentUpdates recentUpdates = ulog.getRecentUpdates()) {
+ hasRecentUpdates = !recentUpdates.getVersions(1).isEmpty();
+ }
+ }
+ // we failed sync, but we have no versions - we can't sync in that case
+ // - we were active before, so continue if no one else has any versions either
+ if (!hasRecentUpdates && !result.getOtherHasVersions().orElse(false)) {
+ log.info(
+ "We failed sync, but we have no versions - we can't sync in that case - so continue");
+ result = PeerSync.PeerSyncResult.success();
+ }
+ }
success = result.isSuccess();
} catch (Exception e) {
log.error("Sync Failed", e);
@@ -173,10 +196,10 @@ public class SyncStrategy {
String shardId,
boolean peerSyncOnlyWithActive)
throws Exception {
- List<ZkCoreNodeProps> nodes =
+ List<Replica> replicas =
zkController
.getZkStateReader()
- .getReplicaProps(
+ .getReplicas(
collection,
shardId,
core.getCoreDescriptor().getCloudDescriptor().getCoreNodeName());
@@ -186,13 +209,13 @@ public class SyncStrategy {
return PeerSync.PeerSyncResult.failure();
}
- if (nodes == null) {
+ if (replicas == null) {
// I have no replicas
return PeerSync.PeerSyncResult.success();
}
- List<String> syncWith = new ArrayList<>(nodes.size());
- for (ZkCoreNodeProps node : nodes) {
+ List<String> syncWith = new ArrayList<>(replicas.size());
+ for (Replica node : replicas) {
syncWith.add(node.getCoreUrl());
}
@@ -230,11 +253,11 @@ public class SyncStrategy {
// sync everyone else
// TODO: we should do this in parallel at least
- List<ZkCoreNodeProps> nodes =
+ List<Replica> replicas =
zkController
.getZkStateReader()
- .getReplicaProps(collection, shardId, cd.getCloudDescriptor().getCoreNodeName());
- if (nodes == null) {
+ .getReplicas(collection, shardId, cd.getCloudDescriptor().getCoreNodeName());
+ if (replicas == null) {
if (log.isInfoEnabled()) {
log.info("{} has no replicas", ZkCoreNodeProps.getCoreUrl(leaderProps));
}
@@ -242,20 +265,36 @@ public class SyncStrategy {
}
ZkCoreNodeProps zkLeader = new ZkCoreNodeProps(leaderProps);
- for (ZkCoreNodeProps node : nodes) {
+ ZkShardTerms shardTerms = zkController.getShardTerms(collection, shardId);
+ for (Replica replica : replicas) {
try {
+ if (shardTerms.registered(replica.getName())
+ && !shardTerms.canBecomeLeader(replica.getName())) {
+ if (log.isInfoEnabled()) {
+ log.info(
+ "{}: do NOT ask {} to sync, as it is not of the same shardTerm. Issue a recovery instead.",
+ ZkCoreNodeProps.getCoreUrl(leaderProps),
+ replica.getCoreUrl());
+ }
+ RecoveryRequest rr = new RecoveryRequest();
+ rr.leaderProps = leaderProps;
+ rr.baseUrl = replica.getBaseUrl();
+ rr.coreName = replica.getCoreName();
+ recoveryRequests.add(rr);
+ continue;
+ }
if (log.isInfoEnabled()) {
log.info(
"{}: try and ask {} to sync",
ZkCoreNodeProps.getCoreUrl(leaderProps),
- node.getCoreUrl());
+ replica.getCoreUrl());
}
requestSync(
- node.getBaseUrl(),
- node.getCoreUrl(),
+ replica.getBaseUrl(),
+ replica.getCoreUrl(),
zkLeader.getCoreUrl(),
- node.getCoreName(),
+ replica.getCoreName(),
nUpdates);
} catch (Exception e) {
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkShardTerms.java b/solr/core/src/java/org/apache/solr/cloud/ZkShardTerms.java
index c18bc6430d5..1b692f3db38 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkShardTerms.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkShardTerms.java
@@ -25,7 +25,9 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
+import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.common.SolrException;
+import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.util.ObjectReleaseTracker;
@@ -53,11 +55,17 @@ import org.slf4j.LoggerFactory;
* <p>The values correspond to replicas are called terms. Only replicas with highest term value are
* considered up to date and be able to become leader and serve queries.
*
- * <p>Terms can only updated in two strict ways:
+ * <p>Terms can only be updated in three strict ways:
*
* <ul>
* <li>A replica sets its term equals to leader's term
* <li>The leader increase its term and some other replicas by 1
+ * <li>A subset of replicas can be set to a higher term than the leader, if the collection is in
+ * read-only mode. <br>
+ * This will initiate a leader-election. The read-only check ensures that updates cannot be
+ * distributed while this term update is taking place. Therefore, there is no concern around
+ * updates being handled by a replica that is no longer leader-eligible before that replica
+ * gets the ZkShardTerms updates and realizes that it should give-up leadership.
* </ul>
*
* This class should not be reused after {@link
@@ -120,6 +128,25 @@ public class ZkShardTerms implements AutoCloseable {
mutate(terms -> terms.increaseTerms(leader, replicasNeedingRecovery));
}
+ /**
+ * Ensure that the given cores have the highest terms, even if they are not currently leader
+ * eligible. This operation should only be used when indexes are being loaded into cores, and
+ * therefore can only be done when a collection is in a read-only state.
+ *
+ * @param collection the collection to ensure is read-only
+ * @param mostUpToDateCores the set of cores that should be leader-eligible
+ * @throws SolrServerException if the given collection is not in a read-only state
+ */
+ public void ensureHighestTerms(DocCollection collection, Set<String> mostUpToDateCores)
+ throws SolrServerException {
+ if (mostUpToDateCores.isEmpty()) return;
+ if (!collection.isReadOnly()) {
+ throw new SolrServerException(
+ "Cannot replace highest term cores when collection is not in read-only mode");
+ }
+ mutate(terms -> terms.setHighestTerms(mostUpToDateCores));
+ }
+
public ShardTerms getShardTerms() {
return terms.get();
}
diff --git a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
index 63656b5866b..7b4b7c2fdf9 100644
--- a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
+++ b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
@@ -1300,11 +1300,11 @@ public class IndexFetcher {
Directory indexDir, String filename, Long backupIndexFileLen, Long backupIndexFileChecksum) {
CompareResult compareResult = new CompareResult();
try {
- try (final IndexInput indexInput = indexDir.openInput(filename, IOContext.READONCE)) {
- long indexFileLen = indexInput.length();
- long indexFileChecksum = 0;
-
- if (backupIndexFileChecksum != null) {
+ long indexFileLen;
+ long indexFileChecksum = 0;
+ if (backupIndexFileChecksum != null) {
+ try (final IndexInput indexInput = indexDir.openInput(filename, IOContext.READONCE)) {
+ indexFileLen = indexInput.length();
try {
indexFileChecksum = CodecUtil.retrieveChecksum(indexInput);
compareResult.checkSummed = true;
@@ -1312,47 +1312,49 @@ public class IndexFetcher {
log.warn("Could not retrieve checksum from file.", e);
}
}
+ } else {
+ indexFileLen = indexDir.fileLength(filename);
+ }
- if (!compareResult.checkSummed) {
- // we don't have checksums to compare
-
- if (indexFileLen == backupIndexFileLen) {
- compareResult.equal = true;
- return compareResult;
- } else {
- log.info(
- "File {} did not match. expected length is {} and actual length is {}",
- filename,
- backupIndexFileLen,
- indexFileLen);
- compareResult.equal = false;
- return compareResult;
- }
- }
-
- // we have checksums to compare
+ if (!compareResult.checkSummed) {
+ // we don't have checksums to compare
- if (indexFileLen == backupIndexFileLen && indexFileChecksum == backupIndexFileChecksum) {
+ if (indexFileLen == backupIndexFileLen) {
compareResult.equal = true;
return compareResult;
} else {
- log.warn(
- "File {} did not match. expected checksum is {} and actual is checksum {}. "
- + "expected length is {} and actual length is {}",
+ log.info(
+ "File {} did not match. expected length is {} and actual length is {}",
filename,
- backupIndexFileChecksum,
- indexFileChecksum,
backupIndexFileLen,
indexFileLen);
compareResult.equal = false;
return compareResult;
}
}
+
+ // we have checksums to compare
+
+ if (indexFileLen == backupIndexFileLen && indexFileChecksum == backupIndexFileChecksum) {
+ compareResult.equal = true;
+ return compareResult;
+ } else {
+ log.warn(
+ "File {} did not match. expected checksum is {} and actual is checksum {}. "
+ + "expected length is {} and actual length is {}",
+ filename,
+ backupIndexFileChecksum,
+ indexFileChecksum,
+ backupIndexFileLen,
+ indexFileLen);
+ compareResult.equal = false;
+ return compareResult;
+ }
} catch (NoSuchFileException | FileNotFoundException e) {
compareResult.equal = false;
return compareResult;
} catch (IOException e) {
- log.error("Could not read file {}. Downloading it again", filename, e);
+ log.error("Could not read file {}. Assuming it is out of date", filename, e);
compareResult.equal = false;
return compareResult;
}
@@ -1383,22 +1385,9 @@ public class IndexFetcher {
String filename = (String) file.get(NAME);
Long length = (Long) file.get(SIZE);
Long checksum = (Long) file.get(CHECKSUM);
- if (slowFileExists(dir, filename)) {
- if (checksum != null) {
- if (!(compareFile(dir, filename, length, checksum).equal)) {
- // file exists and size or checksum is different, therefore we must download it again
- return true;
- }
- } else {
- if (length != dir.fileLength(filename)) {
- log.warn(
- "File {} did not match. expected length is {} and actual length is {}",
- filename,
- length,
- dir.fileLength(filename));
- return true;
- }
- }
+ if (!(compareFile(dir, filename, length, checksum).equal)) {
+ // file exists and size or checksum is different, therefore we must download it again
+ return true;
}
}
return false;
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/RequestSyncShardOp.java b/solr/core/src/java/org/apache/solr/handler/admin/RequestSyncShardOp.java
index d781834e3e7..815a3cc7624 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/RequestSyncShardOp.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/RequestSyncShardOp.java
@@ -65,7 +65,7 @@ class RequestSyncShardOp implements CoreAdminHandler.CoreAdminOp {
zkController.getZkStateReader().getBaseUrlForNodeName(zkController.getNodeName()));
boolean success =
- syncStrategy.sync(zkController, core, new ZkNodeProps(props), true).isSuccess();
+ syncStrategy.sync(zkController, core, new ZkNodeProps(props), true, false).isSuccess();
// solrcloud_debug
if (log.isDebugEnabled()) {
try {
@@ -74,7 +74,7 @@ class RequestSyncShardOp implements CoreAdminHandler.CoreAdminOp {
try {
if (log.isDebugEnabled()) {
log.debug(
- "{} synched {}",
+ "{} synced {}",
core.getCoreContainer().getZkController().getNodeName(),
searcher.count(new MatchAllDocsQuery()));
}
diff --git a/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java b/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
index e7c9474f508..55d681de4ec 100644
--- a/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
+++ b/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
@@ -51,7 +51,7 @@ public final class DefaultSolrCoreState extends SolrCoreState
private final ReentrantLock recoveryLock = new ReentrantLock();
- private final ActionThrottle recoveryThrottle = new ActionThrottle("recovery", 10000);
+ private final ActionThrottle recoveryThrottle = new ActionThrottle("recovery", 1000);
private final ActionThrottle leaderThrottle = new ActionThrottle("leader", 5000);
diff --git a/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java b/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
index 839c8a9c619..40c972aefcb 100644
--- a/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
+++ b/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
@@ -39,6 +39,7 @@ import org.apache.solr.client.solrj.request.AbstractUpdateRequest;
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.client.solrj.response.JavaBinResponseParser;
import org.apache.solr.common.SolrException;
+import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.ModifiableSolrParams;
@@ -526,6 +527,10 @@ public class SolrCmdDistributor implements Closeable {
this(nodeProps, null, null, 0);
}
+ public StdNode(Replica replica, String collection, String shardId) {
+ this(new ZkCoreNodeProps(replica), collection, shardId);
+ }
+
public StdNode(ZkCoreNodeProps nodeProps, String collection, String shardId) {
this(nodeProps, collection, shardId, 0);
}
diff --git a/solr/core/src/java/org/apache/solr/update/processor/DistributedZkUpdateProcessor.java b/solr/core/src/java/org/apache/solr/update/processor/DistributedZkUpdateProcessor.java
index 12c2695a29d..e381ae5f827 100644
--- a/solr/core/src/java/org/apache/solr/update/processor/DistributedZkUpdateProcessor.java
+++ b/solr/core/src/java/org/apache/solr/update/processor/DistributedZkUpdateProcessor.java
@@ -554,20 +554,20 @@ public class DistributedZkUpdateProcessor extends DistributedUpdateProcessor {
Replica leaderReplica =
zkController.getZkStateReader().getLeaderRetry(collection, myShardId);
// DBQ forwarded to NRT and TLOG replicas
- List<ZkCoreNodeProps> replicaProps =
+ List<Replica> foundReplicas =
zkController
.getZkStateReader()
- .getReplicaProps(
+ .getReplicas(
collection,
myShardId,
leaderReplica.getName(),
null,
Replica.State.DOWN,
EnumSet.of(Replica.Type.NRT, Replica.Type.TLOG));
- if (replicaProps != null) {
- final List<SolrCmdDistributor.Node> myReplicas = new ArrayList<>(replicaProps.size());
- for (ZkCoreNodeProps replicaProp : replicaProps) {
- myReplicas.add(new SolrCmdDistributor.StdNode(replicaProp, collection, myShardId));
+ if (foundReplicas != null) {
+ final List<SolrCmdDistributor.Node> myReplicas = new ArrayList<>(foundReplicas.size());
+ for (Replica replica : foundReplicas) {
+ myReplicas.add(new SolrCmdDistributor.StdNode(replica, collection, myShardId));
}
cmdDistrib.distribDelete(
cmd, myReplicas, params, false, rollupReplicationTracker, leaderReplicationTracker);
@@ -628,20 +628,20 @@ public class DistributedZkUpdateProcessor extends DistributedUpdateProcessor {
// TODO: what if we are no longer the leader?
forwardToLeader = false;
- List<ZkCoreNodeProps> replicaProps =
+ List<Replica> replicas =
zkController
.getZkStateReader()
- .getReplicaProps(
+ .getReplicas(
collection,
shardId,
leaderReplica.getName(),
null,
Replica.State.DOWN,
EnumSet.of(Replica.Type.NRT, Replica.Type.TLOG));
- if (replicaProps != null) {
- nodes = new ArrayList<>(replicaProps.size());
- for (ZkCoreNodeProps props : replicaProps) {
- nodes.add(new SolrCmdDistributor.StdNode(props, collection, shardId));
+ if (replicas != null) {
+ nodes = new ArrayList<>(replicas.size());
+ for (Replica replica : replicas) {
+ nodes.add(new SolrCmdDistributor.StdNode(replica, collection, shardId));
}
}
} catch (InterruptedException e) {
@@ -1279,14 +1279,14 @@ public class DistributedZkUpdateProcessor extends DistributedUpdateProcessor {
getLeaderExc);
}
- List<ZkCoreNodeProps> myReplicas =
+ List<Replica> myReplicas =
zkController
.getZkStateReader()
- .getReplicaProps(collection, cloudDesc.getShardId(), cloudDesc.getCoreNodeName());
+ .getReplicas(collection, cloudDesc.getShardId(), cloudDesc.getCoreNodeName());
boolean foundErrorNodeInReplicaList = false;
if (myReplicas != null) {
- for (ZkCoreNodeProps replicaProp : myReplicas) {
- if (((Replica) replicaProp.getNodeProps())
+ for (Replica replica : myReplicas) {
+ if (replica
.getName()
.equals(((Replica) stdNode.getNodeProps().getNodeProps()).getName())) {
foundErrorNodeInReplicaList = true;
diff --git a/solr/core/src/test/org/apache/solr/cloud/LeaderVoteWaitTimeoutTest.java b/solr/core/src/test/org/apache/solr/cloud/LeaderVoteWaitTimeoutTest.java
index c8d84db706d..1c042c6f222 100644
--- a/solr/core/src/test/org/apache/solr/cloud/LeaderVoteWaitTimeoutTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/LeaderVoteWaitTimeoutTest.java
@@ -284,6 +284,15 @@ public class LeaderVoteWaitTimeoutTest extends SolrCloudTestCase {
waitForState("Timeout waiting for 1x3 collection", collectionName, clusterShape(1, 3));
assertDocsExistInAllReplicas(Arrays.asList(leader, replica1), collectionName, 1, 3);
+
+ try (ZkShardTerms zkShardTerms =
+ new ZkShardTerms(collectionName, "shard1", cluster.getZkClient())) {
+ assertEquals(3, zkShardTerms.getTerms().size());
+ assertEquals(zkShardTerms.getHighestTerm(), zkShardTerms.getTerm(leader.getName()));
+ assertEquals(zkShardTerms.getHighestTerm(), zkShardTerms.getTerm(replica1.getName()));
+ assertEquals(zkShardTerms.getHighestTerm(), zkShardTerms.getTerm(replica2.getName()));
+ }
+
CollectionAdminRequest.deleteCollection(collectionName).process(cluster.getSolrClient());
}
diff --git a/solr/core/src/test/org/apache/solr/cloud/ShardTermsTest.java b/solr/core/src/test/org/apache/solr/cloud/ShardTermsTest.java
index 586706a92fb..579285d7207 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ShardTermsTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ShardTermsTest.java
@@ -20,6 +20,7 @@ package org.apache.solr.cloud;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.Set;
import org.apache.solr.SolrTestCase;
import org.junit.Test;
@@ -43,4 +44,27 @@ public class ShardTermsTest extends SolrTestCase {
assertEquals(2L, terms.getTerm("leader").longValue());
assertEquals(1L, terms.getTerm("dead-replica").longValue());
}
+
+ @Test
+ public void testSetHighestTerms() {
+ Map<String, Long> map = new HashMap<>();
+ map.put("leader", 0L);
+ ShardTerms terms = new ShardTerms(map, 0);
+ terms = terms.setHighestTerms(Set.of("leader"));
+ assertNull(terms);
+
+ map.put("leader", 2L);
+ map.put("live-replica", 2L);
+ map.put("another-replica", 2L);
+ map.put("bad-replica", 2L);
+ map.put("dead-replica", 1L);
+ terms = new ShardTerms(map, 0);
+
+ terms = terms.setHighestTerms(Set.of("live-replica", "another-replica"));
+ assertEquals(3L, terms.getTerm("live-replica").longValue());
+ assertEquals(3L, terms.getTerm("another-replica").longValue());
+ assertEquals(2L, terms.getTerm("leader").longValue());
+ assertEquals(2L, terms.getTerm("bad-replica").longValue());
+ assertEquals(1L, terms.getTerm("dead-replica").longValue());
+ }
}
diff --git a/solr/core/src/test/org/apache/solr/cloud/ZkShardTermsRecoveryTest.java b/solr/core/src/test/org/apache/solr/cloud/ZkShardTermsRecoveryTest.java
new file mode 100644
index 00000000000..f52b0fce8db
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/ZkShardTermsRecoveryTest.java
@@ -0,0 +1,207 @@
+/*
+ * 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.hamcrest.Matchers.in;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.embedded.JettySolrRunner;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class ZkShardTermsRecoveryTest extends SolrCloudTestCase {
+ private static final String COLLECTION = "collection1";
+ private static final int NUM_SHARDS = 2;
+ private static final int NUM_REPLICAS = 5;
+ private static int NUM_DOCS = 0;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ configureCluster(3).addConfig("conf", configset("cloud-minimal")).configure();
+ assertEquals(
+ 0,
+ CollectionAdminRequest.createCollection(COLLECTION, "conf", NUM_SHARDS, NUM_REPLICAS)
+ .process(cluster.getSolrClient())
+ .getStatus());
+ cluster.waitForActiveCollection(COLLECTION, 10, TimeUnit.SECONDS, 2, NUM_SHARDS * NUM_REPLICAS);
+ }
+
+ @Before
+ public void waitForActiveState() throws Exception {
+ CollectionAdminRequest.modifyCollection(COLLECTION, Map.of("readOnly", false))
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION, 10, TimeUnit.SECONDS, 2, NUM_SHARDS * NUM_REPLICAS);
+ }
+
+ @Test
+ public void testShardTermsInducedReplication() throws Exception {
+ String shard = "shard2";
+ if (random().nextBoolean()) {
+ // Add uncommitted/committed documents, to test that part of the recovery
+ UpdateRequest up = new UpdateRequest();
+ for (int i = 0; i < 1000; i++) {
+ up.add("id", "id2-" + i);
+ }
+ up.process(cluster.getSolrClient(), COLLECTION);
+ NUM_DOCS += 1000;
+ if (random().nextBoolean()) {
+ cluster.getSolrClient().commit(COLLECTION);
+ }
+ }
+ CollectionAdminRequest.modifyCollection(COLLECTION, Map.of("readOnly", true))
+ .process(cluster.getSolrClient());
+
+ DocCollection docCollection = cluster.getZkStateReader().getCollection(COLLECTION);
+ JettySolrRunner jetty = cluster.getRandomJetty(random());
+
+ Slice shard2 = docCollection.getSlice(shard);
+ Replica leader = shard2.getLeader();
+ Replica replica = shard2.getReplicas(r -> !r.isLeader()).getFirst();
+ List<Replica> recoveryReplicas =
+ shard2.getReplicas(
+ r -> r.getType().leaderEligible && !(r.equals(leader) || r.equals(replica)));
+
+ ZkShardTerms shardTerms =
+ jetty.getCoreContainer().getZkController().getShardTerms(COLLECTION, shard);
+ // Increase the leader and another replica's shardTerms
+ shardTerms.ensureHighestTerms(docCollection, Set.of(leader.getName(), replica.getName()));
+ ShardTerms shardTermsSnapshot = shardTerms.getShardTerms();
+ for (Replica r : recoveryReplicas) {
+ assertFalse(shardTermsSnapshot.canBecomeLeader(r.getName()));
+ }
+ waitForReplicasToGoIntoRecovery(shard, recoveryReplicas);
+
+ // Recovery should succeed relatively quickly
+ cluster.waitForActiveCollection(COLLECTION, 5, TimeUnit.SECONDS, 2, NUM_SHARDS * NUM_REPLICAS);
+ shardTerms.refreshTerms();
+ long maxTerm = shardTerms.getHighestTerm();
+ for (Replica r : recoveryReplicas) {
+ assertFalse(shardTerms.isRecovering(r.getName()));
+ assertEquals(maxTerm, shardTerms.getTerm(r.getName()));
+ }
+
+ waitForNumDocsInAllReplicas(NUM_DOCS, shard2.getReplicas(), "*:*");
+ }
+
+ @Test
+ public void testShardTermsInducedLeaderElection() throws IOException, SolrServerException {
+ String shard = "shard1";
+ if (random().nextBoolean()) {
+ // Add uncommitted/committed documents, to test that part of the recovery
+ UpdateRequest up = new UpdateRequest();
+ for (int i = 0; i < 1000; i++) {
+ up.add("id", "id3-" + i);
+ }
+ up.process(cluster.getSolrClient(), COLLECTION);
+ NUM_DOCS += 1000;
+ if (random().nextBoolean()) {
+ cluster.getSolrClient().commit(COLLECTION);
+ }
+ }
+ CollectionAdminRequest.modifyCollection(COLLECTION, Map.of("readOnly", true))
+ .process(cluster.getSolrClient());
+
+ DocCollection docCollection = cluster.getZkStateReader().getCollection(COLLECTION);
+ JettySolrRunner jetty = cluster.getRandomJetty(random());
+
+ // Increase the leader and another replica's shardTerms
+ Slice shard1 = docCollection.getSlice(shard);
+ Replica leader = shard1.getLeader();
+ Set<String> replicasToSetHighest =
+ shard1.getReplicas(r -> !r.isLeader()).subList(1, 3).stream()
+ .map(Replica::getName)
+ .collect(Collectors.toSet());
+ List<Replica> recoveryReplicas =
+ shard1.getReplicas(r -> !replicasToSetHighest.contains(r.getName()));
+ ZkShardTerms shardTerms =
+ jetty.getCoreContainer().getZkController().getShardTerms(COLLECTION, shard);
+ shardTerms.ensureHighestTerms(docCollection, replicasToSetHighest);
+ ShardTerms shardTermsSnapshot = shardTerms.getShardTerms();
+ for (Replica r : recoveryReplicas) {
+ assertFalse(shardTermsSnapshot.canBecomeLeader(r.getName()));
+ }
+
+ waitForReplicasToGoIntoRecovery(shard, recoveryReplicas);
+ Replica newLeader = getCollectionState(COLLECTION).getLeader(shard);
+ assertNotNull(newLeader);
+ assertNotEquals(newLeader, leader);
+
+ cluster.waitForActiveCollection(COLLECTION, 5, TimeUnit.SECONDS, 2, NUM_SHARDS * NUM_REPLICAS);
+ // Make sure that a leader election took place
+ assertThat(
+ cluster.getZkStateReader().getCollection(COLLECTION).getLeader(shard).getName(),
+ in(replicasToSetHighest));
+ shardTerms.refreshTerms();
+ long maxTerm = shardTerms.getHighestTerm();
+ for (Replica r : recoveryReplicas) {
+ assertFalse(shardTerms.isRecovering(r.getName()));
+ assertEquals(maxTerm, shardTerms.getTerm(r.getName()));
+ }
+
+ waitForNumDocsInAllReplicas(NUM_DOCS, shard1.getReplicas(), "*:*");
+ }
+
+ private void waitForNumDocsInAllReplicas(int numDocs, Collection<Replica> replicas, String query)
+ throws IOException, SolrServerException {
+ for (Replica r : replicas) {
+ if (!r.isActive(cluster.getSolrClient().getClusterState().getLiveNodes())) {
+ continue;
+ }
+ try (SolrClient replicaClient = getHttpSolrClient(r)) {
+ assertEquals(
+ "Replica " + r.getName() + " not up to date",
+ numDocs,
+ replicaClient.query(new SolrQuery(query)).getResults().getNumFound());
+ }
+ }
+ }
+
+ private void waitForReplicasToGoIntoRecovery(String shard, Collection<Replica> recoveryReplicas) {
+ Set<String> remainingReplicasToRecover =
+ recoveryReplicas.stream().map(Replica::getName).collect(Collectors.toSet());
+
+ waitForState(
+ "Waiting for replicas to go into recovery",
+ COLLECTION,
+ 5,
+ TimeUnit.SECONDS,
+ state -> {
+ Slice shardState = state.getSlice(shard);
+ for (Replica r : recoveryReplicas) {
+ if (shardState.getReplica(r.getName()).getState() == Replica.State.RECOVERING) {
+ remainingReplicasToRecover.remove(r.getName());
+ }
+ }
+ return remainingReplicasToRecover.isEmpty();
+ });
+ }
+}
diff --git a/solr/core/src/test/org/apache/solr/cloud/ZkShardTermsTest.java b/solr/core/src/test/org/apache/solr/cloud/ZkShardTermsTest.java
index ead1a49e6ea..eb0f3e183b8 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ZkShardTermsTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ZkShardTermsTest.java
@@ -17,6 +17,9 @@
package org.apache.solr.cloud;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
@@ -26,12 +29,14 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.util.TimeSource;
import org.apache.solr.util.TimeOut;
import org.junit.BeforeClass;
@@ -48,6 +53,8 @@ public class ZkShardTermsTest extends SolrCloudTestCase {
.addConfig(
"conf1", TEST_PATH().resolve("configsets").resolve("cloud-minimal").resolve("conf"))
.configure();
+
+ assumeWorkingMockito();
}
@Test
@@ -329,6 +336,47 @@ public class ZkShardTermsTest extends SolrCloudTestCase {
replicaTerms.close();
}
+ public void testSetHighestTerms() throws Exception {
+ String collection = "setHighestTerms";
+ DocCollection readOnlyCollection = mock(DocCollection.class);
+ DocCollection nonReadOnlyCollection = mock(DocCollection.class);
+ when(readOnlyCollection.isReadOnly()).thenReturn(true);
+ when(nonReadOnlyCollection.isReadOnly()).thenReturn(false);
+
+ ZkShardTerms leaderTerms = new ZkShardTerms(collection, "shard1", cluster.getZkClient());
+ ZkShardTerms replica1Terms = new ZkShardTerms(collection, "shard1", cluster.getZkClient());
+ ZkShardTerms replica2Terms = new ZkShardTerms(collection, "shard1", cluster.getZkClient());
+ ZkShardTerms replica3Terms = new ZkShardTerms(collection, "shard1", cluster.getZkClient());
+ leaderTerms.registerTerm("leader");
+ replica1Terms.registerTerm("replica1");
+ replica2Terms.registerTerm("replica2");
+ replica3Terms.registerTerm("replica3");
+
+ expectThrows(
+ SolrServerException.class,
+ () -> leaderTerms.ensureHighestTerms(nonReadOnlyCollection, Set.of("leader", "replica1")));
+
+ leaderTerms.ensureHighestTerms(readOnlyCollection, Set.of("leader", "replica1"));
+ waitFor(true, () -> leaderTerms.canBecomeLeader("leader"));
+ waitFor(true, () -> replica1Terms.canBecomeLeader("replica1"));
+ waitFor(false, () -> replica2Terms.canBecomeLeader("replica2"));
+ waitFor(false, () -> replica3Terms.canBecomeLeader("replica3"));
+ waitFor(false, () -> leaderTerms.skipSendingUpdatesTo("replica1"));
+ waitFor(true, () -> leaderTerms.skipSendingUpdatesTo("replica2"));
+ waitFor(true, () -> leaderTerms.skipSendingUpdatesTo("replica3"));
+
+ leaderTerms.ensureHighestTerms(readOnlyCollection, Set.of("replica2", "replica3"));
+ waitFor(false, () -> leaderTerms.canBecomeLeader("leader"));
+ waitFor(false, () -> replica1Terms.canBecomeLeader("replica1"));
+ waitFor(true, () -> replica2Terms.canBecomeLeader("replica2"));
+ waitFor(true, () -> replica3Terms.canBecomeLeader("replica3"));
+
+ leaderTerms.close();
+ replica1Terms.close();
+ replica2Terms.close();
+ replica3Terms.close();
+ }
+
private <T> void waitFor(T expected, Supplier<T> supplier) throws InterruptedException {
TimeOut timeOut = new TimeOut(10, TimeUnit.SECONDS, new TimeSource.CurrentTimeSource());
while (!timeOut.hasTimedOut()) {
diff --git a/solr/modules/opentelemetry/src/test/org/apache/solr/opentelemetry/TestDistributedTracing.java b/solr/modules/opentelemetry/src/test/org/apache/solr/opentelemetry/TestDistributedTracing.java
index e3eb646896c..ad5b474b771 100644
--- a/solr/modules/opentelemetry/src/test/org/apache/solr/opentelemetry/TestDistributedTracing.java
+++ b/solr/modules/opentelemetry/src/test/org/apache/solr/opentelemetry/TestDistributedTracing.java
@@ -227,7 +227,15 @@ public class TestDistributedTracing extends SolrCloudTestCase {
// db.instance=testInternalCollectionApiCommands_shard2_replica_n1
// db.instance=testInternalCollectionApiCommands_shard1_replica_n6
//
- // 7..8 (2 times) name=post:/{core}/get
+ // 7..8 (2 times) name=post:/{core}/get (FingerPrinting to get versions from non-leaders)
+ // db.instance=testInternalCollectionApiCommands_shard2_replica_n1
+ // db.instance=testInternalCollectionApiCommands_shard1_replica_n6
+ //
+ // 9..10 (2 times) name=post:/{core}/get (PeerSync request to non-leaders)
+ // db.instance=testInternalCollectionApiCommands_shard2_replica_n1
+ // db.instance=testInternalCollectionApiCommands_shard1_replica_n6
+ //
+ // 11..12 (2 times) name=post:/{core}/get (FingerPrinting to get versions from leaders PeerSync)
// db.instance=testInternalCollectionApiCommands_shard2_replica_n4
// db.instance=testInternalCollectionApiCommands_shard1_replica_n2
@@ -237,7 +245,7 @@ public class TestDistributedTracing extends SolrCloudTestCase {
assertEquals("create:/admin/collections", s0.getName());
Map<String, Integer> ops = new HashMap<>();
- assertEquals(7, finishedSpans.size());
+ assertEquals(11, finishedSpans.size());
var parentTraceId = getRootTraceId(finishedSpans);
for (var span : finishedSpans) {
if (isRootSpan(span)) {
@@ -250,7 +258,7 @@ public class TestDistributedTracing extends SolrCloudTestCase {
ops.put(span.getName(), ops.getOrDefault(span.getName(), 0) + 1);
}
var expectedOps =
- Map.of("CreateCollectionCmd", 1, "post:/admin/cores", 4, "post:/{core}/get", 2);
+ Map.of("CreateCollectionCmd", 1, "post:/admin/cores", 4, "post:/{core}/get", 6);
assertEquals(expectedOps, ops);
}
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 88033012b9f..7be598a520f 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
@@ -1046,10 +1046,9 @@ public class ZkStateReader implements SolrCloseable {
+ (shardId != null ? ("/" + shardId + "/" + ELECTION_NODE) : "");
}
- public List<ZkCoreNodeProps> getReplicaProps(
- String collection, String shardId, String thisCoreNodeName) {
+ public List<Replica> getReplicas(String collection, String shardId, String thisCoreNodeName) {
// TODO: It's odd that the default is to return replicas of type TLOG and NRT only
- return getReplicaProps(
+ return getReplicas(
collection,
shardId,
thisCoreNodeName,
@@ -1058,7 +1057,7 @@ public class ZkStateReader implements SolrCloseable {
EnumSet.of(Replica.Type.TLOG, Replica.Type.NRT));
}
- public List<ZkCoreNodeProps> getReplicaProps(
+ public List<Replica> getReplicas(
String collection,
String shardId,
String thisCoreNodeName,
@@ -1077,39 +1076,37 @@ public class ZkStateReader implements SolrCloseable {
}
Map<String, Slice> slices = docCollection.getSlicesMap();
- Slice replicas = slices.get(shardId);
- if (replicas == null) {
+ Slice shard = slices.get(shardId);
+ if (shard == null) {
throw new ZooKeeperException(
ErrorCode.BAD_REQUEST, "Could not find shardId in zk: " + shardId);
}
- Map<String, Replica> shardMap = replicas.getReplicasMap();
- List<ZkCoreNodeProps> nodes = new ArrayList<>(shardMap.size());
+ Map<String, Replica> shardMap = shard.getReplicasMap();
+ List<Replica> replicas = new ArrayList<>(shardMap.size());
for (Entry<String, Replica> entry :
shardMap.entrySet().stream()
.filter((e) -> acceptReplicaType.contains(e.getValue().getType()))
.collect(Collectors.toList())) {
- ZkCoreNodeProps nodeProps = new ZkCoreNodeProps(entry.getValue());
+ Replica replica = entry.getValue();
String coreNodeName = entry.getValue().getName();
- if (clusterState.liveNodesContain(nodeProps.getNodeName())
+ if (clusterState.liveNodesContain(replica.getNodeName())
&& !coreNodeName.equals(thisCoreNodeName)) {
- if (mustMatchStateFilter == null
- || mustMatchStateFilter == Replica.State.getState(nodeProps.getState())) {
- if (mustNotMatchStateFilter == null
- || mustNotMatchStateFilter != Replica.State.getState(nodeProps.getState())) {
- nodes.add(nodeProps);
+ if (mustMatchStateFilter == null || mustMatchStateFilter == replica.getState()) {
+ if (mustNotMatchStateFilter == null || mustNotMatchStateFilter != replica.getState()) {
+ replicas.add(replica);
}
}
}
}
- if (nodes.size() == 0) {
+ if (replicas.size() == 0) {
// no replicas
return null;
}
- return nodes;
+ return replicas;
}
public SolrZkClient getZkClient() {
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCoreNodeProps.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCoreNodeProps.java
index 34da8d239a0..788cb3866f1 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCoreNodeProps.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCoreNodeProps.java
@@ -47,6 +47,10 @@ public class ZkCoreNodeProps {
return nodeProps.getStr(ReplicaStateProps.CORE_NAME);
}
+ public String getCoreNodeName() {
+ return nodeProps.getStr(ReplicaStateProps.CORE_NODE_NAME);
+ }
+
private static String getBaseUrl(ZkNodeProps nodeProps) {
// if storing baseUrl in ZK is enabled, and it's stored, just use what's stored, i.e. no
// self-healing here