You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by da...@apache.org on 2020/08/25 07:59:47 UTC
[lucene-solr] branch datcm/bloomberg-abdicate-leadership created
(now b9f24cc)
This is an automated email from the ASF dual-hosted git repository.
datcm pushed a change to branch datcm/bloomberg-abdicate-leadership
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git.
at b9f24cc Remove unnecessary changes
This branch includes the following new commits:
new 9432982 First commit
new 3d05867 Move tracking to coreDescriptor
new b9f24cc Remove unnecessary changes
The 3 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.
[lucene-solr] 02/03: Move tracking to coreDescriptor
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
datcm pushed a commit to branch datcm/bloomberg-abdicate-leadership
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 3d058672fdde7b1a637be60cdd5482c45b720ba7
Author: Cao Manh Dat <da...@apache.org>
AuthorDate: Tue Aug 25 14:55:00 2020 +0700
Move tracking to coreDescriptor
---
.../java/org/apache/solr/core/CoreContainer.java | 83 --------------------
.../java/org/apache/solr/core/CoreDescriptor.java | 37 +++++++++
.../solr/handler/admin/CoreAdminOperation.java | 91 +++++++++++++---------
.../java/org/apache/solr/servlet/HttpSolrCall.java | 4 +-
.../apache/solr/common/params/CoreAdminParams.java | 1 +
5 files changed, 93 insertions(+), 123 deletions(-)
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 711b5c7..c537791 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -261,89 +261,6 @@ public class CoreContainer {
private ExecutorService coreContainerAsyncTaskExecutor = ExecutorUtil.newMDCAwareCachedThreadPool("Core Container Async Task");
private final Object lock = new Object();
- private Set<String> pausedUpdatesCores = Collections.synchronizedSet(new HashSet<>());
- private Map<String, AtomicInteger> currentCoreUpdateCounters = new HashMap<>();
-
- public boolean startTrackUpdateRequest(String coreName) {
- synchronized (lock) {
- if (pausedUpdatesCores.contains(coreName)) {
- return false;
- }
- if (!currentCoreUpdateCounters.containsKey(coreName)) {
- currentCoreUpdateCounters.put(coreName, new AtomicInteger());
- }
- currentCoreUpdateCounters.get(coreName).incrementAndGet();
- return true;
- }
- }
-
- public void endTrackUpdateRequest(String coreName) {
- synchronized (lock) {
- currentCoreUpdateCounters.get(coreName).decrementAndGet();
- }
- }
-
- public void pauseUpdateFor(String coreName) {
- synchronized (lock) {
- pausedUpdatesCores.add(coreName);
- }
- }
-
- public void unpauseUpdateFor(String coreName) {
- synchronized (lock) {
- pausedUpdatesCores.remove(coreName);
- }
- }
-
- public void waitForEmptyUpdates(String coreName) {
- AtomicInteger counter = currentCoreUpdateCounters.get(coreName);
- if (counter == null)
- return;
-
- while (counter.get() != 0) {
- log.info("Datcm waiting for finish pending updates:{}", counter.get());
- try {
- Thread.sleep(300);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- return;
- }
- }
-
- }
-
- public void abdicateLeadership(String coreName) {
- CoreDescriptor cd = getCoreDescriptor(coreName);
- if (!isZooKeeperAware() || cd.getCloudDescriptor() == null || !cd.getCloudDescriptor().isLeader())
- return;
-
- try {
- synchronized (lock) {
- pausedUpdatesCores.add(coreName);
- }
- AtomicInteger counter = currentCoreUpdateCounters.get(coreName);
- while (counter != null && counter.get() != 0) {
- try {
- Thread.sleep(300);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- return;
- }
- }
- try {
- zkSys.zkController.rejoinShardLeaderElection(cd, false);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- log.error("Exception on abdicate leadership of core:{}", coreName, e);
- } catch (Exception e) {
- log.error("Exception on abdicate leadership of core:{}", coreName, e);
- }
- } finally {
- synchronized (lock) {
- pausedUpdatesCores.remove(coreName);
- }
- }
- }
private enum CoreInitFailedAction {fromleader, none}
diff --git a/solr/core/src/java/org/apache/solr/core/CoreDescriptor.java b/solr/core/src/java/org/apache/solr/core/CoreDescriptor.java
index 7b3428d..a51d979 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreDescriptor.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreDescriptor.java
@@ -28,6 +28,7 @@ import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
@@ -141,6 +142,11 @@ public class CoreDescriptor {
/** The properties for this core, substitutable by resource loaders */
protected final Properties substitutableProperties = new Properties();
+ /** If true then all update requests will be refused */
+ protected volatile boolean pausedAcceptingIndexing = false;
+
+ protected final AtomicInteger inflightUpdatesCounter = new AtomicInteger();
+
/** TESTS ONLY */
public CoreDescriptor(String name, Path instanceDir, CoreContainer coreContainer, String... coreProps) {
this(name, instanceDir, toMap(coreProps), coreContainer.getContainerProperties(), coreContainer.getZkController());
@@ -334,6 +340,37 @@ public class CoreDescriptor {
return Boolean.parseBoolean(tmp);
}
+ public void setPausedAcceptingIndexing(boolean pausedAcceptingIndexing) {
+ this.pausedAcceptingIndexing = pausedAcceptingIndexing;
+ }
+
+ public boolean isPausedAcceptingIndexing() {
+ return pausedAcceptingIndexing;
+ }
+
+ public boolean incInflightUpdateCounter() {
+ if (isPausedAcceptingIndexing()) {
+ return false;
+ }
+ inflightUpdatesCounter.incrementAndGet();
+ return true;
+ }
+
+ public void decInFlightUpdateCounter() {
+ inflightUpdatesCounter.decrementAndGet();
+ }
+
+ public void waitForNoInflightUpdates() {
+ while (inflightUpdatesCounter.get() != 0) {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ }
+
public boolean isTransient() {
String tmp = coreProperties.getProperty(CORE_TRANSIENT, "false");
return PropertiesUtil.toBoolean(tmp);
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java
index 2b852a4..6c0021b 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java
@@ -16,26 +16,14 @@
*/
package org.apache.solr.handler.admin;
-import java.io.IOException;
-import java.lang.invoke.MethodHandles;
-import java.nio.file.Path;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.commons.lang3.StringUtils;
import org.apache.solr.cloud.CloudDescriptor;
-import org.apache.solr.cloud.Overseer;
import org.apache.solr.cloud.ZkController;
-import org.apache.solr.cloud.overseer.OverseerAction;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.Slice;
-import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.CoreAdminParams;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.NamedList;
@@ -48,7 +36,7 @@ import org.apache.solr.core.SolrInfoBean;
import org.apache.solr.core.snapshots.SolrSnapshotManager;
import org.apache.solr.core.snapshots.SolrSnapshotMetaDataManager;
import org.apache.solr.core.snapshots.SolrSnapshotMetaDataManager.SnapshotMetaData;
-import org.apache.solr.handler.admin.CoreAdminHandler.CoreAdminOp;
+import org.apache.solr.handler.admin.CoreAdminHandler.*;
import org.apache.solr.metrics.SolrMetricManager;
import org.apache.solr.search.SolrIndexSearcher;
import org.apache.solr.update.UpdateLog;
@@ -59,6 +47,16 @@ import org.apache.solr.util.TestInjection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
import static org.apache.solr.common.params.CommonParams.NAME;
import static org.apache.solr.common.params.CoreAdminParams.COLLECTION;
import static org.apache.solr.common.params.CoreAdminParams.CoreAdminAction.*;
@@ -238,38 +236,55 @@ enum CoreAdminOperation implements CoreAdminOp {
}
}
}),
+
ABDICATE_LEADERSHIP_OP(ABDICATE_LEADERSHIP, it -> {
CoreContainer cc = it.handler.coreContainer;
if (!cc.isZooKeeperAware()) {
throw new SolrException(ErrorCode.BAD_REQUEST, "This api only works in SolrCloud env");
}
- for (String coreName: cc.getAllCoreNames()) {
- try {
- CoreDescriptor cd = cc.getCoreDescriptor(coreName);
- if (cd == null || cd.getCloudDescriptor() == null || !cd.getCloudDescriptor().isLeader()) {
- continue;
- }
-
- ClusterState clusterState = cc.getZkController().getClusterState();
- DocCollection docCollection = clusterState.getCollection(cd.getCollectionName());
- Slice slice = docCollection.getSlice(cd.getCloudDescriptor().getShardId());
- if (slice.getReplicas().size() < 2) {
- log().info("Skipping abdicate leadership of core:{} since it is the only one replicas in the shard", coreName);
- }
- if (slice.getReplicas(replica -> replica.isActive(clusterState.getLiveNodes())).size() < 2) {
- log().info("Skipping abdicate leadership of core:{} since it is the only one active replica", coreName);
- }
+ String cores = it.req.getParams().get("cores");
+ Set<String> selectedCores = null;
+ if (cores != null) {
+ selectedCores = new HashSet<>(Arrays.asList(cores.split(",")));
+ }
+ for (CoreDescriptor cd: cc.getCoreDescriptors()) {
+ if (cd == null || cd.getCloudDescriptor() == null || !cd.getCloudDescriptor().isLeader()) {
+ continue;
+ }
+ if (selectedCores != null && !selectedCores.contains(cd.getName())) {
+ continue;
+ }
- cc.pauseUpdateFor(coreName);
- cc.waitForEmptyUpdates(coreName);
- cc.getZkController().rejoinShardLeaderElection(cd, false);
- } finally {
- cc.unpauseUpdateFor(coreName);
- // after this point this core can still receive more updates since it did not explicitly
- // unset its leader role from cluster state, but that fine since it already unset its leader role locally
- // (by setting cloudDescriptor.isLeader() flag) so it gonna keep refusing updates or routing update requests
- // to a new leader
+ try {
+ ClusterState clusterState = cc.getZkController().getClusterState();
+ DocCollection docCollection = clusterState.getCollection(cd.getCollectionName());
+ Slice slice = docCollection.getSlice(cd.getCloudDescriptor().getShardId());
+ if (slice.getReplicas().size() < 2) {
+ log().info("Skipping abdicate leadership of core:{} since it is the only one replicas in the shard", cd.getName());
+ }
+ if (slice.getReplicas(replica -> replica.isActive(clusterState.getLiveNodes())).size() < 2) {
+ log().info("Skipping abdicate leadership of core:{} since it is the only one active replica", cd.getName());
}
+
+ cd.setPausedAcceptingIndexing(true);
+ cd.waitForNoInflightUpdates();
+ cc.getZkController().rejoinShardLeaderElection(cd, false);
+ } finally {
+ cd.setPausedAcceptingIndexing(false);
+ // after this point this core can still receive more updates since it did not explicitly
+ // unset its leader role from cluster state, but that fine since it already unset its leader role locally
+ // (by setting cloudDescriptor.isLeader() flag) so it gonna keep refusing updates or routing update requests
+ // to a new leader
+ }
+ }
+ }),
+
+ UNPAUSE_INDEXING_OP(UNPAUSE_INDEXING, it -> {
+ CoreContainer cc = it.handler.coreContainer;
+ for (CoreDescriptor cd : cc.getCoreDescriptors()) {
+ if (cd != null) {
+ cd.setPausedAcceptingIndexing(false);
+ }
}
}),
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 895955d..27fa106 100644
--- a/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java
+++ b/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java
@@ -572,7 +572,7 @@ public class HttpSolrCall {
remoteQuery(coreUrl + path, resp);
return RETURN;
case PROCESS:
- if (handler instanceof UpdateRequestHandler && !cores.startTrackUpdateRequest(core.getName())) {
+ if (handler instanceof UpdateRequestHandler && !core.getCoreDescriptor().incInflightUpdateCounter()) {
throw new SolrException(ErrorCode.SERVER_ERROR, "Updates are temporary paused for core:"+core.getName());
}
try {
@@ -611,7 +611,7 @@ public class HttpSolrCall {
return RETURN;
} finally {
if (handler instanceof UpdateRequestHandler) {
- cores.endTrackUpdateRequest(core.getName());
+ core.getCoreDescriptor().decInFlightUpdateCounter();
}
}
default: return action;
diff --git a/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java b/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
index daff1fe..ebee497 100644
--- a/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
+++ b/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
@@ -154,6 +154,7 @@ public abstract class CoreAdminParams
REQUESTAPPLYUPDATES,
OVERSEEROP,
ABDICATE_LEADERSHIP,
+ UNPAUSE_INDEXING,
REQUESTSTATUS(true),
REJOINLEADERELECTION,
//internal API used by force shard leader election
[lucene-solr] 03/03: Remove unnecessary changes
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
datcm pushed a commit to branch datcm/bloomberg-abdicate-leadership
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit b9f24cc3d524cc77d77533314681988de3c5e22e
Author: Cao Manh Dat <da...@apache.org>
AuthorDate: Tue Aug 25 14:59:04 2020 +0700
Remove unnecessary changes
---
.../java/org/apache/solr/cloud/LeaderElector.java | 2 +-
.../java/org/apache/solr/core/CoreContainer.java | 6 ----
.../src/java/org/apache/solr/core/SolrCores.java | 41 ++++++++++++----------
3 files changed, 24 insertions(+), 25 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java b/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
index 4a42321..f50aa11 100644
--- a/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
+++ b/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
@@ -378,7 +378,7 @@ public class LeaderElector {
* Sort n string sequence list.
*/
public static void sortSeqs(List<String> seqs) {
- seqs.sort((o1, o2) -> {
+ Collections.sort(seqs, (o1, o2) -> {
int i = getSeq(o1) - getSeq(o2);
return i == 0 ? o1.compareTo(o2) : i;
});
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 c537791..da95aab 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -30,7 +30,6 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
@@ -43,8 +42,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.BiFunction;
import java.util.function.Supplier;
import com.google.common.annotations.VisibleForTesting;
@@ -81,7 +78,6 @@ import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Replica.State;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.params.MapSolrParams;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.IOUtils;
import org.apache.solr.common.util.ObjectCache;
@@ -260,8 +256,6 @@ public class CoreContainer {
private ExecutorService coreContainerAsyncTaskExecutor = ExecutorUtil.newMDCAwareCachedThreadPool("Core Container Async Task");
- private final Object lock = new Object();
-
private enum CoreInitFailedAction {fromleader, none}
/**
diff --git a/solr/core/src/java/org/apache/solr/core/SolrCores.java b/solr/core/src/java/org/apache/solr/core/SolrCores.java
index f5a57a0..9a453c9 100644
--- a/solr/core/src/java/org/apache/solr/core/SolrCores.java
+++ b/solr/core/src/java/org/apache/solr/core/SolrCores.java
@@ -41,15 +41,15 @@ import java.util.concurrent.TimeUnit;
class SolrCores {
- private static final Object modifyLock = new Object(); // for locking around manipulating any of the core maps.
+ private static Object modifyLock = new Object(); // for locking around manipulating any of the core maps.
private final Map<String, SolrCore> cores = new LinkedHashMap<>(); // For "permanent" cores
// These descriptors, once loaded, will _not_ be unloaded, i.e. they are not "transient".
private final Map<String, CoreDescriptor> residentDesciptors = new LinkedHashMap<>();
private final CoreContainer container;
-
- private final Set<String> currentlyLoadingCores = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+ private Set<String> currentlyLoadingCores = Collections.newSetFromMap(new ConcurrentHashMap<String,Boolean>());
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -63,10 +63,12 @@ class SolrCores {
private TransientSolrCoreCacheFactory transientCoreCache;
+ private TransientSolrCoreCache transientSolrCoreCache = null;
+
SolrCores(CoreContainer container) {
this.container = container;
}
-
+
protected void addCoreDescriptor(CoreDescriptor p) {
synchronized (modifyLock) {
if (p.isTransient()) {
@@ -102,7 +104,7 @@ class SolrCores {
waitForLoadingCoresToFinish(30*1000);
Collection<SolrCore> coreList = new ArrayList<>();
-
+
TransientSolrCoreCache transientSolrCoreCache = getTransientCacheHandler();
// Release observer
if (transientSolrCoreCache != null) {
@@ -125,7 +127,7 @@ class SolrCores {
coreList.addAll(pendingCloses);
pendingCloses.clear();
}
-
+
ExecutorService coreCloseExecutor = ExecutorUtil.newMDCAwareFixedThreadPool(Integer.MAX_VALUE,
new SolrNamedThreadFactory("coreCloseExecutor"));
try {
@@ -151,7 +153,7 @@ class SolrCores {
} while (coreList.size() > 0);
}
-
+
// Returns the old core if there was a core of the same name.
//WARNING! This should be the _only_ place you put anything into the list of transient cores!
protected SolrCore putCore(CoreDescriptor cd, SolrCore core) {
@@ -172,11 +174,11 @@ class SolrCores {
/**
*
* @return A list of "permanent" cores, i.e. cores that may not be swapped out and are currently loaded.
- *
+ *
* A core may be non-transient but still lazily loaded. If it is "permanent" and lazy-load _and_
* not yet loaded it will _not_ be returned by this call.
- *
- * Note: This is one of the places where SolrCloud is incompatible with Transient Cores. This call is used in
+ *
+ * Note: This is one of the places where SolrCloud is incompatible with Transient Cores. This call is used in
* cancelRecoveries, transient cores don't participate.
*/
@@ -192,10 +194,10 @@ class SolrCores {
* Gets the cores that are currently loaded, i.e. cores that have
* 1> loadOnStartup=true and are either not-transient or, if transient, have been loaded and have not been aged out
* 2> loadOnStartup=false and have been loaded but either non-transient or have not been aged out.
- *
+ *
* Put another way, this will not return any names of cores that are lazily loaded but have not been called for yet
* or are transient and either not loaded or have been swapped out.
- *
+ *
* @return List of currently loaded cores.
*/
Set<String> getLoadedCoreNames() {
@@ -210,7 +212,7 @@ class SolrCores {
return set;
}
/**
- * Gets a list of all cores, loaded and unloaded
+ * Gets a list of all cores, loaded and unloaded
*
* @return all cores names, whether loaded or unloaded, transient or permanent.
*/
@@ -250,16 +252,16 @@ class SolrCores {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "No such core: " + n1);
}
}
- // When we swap the cores, we also need to swap the associated core descriptors. Note, this changes the
+ // When we swap the cores, we also need to swap the associated core descriptors. Note, this changes the
// name of the coreDescriptor by virtue of the c-tor
- CoreDescriptor cd1 = c1.getCoreDescriptor();
+ CoreDescriptor cd1 = c1.getCoreDescriptor();
addCoreDescriptor(new CoreDescriptor(n1, c0.getCoreDescriptor()));
addCoreDescriptor(new CoreDescriptor(n0, cd1));
cores.put(n0, c1);
cores.put(n1, c0);
c0.setName(n1);
c1.setName(n0);
-
+
container.getMetricManager().swapRegistries(
c0.getCoreMetricManager().getRegistryName(),
c1.getCoreMetricManager().getRegistryName());
@@ -481,7 +483,7 @@ class SolrCores {
}
}
}
-
+
// returns when core is finished loading, throws exception if no such core loading or loaded
public void waitForLoadingCoreToFinish(String core, long timeoutMs) {
long time = System.nanoTime();
@@ -502,7 +504,10 @@ class SolrCores {
}
public boolean isCoreLoading(String name) {
- return currentlyLoadingCores.contains(name);
+ if (currentlyLoadingCores.contains(name)) {
+ return true;
+ }
+ return false;
}
public void queueCoreToClose(SolrCore coreToClose) {
[lucene-solr] 01/03: First commit
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
datcm pushed a commit to branch datcm/bloomberg-abdicate-leadership
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 9432982935d99921ef2ba388b33fdc73b2197ba6
Author: Cao Manh Dat <da...@apache.org>
AuthorDate: Fri Aug 7 14:21:08 2020 +0700
First commit
---
.../java/org/apache/solr/cloud/LeaderElector.java | 2 +-
.../java/org/apache/solr/cloud/ZkController.java | 15 ++++
.../java/org/apache/solr/core/CoreContainer.java | 89 ++++++++++++++++++++++
.../src/java/org/apache/solr/core/SolrCores.java | 11 +--
.../solr/handler/admin/CoreAdminOperation.java | 42 ++++++++++
.../java/org/apache/solr/servlet/HttpSolrCall.java | 58 ++++++++------
.../apache/solr/cloud/TestAbdicateLeadership.java | 73 ++++++++++++++++++
.../client/solrj/request/CoreAdminRequest.java | 10 +++
.../apache/solr/common/params/CoreAdminParams.java | 1 +
9 files changed, 268 insertions(+), 33 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java b/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
index f50aa11..4a42321 100644
--- a/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
+++ b/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
@@ -378,7 +378,7 @@ public class LeaderElector {
* Sort n string sequence list.
*/
public static void sortSeqs(List<String> seqs) {
- Collections.sort(seqs, (o1, o2) -> {
+ seqs.sort((o1, o2) -> {
int i = getSeq(o1) - getSeq(o2);
return i == 0 ? o1.compareTo(o2) : i;
});
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 dc4c210..400002a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -92,6 +92,7 @@ import org.apache.solr.common.cloud.ZooKeeperException;
import org.apache.solr.common.params.CollectionParams;
import org.apache.solr.common.params.CommonParams;
import org.apache.solr.common.params.CoreAdminParams;
+import org.apache.solr.common.params.MapSolrParams;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.IOUtils;
@@ -2185,6 +2186,20 @@ public class ZkController implements Closeable {
}
+ public void rejoinShardLeaderElection(CoreDescriptor coreDescriptor, boolean rejoinAtHead) throws InterruptedException, IOException, KeeperException {
+ CloudDescriptor cloudDescriptor = coreDescriptor.getCloudDescriptor();
+ if (cloudDescriptor == null) {
+ return;
+ }
+ ContextKey contextKey = new ContextKey(coreDescriptor.getCollectionName(), cloudDescriptor.getCoreNodeName());
+ ElectionContext prevContext = electionContexts.get(contextKey);
+ if (prevContext == null) {
+ return;
+ }
+ ShardLeaderElectionContext prevElectionContext = (ShardLeaderElectionContext) prevContext;
+ prevElectionContext.getLeaderElector().retryElection(prevContext, rejoinAtHead);
+ }
+
public void rejoinShardLeaderElection(SolrParams params) {
String collectionName = params.get(COLLECTION_PROP);
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 da95aab..711b5c7 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -30,6 +30,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
@@ -42,6 +43,8 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiFunction;
import java.util.function.Supplier;
import com.google.common.annotations.VisibleForTesting;
@@ -78,6 +81,7 @@ import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Replica.State;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.MapSolrParams;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.IOUtils;
import org.apache.solr.common.util.ObjectCache;
@@ -256,6 +260,91 @@ public class CoreContainer {
private ExecutorService coreContainerAsyncTaskExecutor = ExecutorUtil.newMDCAwareCachedThreadPool("Core Container Async Task");
+ private final Object lock = new Object();
+ private Set<String> pausedUpdatesCores = Collections.synchronizedSet(new HashSet<>());
+ private Map<String, AtomicInteger> currentCoreUpdateCounters = new HashMap<>();
+
+ public boolean startTrackUpdateRequest(String coreName) {
+ synchronized (lock) {
+ if (pausedUpdatesCores.contains(coreName)) {
+ return false;
+ }
+ if (!currentCoreUpdateCounters.containsKey(coreName)) {
+ currentCoreUpdateCounters.put(coreName, new AtomicInteger());
+ }
+ currentCoreUpdateCounters.get(coreName).incrementAndGet();
+ return true;
+ }
+ }
+
+ public void endTrackUpdateRequest(String coreName) {
+ synchronized (lock) {
+ currentCoreUpdateCounters.get(coreName).decrementAndGet();
+ }
+ }
+
+ public void pauseUpdateFor(String coreName) {
+ synchronized (lock) {
+ pausedUpdatesCores.add(coreName);
+ }
+ }
+
+ public void unpauseUpdateFor(String coreName) {
+ synchronized (lock) {
+ pausedUpdatesCores.remove(coreName);
+ }
+ }
+
+ public void waitForEmptyUpdates(String coreName) {
+ AtomicInteger counter = currentCoreUpdateCounters.get(coreName);
+ if (counter == null)
+ return;
+
+ while (counter.get() != 0) {
+ log.info("Datcm waiting for finish pending updates:{}", counter.get());
+ try {
+ Thread.sleep(300);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+
+ }
+
+ public void abdicateLeadership(String coreName) {
+ CoreDescriptor cd = getCoreDescriptor(coreName);
+ if (!isZooKeeperAware() || cd.getCloudDescriptor() == null || !cd.getCloudDescriptor().isLeader())
+ return;
+
+ try {
+ synchronized (lock) {
+ pausedUpdatesCores.add(coreName);
+ }
+ AtomicInteger counter = currentCoreUpdateCounters.get(coreName);
+ while (counter != null && counter.get() != 0) {
+ try {
+ Thread.sleep(300);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ try {
+ zkSys.zkController.rejoinShardLeaderElection(cd, false);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.error("Exception on abdicate leadership of core:{}", coreName, e);
+ } catch (Exception e) {
+ log.error("Exception on abdicate leadership of core:{}", coreName, e);
+ }
+ } finally {
+ synchronized (lock) {
+ pausedUpdatesCores.remove(coreName);
+ }
+ }
+ }
+
private enum CoreInitFailedAction {fromleader, none}
/**
diff --git a/solr/core/src/java/org/apache/solr/core/SolrCores.java b/solr/core/src/java/org/apache/solr/core/SolrCores.java
index 205b70f..f5a57a0 100644
--- a/solr/core/src/java/org/apache/solr/core/SolrCores.java
+++ b/solr/core/src/java/org/apache/solr/core/SolrCores.java
@@ -41,7 +41,7 @@ import java.util.concurrent.TimeUnit;
class SolrCores {
- private static Object modifyLock = new Object(); // for locking around manipulating any of the core maps.
+ private static final Object modifyLock = new Object(); // for locking around manipulating any of the core maps.
private final Map<String, SolrCore> cores = new LinkedHashMap<>(); // For "permanent" cores
// These descriptors, once loaded, will _not_ be unloaded, i.e. they are not "transient".
@@ -49,7 +49,7 @@ class SolrCores {
private final CoreContainer container;
- private Set<String> currentlyLoadingCores = Collections.newSetFromMap(new ConcurrentHashMap<String,Boolean>());
+ private final Set<String> currentlyLoadingCores = Collections.newSetFromMap(new ConcurrentHashMap<>());
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -63,8 +63,6 @@ class SolrCores {
private TransientSolrCoreCacheFactory transientCoreCache;
- private TransientSolrCoreCache transientSolrCoreCache = null;
-
SolrCores(CoreContainer container) {
this.container = container;
}
@@ -504,10 +502,7 @@ class SolrCores {
}
public boolean isCoreLoading(String name) {
- if (currentlyLoadingCores.contains(name)) {
- return true;
- }
- return false;
+ return currentlyLoadingCores.contains(name);
}
public void queueCoreToClose(SolrCore coreToClose) {
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java
index 3036ced..2b852a4 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java
@@ -22,12 +22,20 @@ import java.nio.file.Path;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.lang3.StringUtils;
import org.apache.solr.cloud.CloudDescriptor;
+import org.apache.solr.cloud.Overseer;
import org.apache.solr.cloud.ZkController;
+import org.apache.solr.cloud.overseer.OverseerAction;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.cloud.ZkNodeProps;
+import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.CoreAdminParams;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.NamedList;
@@ -230,6 +238,40 @@ enum CoreAdminOperation implements CoreAdminOp {
}
}
}),
+ ABDICATE_LEADERSHIP_OP(ABDICATE_LEADERSHIP, it -> {
+ CoreContainer cc = it.handler.coreContainer;
+ if (!cc.isZooKeeperAware()) {
+ throw new SolrException(ErrorCode.BAD_REQUEST, "This api only works in SolrCloud env");
+ }
+ for (String coreName: cc.getAllCoreNames()) {
+ try {
+ CoreDescriptor cd = cc.getCoreDescriptor(coreName);
+ if (cd == null || cd.getCloudDescriptor() == null || !cd.getCloudDescriptor().isLeader()) {
+ continue;
+ }
+
+ ClusterState clusterState = cc.getZkController().getClusterState();
+ DocCollection docCollection = clusterState.getCollection(cd.getCollectionName());
+ Slice slice = docCollection.getSlice(cd.getCloudDescriptor().getShardId());
+ if (slice.getReplicas().size() < 2) {
+ log().info("Skipping abdicate leadership of core:{} since it is the only one replicas in the shard", coreName);
+ }
+ if (slice.getReplicas(replica -> replica.isActive(clusterState.getLiveNodes())).size() < 2) {
+ log().info("Skipping abdicate leadership of core:{} since it is the only one active replica", coreName);
+ }
+
+ cc.pauseUpdateFor(coreName);
+ cc.waitForEmptyUpdates(coreName);
+ cc.getZkController().rejoinShardLeaderElection(cd, false);
+ } finally {
+ cc.unpauseUpdateFor(coreName);
+ // after this point this core can still receive more updates since it did not explicitly
+ // unset its leader role from cluster state, but that fine since it already unset its leader role locally
+ // (by setting cloudDescriptor.isLeader() flag) so it gonna keep refusing updates or routing update requests
+ // to a new leader
+ }
+ }
+ }),
REJOINLEADERELECTION_OP(REJOINLEADERELECTION, it -> {
ZkController zkController = it.handler.coreContainer.getZkController();
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 1fed351..895955d 100644
--- a/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java
+++ b/solr/core/src/java/org/apache/solr/servlet/HttpSolrCall.java
@@ -86,6 +86,7 @@ import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.SolrConfig;
import org.apache.solr.core.SolrCore;
import org.apache.solr.handler.ContentStreamHandlerBase;
+import org.apache.solr.handler.UpdateRequestHandler;
import org.apache.solr.logging.MDCLoggingContext;
import org.apache.solr.request.LocalSolrQueryRequest;
import org.apache.solr.request.SolrQueryRequest;
@@ -571,39 +572,48 @@ public class HttpSolrCall {
remoteQuery(coreUrl + path, resp);
return RETURN;
case PROCESS:
- final Method reqMethod = Method.getMethod(req.getMethod());
- HttpCacheHeaderUtil.setCacheControlHeader(config, resp, reqMethod);
- // unless we have been explicitly told not to, do cache validation
- // if we fail cache validation, execute the query
- if (config.getHttpCachingConfig().isNever304() ||
- !HttpCacheHeaderUtil.doCacheHeaderValidation(solrReq, req, reqMethod, resp)) {
- SolrQueryResponse solrRsp = new SolrQueryResponse();
+ if (handler instanceof UpdateRequestHandler && !cores.startTrackUpdateRequest(core.getName())) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, "Updates are temporary paused for core:"+core.getName());
+ }
+ try {
+ final Method reqMethod = Method.getMethod(req.getMethod());
+ HttpCacheHeaderUtil.setCacheControlHeader(config, resp, reqMethod);
+ // unless we have been explicitly told not to, do cache validation
+ // if we fail cache validation, execute the query
+ if (config.getHttpCachingConfig().isNever304() ||
+ !HttpCacheHeaderUtil.doCacheHeaderValidation(solrReq, req, reqMethod, resp)) {
+ SolrQueryResponse solrRsp = new SolrQueryResponse();
/* even for HEAD requests, we need to execute the handler to
* ensure we don't get an error (and to make sure the correct
* QueryResponseWriter is selected and we get the correct
* Content-Type)
*/
- SolrRequestInfo.setRequestInfo(new SolrRequestInfo(solrReq, solrRsp, action));
- mustClearSolrRequestInfo = true;
- execute(solrRsp);
- if (shouldAudit()) {
- EventType eventType = solrRsp.getException() == null ? EventType.COMPLETED : EventType.ERROR;
- if (shouldAudit(eventType)) {
- cores.getAuditLoggerPlugin().doAudit(
- new AuditEvent(eventType, req, getAuthCtx(), solrReq.getRequestTimer().getTime(), solrRsp.getException()));
+ SolrRequestInfo.setRequestInfo(new SolrRequestInfo(solrReq, solrRsp, action));
+ mustClearSolrRequestInfo = true;
+ execute(solrRsp);
+ if (shouldAudit()) {
+ EventType eventType = solrRsp.getException() == null ? EventType.COMPLETED : EventType.ERROR;
+ if (shouldAudit(eventType)) {
+ cores.getAuditLoggerPlugin().doAudit(
+ new AuditEvent(eventType, req, getAuthCtx(), solrReq.getRequestTimer().getTime(), solrRsp.getException()));
+ }
+ }
+ HttpCacheHeaderUtil.checkHttpCachingVeto(solrRsp, resp, reqMethod);
+ Iterator<Map.Entry<String, String>> headers = solrRsp.httpHeaders();
+ while (headers.hasNext()) {
+ Map.Entry<String, String> entry = headers.next();
+ resp.addHeader(entry.getKey(), entry.getValue());
}
+ QueryResponseWriter responseWriter = getResponseWriter();
+ if (invalidStates != null) solrReq.getContext().put(CloudSolrClient.STATE_VERSION, invalidStates);
+ writeResponse(solrRsp, responseWriter, reqMethod);
}
- HttpCacheHeaderUtil.checkHttpCachingVeto(solrRsp, resp, reqMethod);
- Iterator<Map.Entry<String, String>> headers = solrRsp.httpHeaders();
- while (headers.hasNext()) {
- Map.Entry<String, String> entry = headers.next();
- resp.addHeader(entry.getKey(), entry.getValue());
+ return RETURN;
+ } finally {
+ if (handler instanceof UpdateRequestHandler) {
+ cores.endTrackUpdateRequest(core.getName());
}
- QueryResponseWriter responseWriter = getResponseWriter();
- if (invalidStates != null) solrReq.getContext().put(CloudSolrClient.STATE_VERSION, invalidStates);
- writeResponse(solrRsp, responseWriter, reqMethod);
}
- return RETURN;
default: return action;
}
} catch (Throwable ex) {
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestAbdicateLeadership.java b/solr/core/src/test/org/apache/solr/cloud/TestAbdicateLeadership.java
new file mode 100644
index 0000000..442bf28
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/TestAbdicateLeadership.java
@@ -0,0 +1,73 @@
+package org.apache.solr.cloud;
+
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.embedded.JettySolrRunner;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.CoreAdminRequest;
+import org.apache.solr.client.solrj.response.CollectionAdminResponse;
+import org.apache.solr.client.solrj.response.CoreAdminResponse;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.util.TimeSource;
+import org.apache.solr.util.TimeOut;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class TestAbdicateLeadership extends SolrCloudTestCase {
+
+ private static final String COLLECTION_NAME = "collection1";
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+
+ configureCluster(3)
+ .addConfig(COLLECTION_NAME, configset("cloud-minimal"))
+ .configure();
+
+ CollectionAdminResponse resp = CollectionAdminRequest.createCollection(COLLECTION_NAME, COLLECTION_NAME, 1, 3) .process(cluster.getSolrClient());
+ assertEquals("Admin request failed; ", 0, resp.getStatus());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 1, 3);
+ }
+
+ @Test
+ public void test() throws IOException, SolrServerException, TimeoutException, InterruptedException {
+ DocCollection dc = cluster.getSolrClient().getZkStateReader()
+ .getClusterState().getCollection(COLLECTION_NAME);
+ Replica leaderReplica = dc.getSlice("shard1").getLeader();
+ JettySolrRunner runner = cluster.getReplicaJetty(leaderReplica);
+ AtomicBoolean closed = new AtomicBoolean(false);
+ Thread thread = new Thread(() -> {
+ int i = 0;
+ while (!closed.get()) {
+ try {
+ cluster.getSolrClient().add(COLLECTION_NAME, new SolrInputDocument("id", String.valueOf(i++)));
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (Exception e) { e.printStackTrace(); }
+ }
+ });
+ thread.start();
+ try (SolrClient solrClient = getHttpSolrClient(runner.getBaseUrl().toString())){
+ CoreAdminResponse response = CoreAdminRequest.abdicateLeadership(solrClient);
+ }
+ TimeOut timeOut = new TimeOut(10, TimeUnit.SECONDS, TimeSource.CURRENT_TIME);
+ timeOut.waitFor("Timeout waiting for a new leader", () -> {
+ ClusterState state = cluster.getSolrClient().getZkStateReader().getClusterState();
+ Replica newLeader = state.getCollection(COLLECTION_NAME).getSlice("shard1").getLeader();
+ return newLeader != null && !newLeader.getNodeName().equals(runner.getNodeName());
+ });
+ closed.set(true);
+ thread.join();
+ }
+
+}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/request/CoreAdminRequest.java b/solr/solrj/src/java/org/apache/solr/client/solrj/request/CoreAdminRequest.java
index 692b54d..af57cb9 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/request/CoreAdminRequest.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/request/CoreAdminRequest.java
@@ -406,6 +406,12 @@ public class CoreAdminRequest extends SolrRequest<CoreAdminResponse> {
}
}
+ public static class AbdicateLeadership extends CoreAdminRequest {
+ public AbdicateLeadership() {
+ action = CoreAdminAction.ABDICATE_LEADERSHIP;
+ }
+ }
+
public static class Unload extends CoreAdminRequest {
protected boolean deleteIndex;
protected boolean deleteDataDir;
@@ -643,6 +649,10 @@ public class CoreAdminRequest extends SolrRequest<CoreAdminResponse> {
return new CoreStatus(req.process(client).getCoreStatus(coreName));
}
+ public static CoreAdminResponse abdicateLeadership(SolrClient client) throws IOException, SolrServerException {
+ return new AbdicateLeadership().process(client);
+ }
+
public static CoreAdminResponse getStatus( String name, SolrClient client ) throws SolrServerException, IOException
{
CoreAdminRequest req = new CoreAdminRequest();
diff --git a/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java b/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
index 2548e62..daff1fe 100644
--- a/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
+++ b/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
@@ -153,6 +153,7 @@ public abstract class CoreAdminParams
REQUESTBUFFERUPDATES,
REQUESTAPPLYUPDATES,
OVERSEEROP,
+ ABDICATE_LEADERSHIP,
REQUESTSTATUS(true),
REJOINLEADERELECTION,
//internal API used by force shard leader election