You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ma...@apache.org on 2020/11/10 22:10:30 UTC
[lucene-solr] 02/07: @1144 Tweaking zkstatewriter towards right.
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 5e11514801871b615dc4c50d21f2c477f666b6c3
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Tue Nov 10 14:58:16 2020 -0600
@1144 Tweaking zkstatewriter towards right.
---
.../org/apache/solr/cloud/overseer/ZkStateWriter.java | 15 ++++++++++++---
.../java/org/apache/solr/common/cloud/ClusterState.java | 2 +-
2 files changed, 13 insertions(+), 4 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java b/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
index d81d122..512183a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
@@ -25,6 +25,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
@@ -255,6 +256,7 @@ public class ZkStateWriter {
public void writePendingUpdates() {
writeLock.lock();
+ Set<String> collectionsToWaitFor = ConcurrentHashMap.newKeySet();
try {
ourLock.lock();
try {
@@ -271,6 +273,12 @@ public class ZkStateWriter {
failedUpdates.clear();
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, lastFailedException.get());
}
+
+ for (String collectionName : cs.getCollectionStates().keySet()) {
+ DocCollection collection = cs.getCollectionOrNull(collectionName);
+ if (collection == null) return;
+ collectionsToWaitFor.add(collection.getName());
+ }
} finally {
ourLock.unlock();
}
@@ -313,6 +321,7 @@ public class ZkStateWriter {
});
ourLock.lock();
+ AtomicInteger lastVersion = new AtomicInteger();
try {
cs.forEachCollection(collection -> {
if (collectionsToWrite.contains(collection.getName())) {
@@ -333,8 +342,8 @@ public class ZkStateWriter {
if (v != null) {
version = v;
}
-
- reader.getZkClient().setData(path, data, version == 0 ? -1 : version, true);
+ lastVersion.set(version);
+ reader.getZkClient().setData(path, data, version, true);
trackVersions.put(collection.getName(), version + 1);
} catch (KeeperException.NoNodeException e) {
@@ -346,7 +355,7 @@ public class ZkStateWriter {
failedUpdates.put(collection.getName(), collection);
stat = reader.getZkClient().exists(path, null);
// this is a tragic error, we must disallow usage of this instance
- log.warn("Tried to update the cluster state using version={} but we where rejected, found {}", collection.getZNodeVersion(), stat.getVersion(), bve);
+ log.warn("Tried to update the cluster state using version={} but we where rejected, found {}", lastVersion.get(), stat.getVersion(), bve);
}
if (log.isDebugEnabled()) log.debug("Set version for local collection {} to {}", collection.getName(), collection.getZNodeVersion() + 1);
} catch (InterruptedException | AlreadyClosedException e) {
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ClusterState.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ClusterState.java
index 27cb7b6..1653e51 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ClusterState.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ClusterState.java
@@ -356,7 +356,7 @@ public class ClusterState implements JSONWriter.Writable {
* CollectionRef.get() != null
*/
public Map<String, CollectionRef> getCollectionStates() {
- return immutableCollectionStates;
+ return collectionStates;
}
/**