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/13 21:17:23 UTC
[lucene-solr] branch reference_impl_dev updated: @1175 Fix cloud
client / server version reconciliation.
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
The following commit(s) were added to refs/heads/reference_impl_dev by this push:
new 5b3866c @1175 Fix cloud client / server version reconciliation.
5b3866c is described below
commit 5b3866c375651356216c67097d06f815c58f3d67
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Fri Nov 13 15:16:58 2020 -0600
@1175 Fix cloud client / server version reconciliation.
---
.../java/org/apache/solr/servlet/HttpSolrCall.java | 23 ++++++-----
.../client/solrj/impl/BaseCloudSolrClient.java | 45 ++++++++++++++--------
.../solr/client/solrj/impl/CloudSolrClient.java | 2 -
.../client/solrj/io/stream/ExecutorStream.java | 2 +-
4 files changed, 44 insertions(+), 28 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java b/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java
index fe54695..f5fea31 100644
--- a/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java
+++ b/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java
@@ -343,26 +343,27 @@ public class HttpSolrCall {
private void ensureStatesAreAtLeastAtClient() throws InterruptedException, TimeoutException {
if (cores.isZooKeeperAware()) {
- log.info("ensure states are at at least client version");
- invalidStates = checkStateVersionsAreValid(solrReq.getParams().get(CloudSolrClient.STATE_VERSION));
+ log.info("State version for request is {}", queryParams.get(CloudSolrClient.STATE_VERSION));
+ invalidStates = checkStateVersionsAreValid(queryParams.get(CloudSolrClient.STATE_VERSION));
if (invalidStates != null) {
Set<Map.Entry<String,Integer>> entries = invalidStates.entrySet();
for (Map.Entry<String,Integer> entry : entries) {
String collection = entry.getKey();
Integer version = entry.getValue();
-
+ log.info("ensure states are at at least client version {} for collection {}", version, collection);
if (cores.getZkController().getZkStateReader().watched(collection)) {
cores.getZkController().getZkStateReader().waitForState(collection, 5, TimeUnit.SECONDS, (liveNodes, collectionState) -> {
if (collectionState == null) {
return false;
}
+ log.info("found server state version {}", collectionState.getZNodeVersion());
if (collectionState.getZNodeVersion() < version) {
return false;
}
return true;
});
}
- }
+ }
}
}
}
@@ -456,8 +457,11 @@ public class HttpSolrCall {
}
}
- protected void extractRemotePath(String collectionName, String origCorename) throws UnsupportedEncodingException, KeeperException, InterruptedException, SolrException {
+ protected void extractRemotePath(String collectionName, String origCorename) throws UnsupportedEncodingException, KeeperException, InterruptedException, SolrException, TimeoutException {
assert core == null;
+
+ ensureStatesAreAtLeastAtClient();
+
coreUrl = getRemoteCoreUrl(collectionName, origCorename);
// don't proxy for internal update requests
invalidStates = checkStateVersionsAreValid(queryParams.get(CloudSolrClient.STATE_VERSION));
@@ -988,14 +992,15 @@ public class HttpSolrCall {
for (String pair : pairs) {
String[] pcs = StringUtils.split(pair, ':');
if (pcs.length == 2 && !pcs[0].isEmpty() && !pcs[1].isEmpty()) {
- Integer status = cores.getZkController().getZkStateReader().compareStateVersions(pcs[0], Integer.parseInt(pcs[1]));
- if (status != null) {
+ log.info("compare version states {} {}", pcs[0], Integer.parseInt(pcs[1]));
+
if (result == null) result = new HashMap<>();
- result.put(pcs[0], status);
- }
+ result.put(pcs[0], Integer.parseInt(pcs[1]));
+
}
}
}
+ log.info("compare version states result {} {}", stateVer, result);
return result;
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java
index 5c2e36e..99c02d0 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java
@@ -907,24 +907,23 @@ public abstract class BaseCloudSolrClient extends SolrClient {
}
StringBuilder stateVerParamBuilder = null;
+ log.info("build version params for collections {}", requestedCollectionNames);
for (String requestedCollection : requestedCollectionNames) {
// track the version of state we're using on the client side using the _stateVer_ param
- DocCollection coll = getDocCollection(requestedCollection, null);
- if (coll == null) {
- continue;
- }
-
- int collVer = coll.getZNodeVersion();
- if(requestedCollections == null) requestedCollections = new ArrayList<>(requestedCollectionNames.size());
- requestedCollections.add(coll);
+ DocCollection coll = getZkStateReader().getClusterState().getCollectionOrNull(requestedCollection);
+ if (coll != null) {
+ int collVer = coll.getZNodeVersion();
+ if (requestedCollections == null) requestedCollections = new ArrayList<>(requestedCollectionNames.size());
+ requestedCollections.add(coll);
+
+ if (stateVerParamBuilder == null) {
+ stateVerParamBuilder = new StringBuilder();
+ } else {
+ stateVerParamBuilder.append("|"); // hopefully pipe is not an allowed char in a collection name
+ }
- if (stateVerParamBuilder == null) {
- stateVerParamBuilder = new StringBuilder();
- } else {
- stateVerParamBuilder.append("|"); // hopefully pipe is not an allowed char in a collection name
+ stateVerParamBuilder.append(coll.getName()).append(":").append(collVer);
}
-
- stateVerParamBuilder.append(coll.getName()).append(":").append(collVer);
}
if (stateVerParamBuilder != null) {
@@ -937,10 +936,19 @@ public abstract class BaseCloudSolrClient extends SolrClient {
if (stateVerParam != null) {
params.set(STATE_VERSION, stateVerParam);
} else {
- params.remove(STATE_VERSION);
+ // params.remove(STATE_VERSION);
}
- } // else: ??? how to set this ???
+ } else {
+ ModifiableSolrParams sp = new ModifiableSolrParams(request.getParams());
+ if (stateVerParam != null) {
+ sp.set(STATE_VERSION, stateVerParam);
+ } else {
+ //sp.remove(STATE_VERSION);
+ }
+ }
+
+// log.info("state version param {}", request.getParams().get(STATE_VERSION));
NamedList<Object> resp = null;
try {
resp = sendRequest(request, inputCollections);
@@ -1116,6 +1124,11 @@ public abstract class BaseCloudSolrClient extends SolrClient {
protected NamedList<Object> sendRequest(SolrRequest request, List<String> inputCollections)
throws SolrServerException, IOException {
+
+// if (request.getParams().get(STATE_VERSION) == null) {
+// throw new IllegalStateException("State version cannot be null " + request.getParams());
+// }
+
connect();
boolean sendToLeaders = false;
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
index 77c491a..6895a6a 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
@@ -57,8 +57,6 @@ public class CloudSolrClient extends BaseCloudSolrClient {
private final HttpClient myClient;
private final boolean clientIsInternal;
- public static final String STATE_VERSION = BaseCloudSolrClient.STATE_VERSION;
-
/**
* @deprecated since 7.0 Use {@link Builder} methods instead.
*/
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExecutorStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExecutorStream.java
index 0a73276..bb4f7de 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExecutorStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExecutorStream.java
@@ -133,7 +133,7 @@ public class ExecutorStream extends TupleStream implements Expressible {
}
public void open() throws IOException {
- executorService = ParWork.getMyPerThreadExecutor();
+ executorService = ParWork.getRootSharedExecutor();
stream.open();
}