You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by sh...@apache.org on 2020/01/27 06:39:24 UTC
[lucene-solr] branch branch_8x updated: SOLR-13897: Fix unsafe
publication of Terms object in ZkShardTerms that can cause visibility
issues and race conditions under contention
This is an automated email from the ASF dual-hosted git repository.
shalin pushed a commit to branch branch_8x
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
The following commit(s) were added to refs/heads/branch_8x by this push:
new 7316391 SOLR-13897: Fix unsafe publication of Terms object in ZkShardTerms that can cause visibility issues and race conditions under contention
7316391 is described below
commit 7316391d2dd77c486fa25b8435f0bcde33837a6d
Author: Shalin Shekhar Mangar <sh...@apache.org>
AuthorDate: Mon Jan 27 12:08:20 2020 +0530
SOLR-13897: Fix unsafe publication of Terms object in ZkShardTerms that can cause visibility issues and race conditions under contention
(cherry picked from commit 776631254ffa900527fa1ed7bcf789265cb289c1)
---
solr/CHANGES.txt | 3 +
.../org/apache/solr/cloud/ZkCollectionTerms.java | 6 ++
.../java/org/apache/solr/cloud/ZkController.java | 5 +-
.../java/org/apache/solr/cloud/ZkShardTerms.java | 71 ++++++++++++++--------
4 files changed, 56 insertions(+), 29 deletions(-)
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index 180b94d..6546f39 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -166,6 +166,9 @@ Bug Fixes
* SOLR-14189: Switch from String.trim() to StringUtils.isBlank() in query parsers (Andy Webb via Uwe Schindler)
+* SOLR-13897: Fix unsafe publication of Terms object in ZkShardTerms that can cause visibility issues
+ and race conditions under contention. (shalin)
+
Other Changes
---------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkCollectionTerms.java b/solr/core/src/java/org/apache/solr/cloud/ZkCollectionTerms.java
index 629f2bc..671bb469 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkCollectionTerms.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkCollectionTerms.java
@@ -47,6 +47,12 @@ class ZkCollectionTerms implements AutoCloseable {
}
}
+ public void register(String shardId, String coreNodeName) {
+ synchronized (terms) {
+ getShard(shardId).registerTerm(coreNodeName);
+ }
+ }
+
public void remove(String shardId, CoreDescriptor coreDescriptor) {
synchronized (terms) {
if (getShard(shardId).removeTerm(coreDescriptor)) {
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 9d7d9be..6db0c5a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -1204,12 +1204,13 @@ public class ZkController implements Closeable {
throw new SolrException(ErrorCode.SERVER_ERROR, "Error registering SolrCore, replica is removed from clusterstate");
}
- ZkShardTerms shardTerms = getShardTerms(collection, cloudDesc.getShardId());
if (replica.getType() != Type.PULL) {
- shardTerms.registerTerm(coreZkNodeName);
+ getCollectionTerms(collection).register(cloudDesc.getShardId(), coreZkNodeName);
}
+ ZkShardTerms shardTerms = getShardTerms(collection, cloudDesc.getShardId());
+
log.debug("Register replica - core:{} address:{} collection:{} shard:{}",
coreName, baseUrl, collection, shardId);
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 0cab1a8..de16670 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkShardTerms.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkShardTerms.java
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.solr.client.solrj.cloud.ShardTerms;
import org.apache.solr.common.SolrException;
@@ -65,7 +66,6 @@ public class ZkShardTerms implements AutoCloseable{
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final Object writingLock = new Object();
private final String collection;
private final String shard;
private final String znodePath;
@@ -73,11 +73,24 @@ public class ZkShardTerms implements AutoCloseable{
private final Set<CoreTermWatcher> listeners = new HashSet<>();
private final AtomicBoolean isClosed = new AtomicBoolean(false);
- private ShardTerms terms;
+ private AtomicReference<ShardTerms> terms = new AtomicReference<>();
- // Listener of a core for shard's term change events
+ /**
+ * Listener of a core for shard's term change events
+ */
interface CoreTermWatcher {
- // return true if the listener wanna to be triggered in the next time
+ /**
+ * Invoked with a Terms instance after update. <p/>
+ * Concurrent invocations of this method is not allowed so at a given time only one thread
+ * will invoke this method.
+ * <p/>
+ * <b>Note</b> - there is no guarantee that the terms version will be strictly monotonic i.e.
+ * an invocation with a newer terms version <i>can</i> be followed by an invocation with an older
+ * terms version. Implementations are required to be resilient to out-of-order invocations.
+ *
+ * @param terms instance
+ * @return true if the listener wanna to be triggered in the next time
+ */
boolean onTermChanged(ShardTerms terms);
}
@@ -101,13 +114,13 @@ public class ZkShardTerms implements AutoCloseable{
if (replicasNeedingRecovery.isEmpty()) return;
ShardTerms newTerms;
- while( (newTerms = terms.increaseTerms(leader, replicasNeedingRecovery)) != null) {
+ while( (newTerms = terms.get().increaseTerms(leader, replicasNeedingRecovery)) != null) {
if (forceSaveTerms(newTerms)) return;
}
}
public ShardTerms getShardTerms() {
- return terms;
+ return terms.get();
}
/**
* Can this replica become leader?
@@ -115,7 +128,7 @@ public class ZkShardTerms implements AutoCloseable{
* @return true if this replica can become leader, false if otherwise
*/
public boolean canBecomeLeader(String coreNodeName) {
- return terms.canBecomeLeader(coreNodeName);
+ return terms.get().canBecomeLeader(coreNodeName);
}
/**
@@ -124,7 +137,7 @@ public class ZkShardTerms implements AutoCloseable{
* @return true if this replica has term equals to leader's term, false if otherwise
*/
public boolean skipSendingUpdatesTo(String coreNodeName) {
- return !terms.haveHighestTermValue(coreNodeName);
+ return !terms.get().haveHighestTermValue(coreNodeName);
}
/**
@@ -133,7 +146,7 @@ public class ZkShardTerms implements AutoCloseable{
* @return true if this replica registered its term, false if otherwise
*/
public boolean registered(String coreNodeName) {
- return terms.getTerm(coreNodeName) != null;
+ return terms.get().getTerm(coreNodeName) != null;
}
public void close() {
@@ -147,9 +160,7 @@ public class ZkShardTerms implements AutoCloseable{
// package private for testing, only used by tests
Map<String, Long> getTerms() {
- synchronized (writingLock) {
- return terms.getTerms();
- }
+ return new HashMap<>(terms.get().getTerms());
}
/**
@@ -169,7 +180,7 @@ public class ZkShardTerms implements AutoCloseable{
int numListeners;
synchronized (listeners) {
// solrcore already closed
- listeners.removeIf(coreTermWatcher -> !coreTermWatcher.onTermChanged(terms));
+ listeners.removeIf(coreTermWatcher -> !coreTermWatcher.onTermChanged(terms.get()));
numListeners = listeners.size();
}
return removeTerm(cd.getCloudDescriptor().getCoreNodeName()) || numListeners == 0;
@@ -179,7 +190,7 @@ public class ZkShardTerms implements AutoCloseable{
// return true if this object should not be reused
boolean removeTerm(String coreNodeName) {
ShardTerms newTerms;
- while ( (newTerms = terms.removeTerm(coreNodeName)) != null) {
+ while ( (newTerms = terms.get().removeTerm(coreNodeName)) != null) {
try {
if (saveTerms(newTerms)) return false;
} catch (KeeperException.NoNodeException e) {
@@ -196,7 +207,7 @@ public class ZkShardTerms implements AutoCloseable{
*/
void registerTerm(String coreNodeName) {
ShardTerms newTerms;
- while ( (newTerms = terms.registerTerm(coreNodeName)) != null) {
+ while ( (newTerms = terms.get().registerTerm(coreNodeName)) != null) {
if (forceSaveTerms(newTerms)) break;
}
}
@@ -208,14 +219,14 @@ public class ZkShardTerms implements AutoCloseable{
*/
public void setTermEqualsToLeader(String coreNodeName) {
ShardTerms newTerms;
- while ( (newTerms = terms.setTermEqualsToLeader(coreNodeName)) != null) {
+ while ( (newTerms = terms.get().setTermEqualsToLeader(coreNodeName)) != null) {
if (forceSaveTerms(newTerms)) break;
}
}
public void setTermToZero(String coreNodeName) {
ShardTerms newTerms;
- while ( (newTerms = terms.setTermToZero(coreNodeName)) != null) {
+ while ( (newTerms = terms.get().setTermToZero(coreNodeName)) != null) {
if (forceSaveTerms(newTerms)) break;
}
}
@@ -225,7 +236,7 @@ public class ZkShardTerms implements AutoCloseable{
*/
public void startRecovering(String coreNodeName) {
ShardTerms newTerms;
- while ( (newTerms = terms.startRecovering(coreNodeName)) != null) {
+ while ( (newTerms = terms.get().startRecovering(coreNodeName)) != null) {
if (forceSaveTerms(newTerms)) break;
}
}
@@ -235,13 +246,13 @@ public class ZkShardTerms implements AutoCloseable{
*/
public void doneRecovering(String coreNodeName) {
ShardTerms newTerms;
- while ( (newTerms = terms.doneRecovering(coreNodeName)) != null) {
+ while ( (newTerms = terms.get().doneRecovering(coreNodeName)) != null) {
if (forceSaveTerms(newTerms)) break;
}
}
public boolean isRecovering(String name) {
- return terms.isRecovering(name);
+ return terms.get().isRecovering(name);
}
/**
@@ -250,17 +261,17 @@ public class ZkShardTerms implements AutoCloseable{
*/
public void ensureHighestTermsAreNotZero() {
ShardTerms newTerms;
- while ( (newTerms = terms.ensureHighestTermsAreNotZero()) != null) {
+ while ( (newTerms = terms.get().ensureHighestTermsAreNotZero()) != null) {
if (forceSaveTerms(newTerms)) break;
}
}
public long getHighestTerm() {
- return terms.getMaxTerm();
+ return terms.get().getMaxTerm();
}
public long getTerm(String coreNodeName) {
- Long term = terms.getTerm(coreNodeName);
+ Long term = terms.get().getTerm(coreNodeName);
return term == null? -1 : term;
}
@@ -408,12 +419,18 @@ public class ZkShardTerms implements AutoCloseable{
*/
private void setNewTerms(ShardTerms newTerms) {
boolean isChanged = false;
- synchronized (writingLock) {
- if (terms == null || newTerms.getVersion() > terms.getVersion()) {
- terms = newTerms;
- isChanged = true;
+ for (;;) {
+ ShardTerms terms = this.terms.get();
+ if (terms == null || newTerms.getVersion() > terms.getVersion()) {
+ if (this.terms.compareAndSet(terms, newTerms)) {
+ isChanged = true;
+ break;
+ }
+ } else {
+ break;
}
}
+
if (isChanged) onTermUpdates(newTerms);
}