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/14 19:53:04 UTC
[lucene-solr] 02/02: @1202 Remove the rest of update lock.
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 10468b8400405807cb05916d75bdb2a99a9e5101
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Sat Nov 14 13:52:12 2020 -0600
@1202 Remove the rest of update lock.
---
.../solr/cloud/TestDownShardTolerantSearch.java | 42 +++----
.../apache/solr/common/cloud/ZkStateReader.java | 125 ++++++---------------
.../apache/solr/cloud/MiniSolrCloudCluster.java | 2 +-
3 files changed, 60 insertions(+), 109 deletions(-)
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestDownShardTolerantSearch.java b/solr/core/src/test/org/apache/solr/cloud/TestDownShardTolerantSearch.java
index 8b04334..d9f5745 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestDownShardTolerantSearch.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestDownShardTolerantSearch.java
@@ -18,6 +18,7 @@ package org.apache.solr.cloud;
import java.lang.invoke.MethodHandles;
+import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
@@ -46,9 +47,7 @@ public class TestDownShardTolerantSearch extends SolrCloudTestCase {
@BeforeClass
public static void setupCluster() throws Exception {
- configureCluster(2)
- .addConfig("conf", configset("cloud-minimal"))
- .configure();
+ configureCluster(2).addConfig("conf", configset("cloud-minimal")).configure();
}
@Test
@@ -68,26 +67,29 @@ public class TestDownShardTolerantSearch extends SolrCloudTestCase {
JettySolrRunner stoppedServer = cluster.stopJettySolrRunner(0);
- while (true) {
- try {
- response = cluster.getSolrClient().query("tolerant", new SolrQuery("*:*").setRows(1).setParam(ShardParams.SHARDS_TOLERANT, true));
- break;
- } catch (BaseHttpSolrClient.RemoteExecutionException e) {
- // a remote node we are proxied too may still think this is live, try again
- if (!e.getMessage().contains("Connection refused")) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
+ try (SolrClient client = cluster.buildSolrClient()) {
+ for (int i = 0; i < 10; i++) {
+
+ try {
+ response = client.query("tolerant", new SolrQuery("*:*").setRows(1).setParam(ShardParams.SHARDS_TOLERANT, true));
+ break;
+ } catch (BaseHttpSolrClient.RemoteExecutionException e) {
+ // a remote node we are proxied too may still think this is live, try again
+ if (!e.getMessage().contains("Connection refused")) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
+ }
}
}
- }
- assertThat(response.getStatus(), is(0));
- assertTrue(response.getResults().getNumFound() > 0);
+ assertThat(response.getStatus(), is(0));
+ assertTrue(response.getResults().getNumFound() > 0);
- SolrServerException e = expectThrows(SolrServerException.class, "Request should have failed because we killed shard1 jetty",
- () -> cluster.getSolrClient().query("tolerant", new SolrQuery("*:*").setRows(1).setParam(ShardParams.SHARDS_TOLERANT, false)));
- assertNotNull(e.getCause());
- if (!e.getCause().getMessage().contains("Connection refused")) {
- assertTrue("Error message from server should have the name of the down shard " + e.getCause().getMessage(), e.getCause().getMessage().contains("shard"));
+ SolrServerException e = expectThrows(SolrServerException.class, "Request should have failed because we killed shard1 jetty",
+ () -> cluster.getSolrClient().query("tolerant", new SolrQuery("*:*").setRows(1).setParam(ShardParams.SHARDS_TOLERANT, false)));
+ assertNotNull(e.getCause());
+ if (!e.getCause().getMessage().contains("Connection refused")) {
+ assertTrue("Error message from server should have the name of the down shard " + e.getCause().getMessage(), e.getCause().getMessage().contains("shard"));
+ }
}
}
- }
+}
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
index 37e652c..2c7341b 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
@@ -175,8 +175,6 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
public static final String SHARD_LEADERS_ZKNODE = "leaders";
public static final String ELECTION_NODE = "election";
- private final ReentrantLock updateLock = new ReentrantLock(true);
-
/**
* "Interesting" and actively watched Collections.
*/
@@ -347,8 +345,6 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
// don't call this, used in one place
public void forciblyRefreshAllClusterStateSlow() {
- updateLock.lock();
-
// No need to set watchers because we should already have watchers registered for everything.
try {
refreshCollectionList(null);
@@ -369,8 +365,6 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
} catch (InterruptedException e) {
ParWork.propagateInterrupt(e);
throw new SolrException(ErrorCode.SERVER_ERROR, e);
- } finally {
- updateLock.unlock();
}
}
@@ -393,12 +387,7 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
if (nu == null) return -1;
if (nu.getZNodeVersion() > collection.getZNodeVersion()) {
if (updateWatchedCollection(coll, nu)) {
- // updateLock.lock();
- try {
- constructState(Collections.singleton(coll));
- } finally {
- // updateLock.unlock();
- }
+ constructState(Collections.singleton(coll));
}
collection = nu;
}
@@ -520,28 +509,25 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
return;
}
try {
- updateLock.lock();
+
+ log.info("Updating [{}] ... ", SOLR_SECURITY_CONF_PATH);
+
+ // remake watch
+ final Stat stat = new Stat();
+ byte[] data = "{}".getBytes(StandardCharsets.UTF_8);
+ if (EventType.NodeDeleted.equals(event.getType())) {
+ // Node deleted, just recreate watch without attempting a read - SOLR-9679
+ getZkClient().exists(SOLR_SECURITY_CONF_PATH, this);
+ } else {
+ data = getZkClient().getData(SOLR_SECURITY_CONF_PATH, this, stat);
+ }
try {
- log.info("Updating [{}] ... ", SOLR_SECURITY_CONF_PATH);
-
- // remake watch
- final Stat stat = new Stat();
- byte[] data = "{}".getBytes(StandardCharsets.UTF_8);
- if (EventType.NodeDeleted.equals(event.getType())) {
- // Node deleted, just recreate watch without attempting a read - SOLR-9679
- getZkClient().exists(SOLR_SECURITY_CONF_PATH, this);
- } else {
- data = getZkClient().getData(SOLR_SECURITY_CONF_PATH, this, stat);
- }
- try {
- callback.call(new Pair<>(data, stat));
- } catch (Exception e) {
- log.error("Error running collections node listener", e);
- return;
- }
- } finally {
- updateLock.unlock();
+ callback.call(new Pair<>(data, stat));
+ } catch (Exception e) {
+ log.error("Error running collections node listener", e);
+ return;
}
+
} catch (KeeperException e) {
log.error("A ZK error has occurred", e);
return;
@@ -555,7 +541,6 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
/**
* Construct the total state view from all sources.
- * Must hold {@link #getUpdateLock()} before calling this.
*
* @param changedCollections collections that have changed since the last call,
* and that should fire notifications
@@ -756,21 +741,18 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
// Can't lock getUpdateLock() until we release the other, it would cause deadlock.
SortedSet<String> oldLiveNodes;
SortedSet<String> newLiveNodes = null;
- updateLock.lock();
- try {
- try {
- List<String> nodeList = zkClient.getChildren(LIVE_NODES_ZKNODE, watcher, true);
- newLiveNodes = new TreeSet<>(nodeList);
- } catch (KeeperException.NoNodeException e) {
- newLiveNodes = emptySortedSet();
- }
- oldLiveNodes = this.liveNodes;
- this.liveNodes = newLiveNodes;
- clusterState.setLiveNodes(newLiveNodes);
- } finally {
- updateLock.unlock();
+ try {
+ List<String> nodeList = zkClient.getChildren(LIVE_NODES_ZKNODE, watcher, true);
+ newLiveNodes = new TreeSet<>(nodeList);
+ } catch (KeeperException.NoNodeException e) {
+ newLiveNodes = emptySortedSet();
}
+
+ oldLiveNodes = this.liveNodes;
+ this.liveNodes = newLiveNodes;
+ clusterState.setLiveNodes(newLiveNodes);
+
if (oldLiveNodes.size() != newLiveNodes.size()) {
if (log.isInfoEnabled()) {
log.info("Updated live nodes from ZooKeeper... ({}) -> ({})", oldLiveNodes.size(), newLiveNodes.size());
@@ -828,10 +810,6 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
return liveNodes;
}
- public ReentrantLock getUpdateLock() {
- return updateLock;
- }
-
public void close() {
log.info("Closing ZkStateReader");
assert closeTracker.close();
@@ -1323,12 +1301,8 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
}
DocCollection newState = fetchCollectionState(coll, this);
updateWatchedCollection(coll, newState);
- // updateLock.lock();
- try {
- constructState(Collections.singleton(coll));
- } finally {
- // updateLock.unlock();
- }
+
+ constructState(Collections.singleton(coll));
} catch (KeeperException e) {
log.error("Unwatched collection: [{}]", coll, e);
@@ -1348,7 +1322,6 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
public void process(WatchedEvent event) {
log.info("_statupdates event {}", event);
try {
- updateLock.lock();
// if (event.getType() == EventType.NodeDataChanged ||
// event.getType() == EventType.NodeDeleted || event.getType() == EventType.NodeCreated) {
@@ -1371,8 +1344,6 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
log.error("Unwatched collection: [{}]", coll, e);
} catch (InterruptedException e) {
log.error("Unwatched collection: [{}]", coll, e);
- } finally {
- updateLock.unlock();
}
}
@@ -1664,17 +1635,10 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
log.error("An error has occurred", e);
return;
}
- // updateLock.lock();
- try {
- constructState(Collections.emptySet());
- } finally {
- // updateLock.unlock();
- }
- }
- /**
- * Must hold {@link #getUpdateLock()} before calling this method.
- */
+ constructState(Collections.emptySet());
+ }
+
public void refreshAndWatch() {
try {
refreshCollectionList(this);
@@ -1840,12 +1804,7 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
});
if (reconstructState.get()) {
- // updateLock.lock();
- try {
- constructState(Collections.emptySet());
- } finally {
- // updateLock.unlock();
- }
+ constructState(Collections.emptySet());
}
}
@@ -2111,12 +2070,7 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
}
if (reconstructState.get()) {
- // updateLock.lock();
- try {
- constructState(Collections.emptySet());
- } finally {
- // updateLock.unlock();
- }
+ constructState(Collections.emptySet());
}
}
@@ -2305,13 +2259,8 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
// called by createClusterStateWatchersAndUpdate()
private void refreshAliases(AliasesManager watcher) throws KeeperException, InterruptedException {
- // updateLock.lock();
- try {
- constructState(Collections.emptySet());
- zkClient.exists(ALIASES, watcher);
- } finally {
- // updateLock.unlock();
- }
+ constructState(Collections.emptySet());
+ zkClient.exists(ALIASES, watcher);
aliasesManager.update();
}
diff --git a/solr/test-framework/src/java/org/apache/solr/cloud/MiniSolrCloudCluster.java b/solr/test-framework/src/java/org/apache/solr/cloud/MiniSolrCloudCluster.java
index 5cd5123..21bc927 100644
--- a/solr/test-framework/src/java/org/apache/solr/cloud/MiniSolrCloudCluster.java
+++ b/solr/test-framework/src/java/org/apache/solr/cloud/MiniSolrCloudCluster.java
@@ -743,7 +743,7 @@ public class MiniSolrCloudCluster {
return solrZkClient;
}
- protected CloudHttp2SolrClient buildSolrClient() {
+ public CloudHttp2SolrClient buildSolrClient() {
// return new CloudHttp2SolrClient.Builder(Collections.singletonList(zkServer.getZkHost()), Optional.of("/solr")).build();
zkStateReader = new ZkStateReader(zkServer.getZkAddress(), 15000, 30000);
zkStateReader.createClusterStateWatchersAndUpdate();