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 2021/01/21 21:33:02 UTC
[lucene-solr] branch reference_impl_dev updated (8631636 -> 561cdac)
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a change to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git.
from 8631636 @1270 Wait to see published down state before create live node.
new b8945de @1271 Recovery loop fix.
new 3950d42 @1272 Start working in load fail safe correctly.
new 26e2cf7 @1273 Remove exploratory fix that was not the love.
new 72b8db1 @1274 Recovery loop fix 2.
new 561cdac @1275 Fix wait for down and some lock and recovery issues.
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/solr/cloud/RecoveryStrategy.java | 127 ++++++++++++---------
.../solr/cloud/ShardLeaderElectionContext.java | 2 +
.../solr/cloud/ShardLeaderElectionContextBase.java | 9 +-
.../java/org/apache/solr/cloud/StatePublisher.java | 2 +-
.../java/org/apache/solr/cloud/ZkController.java | 10 +-
.../java/org/apache/solr/core/CoreContainer.java | 49 ++++----
.../src/java/org/apache/solr/core/SolrCore.java | 11 +-
.../java/org/apache/solr/handler/IndexFetcher.java | 1 +
.../apache/solr/handler/admin/PrepRecoveryOp.java | 11 +-
.../handler/component/RealTimeGetComponent.java | 3 -
.../org/apache/solr/servlet/SolrQoSFilter.java | 53 ++++++---
.../apache/solr/update/DefaultSolrCoreState.java | 18 ++-
.../java/org/apache/solr/update/SolrCoreState.java | 2 +-
.../org/apache/solr/update/TimedVersionBucket.java | 2 +-
.../processor/DistributedZkUpdateProcessor.java | 2 +-
solr/reference-branch/bench/solr-benchmark.jar | Bin 37946 -> 37932 bytes
.../apache/solr/common/cloud/ZkStateReader.java | 2 +-
.../java/org/apache/solr/common/util/SysStats.java | 19 ++-
18 files changed, 184 insertions(+), 139 deletions(-)
[lucene-solr] 01/05: @1271 Recovery loop fix.
Posted by ma...@apache.org.
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 b8945ded1e55f2d349ae5f478fa5fb82f29c5605
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Thu Jan 21 10:43:38 2021 -0600
@1271 Recovery loop fix.
---
.../org/apache/solr/cloud/RecoveryStrategy.java | 43 ++++++++++++----------
.../apache/solr/handler/admin/PrepRecoveryOp.java | 9 +++--
.../apache/solr/common/cloud/ZkStateReader.java | 2 +-
3 files changed, 29 insertions(+), 25 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
index 2ecf60b..1c4c53b 100644
--- a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
+++ b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
@@ -322,7 +322,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
replicationHandler = (ReplicationHandler) handler;
doRecovery(core);
- }
+ }
} catch (InterruptedException e) {
log.info("InterruptedException, won't do recovery", e);
return;
@@ -340,24 +340,25 @@ public class RecoveryStrategy implements Runnable, Closeable {
while (!isClosed()) {
try {
try {
- Replica leader = zkController.getZkStateReader().getLeaderRetry(coreDescriptor.getCollectionName(), coreDescriptor.getCloudDescriptor().getShardId(), 1500);
- if (leader != null && leader.getName().equals(coreName) && zkController.getZkClient().exists(COLLECTIONS_ZKNODE + "/" + coreDescriptor.getCollectionName() + "/leaders/" + coreDescriptor.getCloudDescriptor().getShardId() + "/leader")) {
- log.info("We are the leader, STOP recovery");
- close = true;
- return;
- }
- if (core.isClosing() || core.getCoreContainer().isShutDown()) {
- log.info("We are closing, STOP recovery");
- close = true;
- return;
+ if (prevSendPreRecoveryHttpUriRequest != null) {
+ prevSendPreRecoveryHttpUriRequest.cancel();
}
- } catch (InterruptedException e) {
- log.info("InterruptedException", e);
- throw new SolrException(ErrorCode.BAD_REQUEST, e);
- } catch (TimeoutException e) {
- log.info("Timeout waiting for leader", e);
- continue;
- // throw new SolrException(ErrorCode.SERVER_ERROR, e);
+ prevSendPreRecoveryHttpUriRequest = null;
+ } catch (NullPointerException e) {
+ // expected
+ }
+
+ Replica leader = zkController.getZkStateReader().getLeaderRetry(coreDescriptor.getCollectionName(), coreDescriptor.getCloudDescriptor().getShardId(), 1500);
+ if (leader != null && leader.getName().equals(coreName) && zkController.getZkClient()
+ .exists(COLLECTIONS_ZKNODE + "/" + coreDescriptor.getCollectionName() + "/leaders/" + coreDescriptor.getCloudDescriptor().getShardId() + "/leader")) {
+ log.info("We are the leader, STOP recovery");
+ close = true;
+ return;
+ }
+ if (core.isClosing() || core.getCoreContainer().isShutDown()) {
+ log.info("We are closing, STOP recovery");
+ close = true;
+ return;
}
if (this.coreDescriptor.getCloudDescriptor().requiresTransactionLog()) {
@@ -368,6 +369,8 @@ public class RecoveryStrategy implements Runnable, Closeable {
doReplicateOnlyRecovery(core);
}
+ break;
+
} catch (Exception e) {
log.info("Exception trying to recover, try again", e);
}
@@ -891,7 +894,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
log.info("Sending prep recovery command to {} for core {} params={}", leaderBaseUrl, leaderCoreName, prepCmd.getParams());
int conflictWaitMs = zkController.getLeaderConflictResolveWait();
- int readTimeout = conflictWaitMs + Integer.parseInt(System.getProperty("prepRecoveryReadTimeoutExtraWait", "15000"));
+ int readTimeout = conflictWaitMs + Integer.parseInt(System.getProperty("prepRecoveryReadTimeoutExtraWait", "5000"));
// nocommit
try (Http2SolrClient client = new Http2SolrClient.Builder(leaderBaseUrl).withHttpClient(cc.getUpdateShardHandler().
getRecoveryOnlyClient()).idleTimeout(readTimeout).markInternalRequest().build()) {
@@ -903,7 +906,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
try {
prevSendPreRecoveryHttpUriRequest = result;
try {
- boolean success = latch.await(15, TimeUnit.SECONDS);
+ boolean success = latch.await(5, TimeUnit.SECONDS);
if (!success) {
result.cancel();
throw new SolrException(ErrorCode.SERVICE_UNAVAILABLE, "Timeout waiting for prep recovery cmd on leader");
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java b/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java
index b31795a..45ef666 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java
@@ -19,6 +19,7 @@ package org.apache.solr.handler.admin;
import org.apache.solr.cloud.ZkController.NotInClusterStateException;
import org.apache.solr.cloud.ZkShardTerms;
+import org.apache.solr.common.ParWork;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.SolrZkClient;
@@ -80,6 +81,7 @@ class PrepRecoveryOp implements CoreAdminHandler.CoreAdminOp {
if (replica != null) {
if (replica.getState() == waitForState || replica.getState() == Replica.State.ACTIVE && coreContainer.getZkController().getZkStateReader().isNodeLive(replica.getNodeName())) {
+ // if (log.isDebugEnabled()) log.debug("replica={} state={} waitForState={}", replica, replica.getState(), waitForState);
log.info("replica={} state={} waitForState={}", replica, replica.getState(), waitForState);
return true;
}
@@ -89,7 +91,7 @@ class PrepRecoveryOp implements CoreAdminHandler.CoreAdminOp {
});
} catch (TimeoutException | InterruptedException e) {
- SolrZkClient.checkInterrupted(e);
+ ParWork.propagateInterrupt(e);
String error = errorMessage.get();
if (error == null)
error = "Timeout waiting for collection state. \n" + coreContainer.getZkController().getZkStateReader().getClusterState().getCollectionOrNull(collection);
@@ -106,9 +108,8 @@ class PrepRecoveryOp implements CoreAdminHandler.CoreAdminOp {
log.info("refresh shard terms for core {}", cname);
shardTerms.refreshTerms(false);
}
- } catch (NullPointerException e) {
- if (log.isDebugEnabled()) log.debug("No shards found", e);
- // likely deleted shard/collection
+ } catch (Exception e) {
+ log.error("Exception while looking at refreshing shard terms", e);
}
}
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 ff87426..a73bb23 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
@@ -2035,7 +2035,7 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
waitLatches.add(latch);
AtomicReference<DocCollection> docCollection = new AtomicReference<>();
CollectionStateWatcher watcher = (n, c) -> {
- if (isClosed()) return true;
+ // if (isClosed()) return true;
docCollection.set(c);
boolean matches = predicate.matches(this.liveNodes, c);
if (matches)
[lucene-solr] 03/05: @1273 Remove exploratory fix that was not the
love.
Posted by ma...@apache.org.
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 26e2cf7a5e2f225eec411146a1ee62d632fdf3eb
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Thu Jan 21 10:47:03 2021 -0600
@1273 Remove exploratory fix that was not the love.
---
solr/core/src/java/org/apache/solr/core/SolrCore.java | 9 ---------
1 file changed, 9 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/core/SolrCore.java b/solr/core/src/java/org/apache/solr/core/SolrCore.java
index e363319..8d06233 100644
--- a/solr/core/src/java/org/apache/solr/core/SolrCore.java
+++ b/solr/core/src/java/org/apache/solr/core/SolrCore.java
@@ -717,12 +717,6 @@ public final class SolrCore implements SolrInfoBean, Closeable {
if (coreContainer.isShutDown() || isClosed() || closing) {
throw new AlreadyClosedException();
}
- ReentrantLock schemaLock = null;
- if (schema instanceof ManagedIndexSchema) {
- schemaLock = ((ManagedIndexSchema) schema).getSchemaLock();
- schemaLock.lock();
- }
-
// only one reload at a time
ReentrantLock lock = getUpdateHandler().getSolrCoreState().getReloadLock();
boolean locked = lock.tryLock();
@@ -808,9 +802,6 @@ public final class SolrCore implements SolrInfoBean, Closeable {
if (lock != null && lock.isHeldByCurrentThread()) {
lock.unlock();
}
- if (schemaLock != null && schemaLock.isHeldByCurrentThread()) {
- schemaLock.unlock();
- }
}
}
[lucene-solr] 04/05: @1274 Recovery loop fix 2.
Posted by ma...@apache.org.
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 72b8db12062dcb07755643d985bd38a4c7614951
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Thu Jan 21 11:06:44 2021 -0600
@1274 Recovery loop fix 2.
---
.../java/org/apache/solr/cloud/RecoveryStrategy.java | 20 +++++++++++++-------
1 file changed, 13 insertions(+), 7 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
index 1c4c53b..32aa54b 100644
--- a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
+++ b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
@@ -360,16 +360,18 @@ public class RecoveryStrategy implements Runnable, Closeable {
close = true;
return;
}
-
+ boolean successfulRecovery;
if (this.coreDescriptor.getCloudDescriptor().requiresTransactionLog()) {
if (log.isDebugEnabled()) log.debug("Sync or replica recovery");
- doSyncOrReplicateRecovery(core);
+ successfulRecovery = doSyncOrReplicateRecovery(core);
} else {
if (log.isDebugEnabled()) log.debug("Replicate only recovery");
- doReplicateOnlyRecovery(core);
+ successfulRecovery = doReplicateOnlyRecovery(core);
}
- break;
+ if (successfulRecovery) {
+ break;
+ }
} catch (Exception e) {
log.info("Exception trying to recover, try again", e);
@@ -377,7 +379,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
}
}
- final private void doReplicateOnlyRecovery(SolrCore core) throws Exception {
+ final private boolean doReplicateOnlyRecovery(SolrCore core) throws Exception {
boolean successfulRecovery = false;
// if (core.getUpdateHandler().getUpdateLog() != null) {
@@ -485,10 +487,12 @@ public class RecoveryStrategy implements Runnable, Closeable {
}
log.info("Finished recovery process, successful=[{}]", successfulRecovery);
+
+ return successfulRecovery;
}
// TODO: perhaps make this grab a new core each time through the loop to handle core reloads?
- public final void doSyncOrReplicateRecovery(SolrCore core) throws Exception {
+ public final boolean doSyncOrReplicateRecovery(SolrCore core) throws Exception {
log.info("Do peersync or replication recovery core={} collection={}", coreName, coreDescriptor.getCollectionName());
Replica leader = zkController.getZkStateReader().getLeaderRetry(coreDescriptor.getCollectionName(), coreDescriptor.getCloudDescriptor().getShardId(), 1500);
@@ -511,7 +515,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
SolrException.log(log, "No UpdateLog found - cannot recover.");
close = true;
recoveryFailed(zkController, baseUrl, this.coreDescriptor);
- return;
+ return false;
}
// we temporary ignore peersync for tlog replicas
@@ -767,6 +771,8 @@ public class RecoveryStrategy implements Runnable, Closeable {
log.error("Illegal state, successful recovery, but did not publish active");
throw new SolrException(ErrorCode.SERVER_ERROR, "Illegal state, successful recovery, but did not publish active");
}
+
+ return successfulRecovery;
}
private final void waitForRetry() {
[lucene-solr] 05/05: @1275 Fix wait for down and some lock and
recovery issues.
Posted by ma...@apache.org.
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 561cdac5214c9927c8c1f2adeae2da3d0c5e0c26
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Thu Jan 21 15:17:01 2021 -0600
@1275 Fix wait for down and some lock and recovery issues.
---
.../org/apache/solr/cloud/RecoveryStrategy.java | 68 +++++++++++++---------
.../solr/cloud/ShardLeaderElectionContext.java | 2 +
.../solr/cloud/ShardLeaderElectionContextBase.java | 9 ++-
.../java/org/apache/solr/cloud/StatePublisher.java | 2 +-
.../java/org/apache/solr/cloud/ZkController.java | 10 +++-
.../java/org/apache/solr/core/CoreContainer.java | 49 ++++++++--------
.../src/java/org/apache/solr/core/SolrCore.java | 2 +-
.../java/org/apache/solr/handler/IndexFetcher.java | 1 +
.../apache/solr/handler/admin/PrepRecoveryOp.java | 2 +-
.../handler/component/RealTimeGetComponent.java | 3 -
.../apache/solr/update/DefaultSolrCoreState.java | 18 +++---
.../java/org/apache/solr/update/SolrCoreState.java | 2 +-
.../org/apache/solr/update/TimedVersionBucket.java | 2 +-
.../processor/DistributedZkUpdateProcessor.java | 2 +-
14 files changed, 96 insertions(+), 76 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
index 32aa54b..1272ad8 100644
--- a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
+++ b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
@@ -337,7 +337,9 @@ public class RecoveryStrategy implements Runnable, Closeable {
}
final public void doRecovery(SolrCore core) throws Exception {
+ int tries = 0;
while (!isClosed()) {
+ tries++;
try {
try {
if (prevSendPreRecoveryHttpUriRequest != null) {
@@ -370,11 +372,14 @@ public class RecoveryStrategy implements Runnable, Closeable {
}
if (successfulRecovery) {
+ close = true;
break;
+ } else {
+ log.info("Trying another loop to recover after failing try={}", tries);
}
} catch (Exception e) {
- log.info("Exception trying to recover, try again", e);
+ log.info("Exception trying to recover, try again try={}", tries, e);
}
}
}
@@ -593,10 +598,6 @@ public class RecoveryStrategy implements Runnable, Closeable {
CloudDescriptor cloudDesc = this.coreDescriptor.getCloudDescriptor();
leader = zkStateReader.getLeaderRetry(cloudDesc.getCollectionName(), cloudDesc.getShardId(), 1500);
- if (isClosed()) {
- throw new AlreadyClosedException();
- }
-
log.info("Begin buffering updates. core=[{}]", coreName);
// recalling buffer updates will drop the old buffer tlog
ulog.bufferUpdates();
@@ -632,26 +633,34 @@ public class RecoveryStrategy implements Runnable, Closeable {
// System.out.println("Attempting to PeerSync from " + leaderUrl
// + " i am:" + zkController.getNodeName());
- boolean syncSuccess;
- try (PeerSyncWithLeader peerSyncWithLeader = new PeerSyncWithLeader(core, leader.getCoreUrl(), ulog.getNumRecordsToKeep())) {
- syncSuccess = peerSyncWithLeader.sync(recentVersions).isSuccess();
- }
- if (syncSuccess) {
- SolrQueryRequest req = new LocalSolrQueryRequest(core, new ModifiableSolrParams());
- log.info("PeerSync was successful, commit to force open a new searcher");
- // force open a new searcher
- core.getUpdateHandler().commit(new CommitUpdateCommand(req, false));
- req.close();
- log.info("PeerSync stage of recovery was successful.");
-
- // solrcloud_debug
- // cloudDebugLog(core, "synced");
-
- log.info("Replaying updates buffered during PeerSync.");
- replay(core);
+ try {
+ boolean syncSuccess;
+ try (PeerSyncWithLeader peerSyncWithLeader = new PeerSyncWithLeader(core, leader.getCoreUrl(), ulog.getNumRecordsToKeep())) {
+ syncSuccess = peerSyncWithLeader.sync(recentVersions).isSuccess();
+ }
+ if (syncSuccess) {
+ SolrQueryRequest req = new LocalSolrQueryRequest(core, new ModifiableSolrParams());
+ log.info("PeerSync was successful, commit to force open a new searcher");
+ // force open a new searcher
+ core.getUpdateHandler().commit(new CommitUpdateCommand(req, false));
+ req.close();
+ log.info("PeerSync stage of recovery was successful.");
+
+ // solrcloud_debug
+ // cloudDebugLog(core, "synced");
+
+ log.info("Replaying updates buffered during PeerSync.");
+ replay(core);
+
+ // sync success
+ successfulRecovery = true;
+ } else {
+ successfulRecovery = false;
+ }
- // sync success
- successfulRecovery = true;
+ } catch (Exception e) {
+ log.error("PeerSync exception", e);
+ successfulRecovery = false;
}
if (!successfulRecovery) {
@@ -681,6 +690,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
log.info("Interrupted or already closed, bailing on recovery");
close = true;
successfulRecovery = false;
+ break;
} catch (Exception e) {
successfulRecovery = false;
log.error("Error while trying to recover", e);
@@ -710,14 +720,16 @@ public class RecoveryStrategy implements Runnable, Closeable {
zkController.publish(this.coreDescriptor, Replica.State.ACTIVE);
publishedActive = true;
+ close = true;
} catch (AlreadyClosedException e) {
log.error("Already closed");
successfulRecovery = false;
+ close = true;
} catch (Exception e) {
log.error("Could not publish as ACTIVE after successful recovery", e);
successfulRecovery = false;
- // core.getSolrCoreState().doRecovery(core);
+ close = false;
}
@@ -731,7 +743,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
}
- if (!successfulRecovery) {
+ if (!successfulRecovery && !isClosed()) {
// lets pause for a moment and we need to try again...
// TODO: we don't want to retry for some problems?
// Or do a fall off retry...
@@ -754,9 +766,9 @@ public class RecoveryStrategy implements Runnable, Closeable {
}
}
- if (!successfulRecovery) {
+ if (!successfulRecovery && !isClosed()) {
waitForRetry();
- } else {
+ } else if (successfulRecovery) {
break;
}
}
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 8256084..56eb674 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java
@@ -319,6 +319,8 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
log.info("There may be a better leader candidate than us - will cancel election, rejoin election, and kick off recovery");
leaderElector.retryElection(false);
+
+ core.getSolrCoreState().doRecovery(core);
}
public String getShardId() {
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 b11c180..abaee54 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java
@@ -181,11 +181,16 @@ class ShardLeaderElectionContextBase extends ElectionContext {
//assert leaderZkNodeParentVersion != null;
} catch (NoNodeException e) {
+ log.warn("No node exists for election", e);
throw new AlreadyClosedException("No node exists for election");
} catch (KeeperException.NodeExistsException e) {
- throw new AlreadyClosedException("Node already exists for election");
+ log.warn("Node already exists for election", e);
+
+ zkClient.delete(leaderPath, -1);
+
+ runLeaderProcess(context, weAreReplacement, pauseBeforeStartMs);
} catch (Throwable t) {
- ParWork.propagateInterrupt(t);
+ log.warn("Could not register as the leader because creating the ephemeral registration node in ZooKeeper failed: ", t);
throw new SolrException(ErrorCode.SERVER_ERROR, "Could not register as the leader because creating the ephemeral registration node in ZooKeeper failed: " + errors, t);
}
return true;
diff --git a/solr/core/src/java/org/apache/solr/cloud/StatePublisher.java b/solr/core/src/java/org/apache/solr/cloud/StatePublisher.java
index 341ffa3..45d5fda 100644
--- a/solr/core/src/java/org/apache/solr/cloud/StatePublisher.java
+++ b/solr/core/src/java/org/apache/solr/cloud/StatePublisher.java
@@ -148,7 +148,7 @@ public class StatePublisher implements Closeable {
Replica replica = zkStateReader.getClusterState().getCollection(collection).getReplica(core);
String lastState = stateCache.get(core);
// nocommit
- if (collection != null && replica != null && state.equals(lastState) && replica.getState().toString().equals(state)) {
+ if (collection != null && replica != null && !state.equals(Replica.State.ACTIVE) && state.equals(lastState) && replica.getState().toString().equals(state)) {
log.info("Skipping publish state as {} for {}, because it was the last state published", state, core);
// nocommit
return;
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 8ebdfe9..c4e9db6 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -417,6 +417,14 @@ public class ZkController implements Closeable, Runnable {
started = true;
+ try {
+ if (zkClient.exists( ZkStateReader.LIVE_NODES_ZKNODE + "/" + getNodeName())) {
+ removeEphemeralLiveNode();
+ }
+ } catch (Exception e) {
+ ParWork.propagateInterrupt("Error Removing ephemeral live node. Continuing to close CoreContainer", e);
+ }
+
this.overseer = new Overseer(cc.getUpdateShardHandler(), CommonParams.CORES_HANDLER_PATH, this, cloudConfig);
try {
this.overseerRunningMap = Overseer.getRunningMap(zkClient);
@@ -1407,7 +1415,7 @@ public class ZkController implements Closeable, Runnable {
throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
}
- if (log.isDebugEnabled()) log.debug("Wait to see leader for {}, {}", collection, shardId);
+ log.info("Wait to see leader for {}, {}", collection, shardId);
Replica leader = null;
for (int i = 0; i < 30; i++) {
// if (leaderElector.isLeader()) {
diff --git a/solr/core/src/java/org/apache/solr/core/CoreContainer.java b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
index 61566b6..f7be9a1 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -28,6 +28,7 @@ import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.store.Directory;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
+import org.apache.solr.client.solrj.embedded.EmbeddedSolrServer;
import org.apache.solr.client.solrj.impl.CloudHttp2SolrClient;
import org.apache.solr.client.solrj.impl.HttpClientUtil;
import org.apache.solr.client.solrj.impl.SolrHttpClientBuilder;
@@ -885,6 +886,27 @@ public class CoreContainer implements Closeable {
status |= CORE_DISCOVERY_COMPLETE;
for (final CoreDescriptor cd : cds) {
+ if (isZooKeeperAware()) {
+ String collection = cd.getCollectionName();
+ try {
+ zkSys.zkController.zkStateReader.waitForState(collection, 5, TimeUnit.SECONDS, (n, c) -> {
+ if (c != null) {
+ Replica replica = c.getReplica(cd.getName());
+
+ if (replica.getState().equals(State.DOWN)) {
+ return true;
+ }
+
+ }
+ return false;
+ });
+ } catch (InterruptedException e) {
+ ParWork.propagateInterrupt(e);
+ } catch (TimeoutException e) {
+ log.error("Timeout", e);
+ }
+ }
+
if (log.isDebugEnabled()) log.debug("Process core descriptor {} {} {}", cd.getName(), cd.isTransient(), cd.isLoadOnStartup());
if (cd.isTransient() || !cd.isLoadOnStartup()) {
solrCores.addCoreDescriptor(cd);
@@ -914,32 +936,9 @@ public class CoreContainer implements Closeable {
}));
}
}
- if (zkSys != null && zkSys.getZkController() != null) {
+ if (isZooKeeperAware()) {
+
ParWork.getRootSharedExecutor().submit(() -> {
- Collection<SolrCore> cores = getCores(); // TODO use the cores we just launched, this may not be populated yet
- for (SolrCore core : cores) {
- CoreDescriptor desc = core.getCoreDescriptor();
- String collection = desc.getCollectionName();
- try {
- zkSys.zkController.zkStateReader.waitForState(collection, 5, TimeUnit.SECONDS, (n, c) -> {
- if (c != null) {
- List<Replica> replicas = c.getReplicas();
- for (Replica replica : replicas) {
- if (replica.getNodeName().equals(zkSys.zkController.getNodeName())) {
- if (!replica.getState().equals(Replica.State.DOWN)) {
- return false;
- }
- }
- }
- }
- return true;
- });
- } catch (InterruptedException e) {
- ParWork.propagateInterrupt(e);
- } catch (TimeoutException e) {
- log.error("Timeout", e);
- }
- }
zkSys.getZkController().createEphemeralLiveNode();
});
}
diff --git a/solr/core/src/java/org/apache/solr/core/SolrCore.java b/solr/core/src/java/org/apache/solr/core/SolrCore.java
index 8d06233..29bc5bc 100644
--- a/solr/core/src/java/org/apache/solr/core/SolrCore.java
+++ b/solr/core/src/java/org/apache/solr/core/SolrCore.java
@@ -748,7 +748,7 @@ public final class SolrCore implements SolrInfoBean, Closeable {
// another recovery waiting behind us, let it run now instead of after we finish
log.info("Skipping reload because there is another in line behind");
lock.unlock();
- Thread.sleep(50);
+ Thread.sleep(10);
lock.lock();
lock.unlock();
return null;
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 ab4821b..972d012 100644
--- a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
+++ b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
@@ -1108,6 +1108,7 @@ public class IndexFetcher {
throw e;
} catch (Exception e) {
log.error("Problem downloading file {}", file, e);
+ throw e;
} finally {
fileFetchRequests.remove(file.get(NAME));
}
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java b/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java
index 45ef666..f83075d 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java
@@ -80,7 +80,7 @@ class PrepRecoveryOp implements CoreAdminHandler.CoreAdminOp {
final Replica replica = c.getReplica(cname);
if (replica != null) {
- if (replica.getState() == waitForState || replica.getState() == Replica.State.ACTIVE && coreContainer.getZkController().getZkStateReader().isNodeLive(replica.getNodeName())) {
+ if (replica.getState() == waitForState && coreContainer.getZkController().getZkStateReader().isNodeLive(replica.getNodeName())) {
// if (log.isDebugEnabled()) log.debug("replica={} state={} waitForState={}", replica, replica.getState(), waitForState);
log.info("replica={} state={} waitForState={}", replica, replica.getState(), waitForState);
return true;
diff --git a/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java b/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
index a7d71d1..63eff4b 100644
--- a/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
+++ b/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
@@ -1144,9 +1144,6 @@ public class RealTimeGetComponent extends SearchComponent
// TODO: more complex response?
rb.rsp.add("sync", success);
- if (!success && rb.req.getCore().getCoreContainer().isZooKeeperAware() && !rb.req.getCore().getSolrCoreState().isRecoverying()) {
- rb.req.getCore().getSolrCoreState().doRecovery(rb.req.getCore().getCoreContainer(), rb.req.getCore().getCoreDescriptor());
- }
} catch (IOException e) {
log.error("Error while closing", e);
}
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 475abe4..7c1f4f5 100644
--- a/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
+++ b/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
@@ -52,7 +52,7 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
private final boolean SKIP_AUTO_RECOVERY = Boolean.getBoolean("solrcloud.skip.autorecovery");
- private final ReentrantLock recoveryLock = new ReentrantLock(true);
+ private final ReentrantLock recoveryLock = new ReentrantLock(false);
private final ActionThrottle recoveryThrottle = new ActionThrottle("recovery", Integer.getInteger("solr.recoveryThrottle", 0));
@@ -190,7 +190,7 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
boolean acquired = false;
do {
try {
- acquired = lock.tryLock(100, TimeUnit.MILLISECONDS);
+ acquired = lock.tryLock() || lock.tryLock(100, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
log.warn("WARNING - Dangerous interrupt", e);
}
@@ -346,25 +346,19 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
return;
}
-
// if we can't get the lock, another recovery is running
// we check to see if there is already one waiting to go
// after the current one, and if there is, bail
boolean locked = recoveryLock.tryLock();
- if (!locked && recoveryWaiting.get() > 1) {
- log.info("Skipping recovery because there is another already queued");
- return;
- }
-
// if (closed || prepForClose) {
// return;
// }
if (!locked) {
recoveryWaiting.incrementAndGet();
if (log.isDebugEnabled()) log.debug("Wait for recovery lock");
- cancelRecovery();
- while (!recoveryLock.tryLock(250, TimeUnit.MILLISECONDS)) {
+ cancelRecovery(true, false);
+ while (!(recoveryLock.tryLock() || recoveryLock.tryLock(500, TimeUnit.MILLISECONDS))) {
if (closed || prepForClose) {
log.warn("Skipping recovery because we are closed");
recoveryWaiting.decrementAndGet();
@@ -379,12 +373,13 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
}
}
+ recoverying = true;
+
// to be air tight we must also check after lock
if (prepForClose || closed || corecontainer.isShutDown()) {
log.info("Skipping recovery due to being closed");
return;
}
- recoverying = true;
recoveryThrottle.minimumWaitBetweenActions();
recoveryThrottle.markAttemptingAction();
@@ -435,6 +430,7 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
@Override
public void cancelRecovery(boolean wait, boolean prepForClose) {
+ log.info("Cancel recovery");
recoverying = false;
if (prepForClose) {
diff --git a/solr/core/src/java/org/apache/solr/update/SolrCoreState.java b/solr/core/src/java/org/apache/solr/update/SolrCoreState.java
index 556ac75..c53fe21 100644
--- a/solr/core/src/java/org/apache/solr/update/SolrCoreState.java
+++ b/solr/core/src/java/org/apache/solr/update/SolrCoreState.java
@@ -44,7 +44,7 @@ public abstract class SolrCoreState {
protected volatile boolean closed = false;
private final Object updateLock = new Object();
- private final ReentrantLock reloadLock = new ReentrantLock(true);
+ private final ReentrantLock reloadLock = new ReentrantLock(false);
public Object getUpdateLock() {
return updateLock;
diff --git a/solr/core/src/java/org/apache/solr/update/TimedVersionBucket.java b/solr/core/src/java/org/apache/solr/update/TimedVersionBucket.java
index 94e7551..abeca02 100644
--- a/solr/core/src/java/org/apache/solr/update/TimedVersionBucket.java
+++ b/solr/core/src/java/org/apache/solr/update/TimedVersionBucket.java
@@ -44,7 +44,7 @@ public class TimedVersionBucket extends VersionBucket {
boolean success = false;
try {
- success = lock.tryLock(lockTimeoutMs, TimeUnit.MILLISECONDS);
+ success = lock.tryLock() || lock.tryLock(lockTimeoutMs, TimeUnit.MILLISECONDS);
if (success) {
return function.apply();
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 7ec4dd6..7e8faeb 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
@@ -1036,7 +1036,7 @@ public class DistributedZkUpdateProcessor extends DistributedUpdateProcessor {
ReentrantLock ruleExpiryLock = req.getCore().getRuleExpiryLock();
if (!ruleExpiryLock.isLocked()) {
try {
- if (ruleExpiryLock.tryLock(10, TimeUnit.MILLISECONDS)) {
+ if (ruleExpiryLock.tryLock() || ruleExpiryLock.tryLock(10, TimeUnit.MILLISECONDS)) {
log.info("Going to expire routing rule");
try {
// nocommit TODO: needs to use the statepublisher
[lucene-solr] 02/05: @1272 Start working in load fail safe
correctly.
Posted by ma...@apache.org.
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 3950d42d1e585c3c772b071357212036b95bdc8f
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Thu Jan 21 10:44:14 2021 -0600
@1272 Start working in load fail safe correctly.
---
.../org/apache/solr/servlet/SolrQoSFilter.java | 53 +++++++++++++--------
solr/reference-branch/bench/solr-benchmark.jar | Bin 37946 -> 37932 bytes
.../java/org/apache/solr/common/util/SysStats.java | 19 ++++++--
3 files changed, 48 insertions(+), 24 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/servlet/SolrQoSFilter.java b/solr/core/src/java/org/apache/solr/servlet/SolrQoSFilter.java
index 0df8fe7..a788ca4 100644
--- a/solr/core/src/java/org/apache/solr/servlet/SolrQoSFilter.java
+++ b/solr/core/src/java/org/apache/solr/servlet/SolrQoSFilter.java
@@ -50,17 +50,16 @@ public class SolrQoSFilter extends QoSFilter {
@Override
public void init(FilterConfig filterConfig) {
super.init(filterConfig);
- _origMaxRequests = Integer.getInteger("solr.concurrentRequests.max", 10000);
+ _origMaxRequests = Integer.getInteger("solr.concurrentRequests.max", 5000);
super.setMaxRequests(_origMaxRequests);
super.setSuspendMs(Integer.getInteger("solr.concurrentRequests.suspendms", 20000));
- super.setWaitMs(Integer.getInteger("solr.concurrentRequests.waitms", 500));
+ super.setWaitMs(Integer.getInteger("solr.concurrentRequests.waitms", 2000));
}
@Override
// nocommit - this is all just test/prototype - we should extract an actual strategy for adjusting on load
// allow the user to select and configure one
- public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
- throws IOException, ServletException {
+ public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
HttpServletRequest req = (HttpServletRequest) request;
String source = req.getHeader(QoSParams.REQUEST_SOURCE);
boolean imagePath = req.getPathInfo() != null && req.getPathInfo().startsWith("/img/");
@@ -71,27 +70,42 @@ public class SolrQoSFilter extends QoSFilter {
if (log.isDebugEnabled()) log.debug("external request"); //nocommit: remove when testing is done
double ourLoad = sysStats.getTotalUsage();
if (log.isDebugEnabled()) log.debug("Our individual load is {}", ourLoad);
+ double sLoad = sysStats.getSystemLoad();
if (ourLoad > SysStats.OUR_LOAD_HIGH) {
+
int cMax = getMaxRequests();
if (cMax > 5) {
- int max = Math.max(5, (int) ((double)cMax * 0.60D));
- log.warn("Our individual load is {}, set max concurrent requests to {}", ourLoad, max);
- updateMaxRequests(max);
+ int max = Math.max(5, (int) ((double) cMax * 0.30D));
+ log.warn("Our individual load is {}", ourLoad);
+ updateMaxRequests(max, sLoad, ourLoad);
}
+
} else {
// nocommit - deal with no supported, use this as a fail safe with high and low watermark?
- double sLoad = sysStats.getSystemLoad();
- if (sLoad > 1.3) {
- int cMax = getMaxRequests();
- if (cMax > 5) {
- int max = Math.max(5, (int) ((double) cMax * 0.60D));
- log.warn("System load is {}, set max concurrent requests to {}", sLoad, max);
- updateMaxRequests(max);
+ if (ourLoad < 0.70 && sLoad < 1.0 && _origMaxRequests != getMaxRequests()) {
+ if (sLoad < 0.9) {
+ if (log.isDebugEnabled()) log.debug("set max concurrent requests to orig value {}", _origMaxRequests);
+ updateMaxRequests(_origMaxRequests, sLoad, ourLoad);
+ } else {
+ updateMaxRequests(Math.min(_origMaxRequests, (int) Math.round(getMaxRequests() * 1.5D)), sLoad, ourLoad);
}
- } else if (sLoad < 0.95 && _origMaxRequests != getMaxRequests()) {
+ } else {
+ if (sLoad > 1.1) {
+ int cMax = getMaxRequests();
+ if (cMax > 5) {
+ int max = Math.max(5, (int) ((double) cMax * 0.30D));
+ // log.warn("System load is {} and our load is {} procs is {}, set max concurrent requests to {}", sLoad, ourLoad, SysStats.PROC_COUNT, max);
+ updateMaxRequests(max, sLoad, ourLoad);
+ }
+ } else if (ourLoad < 0.70 && sLoad < 1.0 && _origMaxRequests != getMaxRequests()) {
+ if (sLoad < 0.9) {
+ if (log.isDebugEnabled()) log.debug("set max concurrent requests to orig value {}", _origMaxRequests);
+ updateMaxRequests(_origMaxRequests, sLoad, ourLoad);
+ } else {
+ updateMaxRequests(Math.min(_origMaxRequests, (int) Math.round(getMaxRequests() * 1.5D)), sLoad, ourLoad);
+ }
- if (log.isDebugEnabled()) log.debug("set max concurrent requests to orig value {}", _origMaxRequests);
- updateMaxRequests(_origMaxRequests);
+ }
}
}
@@ -104,8 +118,9 @@ public class SolrQoSFilter extends QoSFilter {
}
}
- private void updateMaxRequests(int max) {
- if (System.currentTimeMillis() - lastUpdate < 3000) {
+ private void updateMaxRequests(int max, double sLoad, double ourLoad) {
+ if (System.currentTimeMillis() - lastUpdate > 2000) {
+ log.warn("Set max request to {} sload={} ourload={}", max, sLoad, ourLoad);
lastUpdate = System.currentTimeMillis();
setMaxRequests(max);
}
diff --git a/solr/reference-branch/bench/solr-benchmark.jar b/solr/reference-branch/bench/solr-benchmark.jar
index e1ca7c3..b17bfc8 100644
Binary files a/solr/reference-branch/bench/solr-benchmark.jar and b/solr/reference-branch/bench/solr-benchmark.jar differ
diff --git a/solr/solrj/src/java/org/apache/solr/common/util/SysStats.java b/solr/solrj/src/java/org/apache/solr/common/util/SysStats.java
index 28da816..05417d2 100644
--- a/solr/solrj/src/java/org/apache/solr/common/util/SysStats.java
+++ b/solr/solrj/src/java/org/apache/solr/common/util/SysStats.java
@@ -19,9 +19,9 @@ public class SysStats extends Thread {
private static final Logger log = LoggerFactory
.getLogger(MethodHandles.lookup().lookupClass());
- public static final double OUR_LOAD_HIGH = 1.5;
+ public static final double OUR_LOAD_HIGH = 1.0;
public static final long REFRESH_INTERVAL = TimeUnit.NANOSECONDS.convert(5000, TimeUnit.MILLISECONDS);
- static final int PROC_COUNT = ManagementFactory.getOperatingSystemMXBean().getAvailableProcessors();
+ public static final int PROC_COUNT = ManagementFactory.getOperatingSystemMXBean().getAvailableProcessors();
private final long refreshIntervalMs;
private long refreshInterval;
@@ -35,6 +35,10 @@ public class SysStats extends Thread {
private volatile double totalUsage;
private volatile double sysLoad;
+ public static void main(String[] args) {
+ System.out.println("avail procs:" + PROC_COUNT);
+ }
+
public static SysStats getSysStats() {
if (sysStats == null) {
synchronized (SysStats.class) {
@@ -82,7 +86,7 @@ public class SysStats extends Thread {
values = new HashSet<>(threadTimeMap.values());
for (ThreadTime threadTime : values) {
- threadTime.setCurrent(threadBean.getThreadCpuTime(threadTime.getId()));
+ threadTime.setLast(threadBean.getThreadCpuTime(threadTime.getId()));
}
double load = ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage();
@@ -116,8 +120,13 @@ public class SysStats extends Thread {
if (gatherThreadStats) {
double usage = 0D;
for (ThreadTime threadTime : values) {
- threadTime.setLast(threadBean.getThreadCpuTime(threadTime.getId()));
- usage += ( threadTime.getLast() - threadTime.getCurrent()) / (refreshInterval * 2.0f);
+ long time = threadBean.getThreadCpuTime(threadTime.getId());
+ if (time > 0) {
+ long diff = time - threadTime.getLast();
+ if (diff > 0) {
+ usage += diff / (refreshInterval * 2.0f);
+ }
+ }
}
totalUsage = usage;
gatherThreadStats = false;