You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ab...@apache.org on 2017/11/30 13:47:25 UTC
lucene-solr:jira/solr-11285-sim: SOLR-11403: Fix split shard
refactoring side-effects.
Repository: lucene-solr
Updated Branches:
refs/heads/jira/solr-11285-sim 30ce59556 -> fb408731b
SOLR-11403: Fix split shard refactoring side-effects.
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/fb408731
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/fb408731
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/fb408731
Branch: refs/heads/jira/solr-11285-sim
Commit: fb408731b1aa519460753473b375f114ee712d32
Parents: 30ce595
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Thu Nov 30 14:46:40 2017 +0100
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Thu Nov 30 14:46:40 2017 +0100
----------------------------------------------------------------------
.../org/apache/solr/cloud/ActionThrottle.java | 4 ++
.../org/apache/solr/cloud/AddReplicaCmd.java | 4 +-
.../org/apache/solr/cloud/SplitShardCmd.java | 18 ++---
.../cloud/autoscaling/ExecutePlanAction.java | 4 +-
.../cloud/autoscaling/ScheduledTriggers.java | 33 ++++-----
...verseerCollectionConfigSetProcessorTest.java | 6 ++
.../AutoAddReplicasPlanActionTest.java | 4 +-
.../autoscaling/TriggerIntegrationTest.java | 14 ++--
.../sim/SimClusterStateProvider.java | 72 +++++++++++++++-----
.../autoscaling/sim/SimSolrCloudTestCase.java | 11 ++-
.../autoscaling/sim/TestComputePlanAction.java | 2 +-
.../autoscaling/sim/TestExecutePlanAction.java | 2 +-
.../autoscaling/sim/TestTriggerIntegration.java | 11 +--
13 files changed, 122 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/java/org/apache/solr/cloud/ActionThrottle.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ActionThrottle.java b/solr/core/src/java/org/apache/solr/cloud/ActionThrottle.java
index ff4220d..520a269 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ActionThrottle.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ActionThrottle.java
@@ -55,6 +55,10 @@ public class ActionThrottle {
this.timeSource = timeSource;
}
+ public void reset() {
+ lastActionStartedAt = null;
+ }
+
public void markAttemptingAction() {
lastActionStartedAt = timeSource.getTime();
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
index bfea2fa..5887b78 100644
--- a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
@@ -272,7 +272,9 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
}
}
}
- message = message.plus(CoreAdminParams.CORE_NODE_NAME, coreNodeName);
+ if (coreNodeName != null) {
+ message = message.plus(CoreAdminParams.CORE_NODE_NAME, coreNodeName);
+ }
message = message.plus(CoreAdminParams.NAME, coreName);
message = message.plus(CoreAdminParams.NODE, node);
return message;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
index df15f65..a3970a2 100644
--- a/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
@@ -26,6 +26,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
@@ -86,7 +87,8 @@ public class SplitShardCmd implements Cmd {
log.info("Split shard invoked");
ZkStateReader zkStateReader = ocmh.zkStateReader;
zkStateReader.forceUpdateCollection(collectionName);
- String slice = message.getStr(ZkStateReader.SHARD_ID_PROP);
+ AtomicReference<String> slice = new AtomicReference<>();
+ slice.set(message.getStr(ZkStateReader.SHARD_ID_PROP));
String splitKey = message.getStr("split.key");
DocCollection collection = clusterState.getCollection(collectionName);
@@ -96,7 +98,7 @@ public class SplitShardCmd implements Cmd {
// find the leader for the shard
Replica parentShardLeader = null;
try {
- parentShardLeader = zkStateReader.getLeaderRetry(collectionName, slice, 10000);
+ parentShardLeader = zkStateReader.getLeaderRetry(collectionName, slice.get(), 10000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
@@ -381,7 +383,7 @@ public class SplitShardCmd implements Cmd {
DistributedQueue inQueue = Overseer.getStateUpdateQueue(zkStateReader.getZkClient());
Map<String, Object> propMap = new HashMap<>();
propMap.put(Overseer.QUEUE_OPERATION, OverseerAction.UPDATESHARDSTATE.toLower());
- propMap.put(slice, Slice.State.INACTIVE.toString());
+ propMap.put(slice.get(), Slice.State.INACTIVE.toString());
for (String subSlice : subSlices) {
propMap.put(subSlice, Slice.State.ACTIVE.toString());
}
@@ -410,7 +412,7 @@ public class SplitShardCmd implements Cmd {
log.info("Successfully created all replica shards for all sub-slices " + subSlices);
- ocmh.commit(results, slice, parentShardLeader);
+ ocmh.commit(results, slice.get(), parentShardLeader);
return true;
} catch (SolrException e) {
@@ -423,13 +425,13 @@ public class SplitShardCmd implements Cmd {
}
}
- public static Slice getParentSlice(ClusterState clusterState, String collectionName, String slice, String splitKey) {
+ public static Slice getParentSlice(ClusterState clusterState, String collectionName, AtomicReference<String> slice, String splitKey) {
DocCollection collection = clusterState.getCollection(collectionName);
DocRouter router = collection.getRouter() != null ? collection.getRouter() : DocRouter.DEFAULT;
Slice parentSlice;
- if (slice == null) {
+ if (slice.get() == null) {
if (router instanceof CompositeIdRouter) {
Collection<Slice> searchSlices = router.getSearchSlicesSingle(splitKey, new ModifiableSolrParams(), collection);
if (searchSlices.isEmpty()) {
@@ -440,7 +442,7 @@ public class SplitShardCmd implements Cmd {
"Splitting a split.key: " + splitKey + " which spans multiple shards is not supported");
}
parentSlice = searchSlices.iterator().next();
- slice = parentSlice.getName();
+ slice.set(parentSlice.getName());
log.info("Split by route.key: {}, parent shard is: {} ", splitKey, slice);
} else {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
@@ -448,7 +450,7 @@ public class SplitShardCmd implements Cmd {
+ router.getClass().getName());
}
} else {
- parentSlice = collection.getSlice(slice);
+ parentSlice = collection.getSlice(slice.get());
}
if (parentSlice == null) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
index 578a280..f3965d6 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
@@ -51,7 +51,7 @@ public class ExecutePlanAction extends TriggerActionBase {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private static final String PREFIX = "op-";
- static final int DEFAULT_TASK_TIMEOUT_MS = 120000;
+ static final int DEFAULT_TASK_TIMEOUT = 120;
@Override
public void process(TriggerEvent event, ActionContext context) throws Exception {
@@ -83,7 +83,7 @@ public class ExecutePlanAction extends TriggerActionBase {
}
asyncId = (String)asyncResponse.getResponse().get("requestid");
CollectionAdminRequest.RequestStatusResponse statusResponse = waitForTaskToFinish(cloudManager, asyncId,
- DEFAULT_TASK_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ DEFAULT_TASK_TIMEOUT, TimeUnit.SECONDS);
if (statusResponse != null) {
RequestStatusState state = statusResponse.getRequestStatus();
if (state == RequestStatusState.COMPLETED || state == RequestStatusState.FAILED || state == RequestStatusState.NOT_FOUND) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
index e5f65d6..f22e8af 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
@@ -78,20 +78,19 @@ import static org.apache.solr.common.params.AutoScalingParams.TRIGGER_SCHEDULE_D
*/
public class ScheduledTriggers implements Closeable {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- public static final int DEFAULT_SCHEDULED_TRIGGER_DELAY_MS = 1000;
- public static final int DEFAULT_MIN_BETWEEN_ACTIONS_MS = 5000;
- public static final int DEFAULT_COOLDOWN_PERIOD_MS = 5000;
- static final int DEFAULT_TRIGGER_CORE_POOL_SIZE = 4;
+ public static final int DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS = 1;
+ public static final int DEFAULT_ACTION_THROTTLE_PERIOD_SECONDS = 5;
+ public static final int DEFAULT_COOLDOWN_PERIOD_SECONDS = 5;
+ public static final int DEFAULT_TRIGGER_CORE_POOL_SIZE = 4;
static final Map<String, Object> DEFAULT_PROPERTIES = new HashMap<>();
// Note: values must be all in milliseconds!
static {
- DEFAULT_PROPERTIES.put(TRIGGER_SCHEDULE_DELAY_SECONDS, DEFAULT_SCHEDULED_TRIGGER_DELAY_MS);
- DEFAULT_PROPERTIES.put(TRIGGER_COOLDOWN_PERIOD_SECONDS, DEFAULT_COOLDOWN_PERIOD_MS);
+ DEFAULT_PROPERTIES.put(TRIGGER_SCHEDULE_DELAY_SECONDS, DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS);
+ DEFAULT_PROPERTIES.put(TRIGGER_COOLDOWN_PERIOD_SECONDS, DEFAULT_COOLDOWN_PERIOD_SECONDS);
DEFAULT_PROPERTIES.put(TRIGGER_CORE_POOL_SIZE, DEFAULT_TRIGGER_CORE_POOL_SIZE);
- DEFAULT_PROPERTIES.put(ACTION_THROTTLE_PERIOD_SECONDS, DEFAULT_MIN_BETWEEN_ACTIONS_MS);
+ DEFAULT_PROPERTIES.put(ACTION_THROTTLE_PERIOD_SECONDS, DEFAULT_ACTION_THROTTLE_PERIOD_SECONDS);
}
private final Map<String, ScheduledTrigger> scheduledTriggers = new ConcurrentHashMap<>();
@@ -114,9 +113,9 @@ public class ScheduledTriggers implements Closeable {
private final AtomicLong cooldownStart = new AtomicLong();
- private final AtomicLong cooldownPeriod = new AtomicLong(TimeUnit.MILLISECONDS.toNanos(DEFAULT_COOLDOWN_PERIOD_MS));
+ private final AtomicLong cooldownPeriod = new AtomicLong(TimeUnit.SECONDS.toNanos(DEFAULT_COOLDOWN_PERIOD_SECONDS));
- private final AtomicLong triggerDelay = new AtomicLong(DEFAULT_SCHEDULED_TRIGGER_DELAY_MS);
+ private final AtomicLong triggerDelay = new AtomicLong(DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS);
private final AtomicReference<ActionThrottle> actionThrottle;
@@ -138,14 +137,13 @@ public class ScheduledTriggers implements Closeable {
scheduledThreadPoolExecutor.setRemoveOnCancelPolicy(true);
scheduledThreadPoolExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
actionExecutor = ExecutorUtil.newMDCAwareSingleThreadExecutor(new DefaultSolrThreadFactory("AutoscalingActionExecutor"));
- actionThrottle = new AtomicReference<>(new ActionThrottle("action", DEFAULT_MIN_BETWEEN_ACTIONS_MS, cloudManager.getTimeSource()));
+ actionThrottle = new AtomicReference<>(new ActionThrottle("action", TimeUnit.SECONDS.toMillis(DEFAULT_ACTION_THROTTLE_PERIOD_SECONDS), cloudManager.getTimeSource()));
this.cloudManager = cloudManager;
this.stateManager = cloudManager.getDistribStateManager();
this.loader = loader;
queueStats = new Stats();
listeners = new TriggerListeners();
// initialize cooldown timer
- // todo: make the cooldownPeriod configurable
cooldownStart.set(cloudManager.getTimeSource().getTime() - cooldownPeriod.get());
}
@@ -166,13 +164,13 @@ public class ScheduledTriggers implements Closeable {
log.debug("Changing value of autoscaling property: {} from: {} to: {}", key, entry.getValue(), newProps.get(key));
switch (key) {
case TRIGGER_SCHEDULE_DELAY_SECONDS:
- triggerDelay.set(TimeUnit.MILLISECONDS.convert(((Number) newProps.get(key)).longValue(), TimeUnit.SECONDS));
+ triggerDelay.set(((Number) newProps.get(key)).intValue());
synchronized (this) {
scheduledTriggers.forEach((s, scheduledTrigger) -> {
if (scheduledTrigger.scheduledFuture.cancel(false)) {
scheduledTrigger.scheduledFuture = scheduledThreadPoolExecutor.scheduleWithFixedDelay(
scheduledTrigger, 0,
- cloudManager.getTimeSource().convertDelay(TimeUnit.MILLISECONDS, triggerDelay.get(), TimeUnit.MILLISECONDS),
+ cloudManager.getTimeSource().convertDelay(TimeUnit.SECONDS, triggerDelay.get(), TimeUnit.MILLISECONDS),
TimeUnit.MILLISECONDS);
} else {
log.debug("Failed to cancel scheduled task: {}", s);
@@ -337,7 +335,7 @@ public class ScheduledTriggers implements Closeable {
hasPendingActions.set(false);
}
log.debug("-- processing took {} ms for event id={}",
- TimeUnit.MILLISECONDS.convert(cloudManager.getTimeSource().getTime() - eventProcessingStart, TimeUnit.NANOSECONDS), event.id);
+ TimeUnit.NANOSECONDS.toMillis(cloudManager.getTimeSource().getTime() - eventProcessingStart), event.id);
});
} else {
if (enqueued) {
@@ -359,7 +357,7 @@ public class ScheduledTriggers implements Closeable {
});
newTrigger.init(); // mark as ready for scheduling
scheduledTrigger.scheduledFuture = scheduledThreadPoolExecutor.scheduleWithFixedDelay(scheduledTrigger, 0,
- cloudManager.getTimeSource().convertDelay(TimeUnit.MILLISECONDS, triggerDelay.get(), TimeUnit.MILLISECONDS),
+ cloudManager.getTimeSource().convertDelay(TimeUnit.SECONDS, triggerDelay.get(), TimeUnit.MILLISECONDS),
TimeUnit.MILLISECONDS);
}
@@ -384,8 +382,7 @@ public class ScheduledTriggers implements Closeable {
try {
log.debug("Found pending task with requestid={}", requestid);
RequestStatusResponse statusResponse = waitForTaskToFinish(cloudManager, requestid,
- cloudManager.getTimeSource().convertDelay(TimeUnit.MILLISECONDS, ExecutePlanAction.DEFAULT_TASK_TIMEOUT_MS, TimeUnit.MILLISECONDS),
- TimeUnit.MILLISECONDS);
+ ExecutePlanAction.DEFAULT_TASK_TIMEOUT, TimeUnit.SECONDS);
if (statusResponse != null) {
RequestStatusState state = statusResponse.getRequestStatus();
if (state == RequestStatusState.COMPLETED || state == RequestStatusState.FAILED || state == RequestStatusState.NOT_FOUND) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java b/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
index 8e28c9e..8a46808 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
@@ -44,6 +44,7 @@ import org.apache.solr.common.params.CollectionParams;
import org.apache.solr.common.params.CoreAdminParams;
import org.apache.solr.common.params.CoreAdminParams.CoreAdminAction;
import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ObjectCache;
import org.apache.solr.common.util.StrUtils;
import org.apache.solr.common.util.TimeSource;
import org.apache.solr.common.util.Utils;
@@ -88,6 +89,7 @@ public class OverseerCollectionConfigSetProcessorTest extends SolrTestCaseJ4 {
private static ClusterState clusterStateMock;
private static SolrZkClient solrZkClientMock;
private static DistribStateManager stateManagerMock;
+ private static ObjectCache objectCache;
private static AutoScalingConfig autoScalingConfig = new AutoScalingConfig(Collections.emptyMap());
private final Map zkMap = new HashMap();
private final Map<String, ClusterState.CollectionRef> collectionsSet = new HashMap<>();
@@ -137,6 +139,7 @@ public class OverseerCollectionConfigSetProcessorTest extends SolrTestCaseJ4 {
overseerMock = mock(Overseer.class);
zkControllerMock = mock(ZkController.class);
cloudDataProviderMock = mock(SolrCloudManager.class);
+ objectCache = new ObjectCache();
clusterStateProviderMock = mock(ClusterStateProvider.class);
stateManagerMock = mock(DistribStateManager.class);
}
@@ -174,6 +177,9 @@ public class OverseerCollectionConfigSetProcessorTest extends SolrTestCaseJ4 {
reset(overseerMock);
reset(zkControllerMock);
reset(cloudDataProviderMock);
+ objectCache.clear();
+ when(cloudDataProviderMock.getObjectCache()).thenReturn(objectCache);
+ when(cloudDataProviderMock.getTimeSource()).thenReturn(TimeSource.NANO_TIME);
reset(clusterStateProviderMock);
reset(stateManagerMock);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoAddReplicasPlanActionTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoAddReplicasPlanActionTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoAddReplicasPlanActionTest.java
index 4c8ed69..d31bba5 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoAddReplicasPlanActionTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoAddReplicasPlanActionTest.java
@@ -37,6 +37,7 @@ import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.SuppressForbidden;
+import org.apache.solr.common.util.TimeSource;
import org.apache.solr.util.TimeOut;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -142,8 +143,7 @@ public class AutoAddReplicasPlanActionTest extends SolrCloudTestCase{
private void waitForNodeLeave(String lostNodeName) throws InterruptedException {
ZkStateReader reader = cluster.getSolrClient().getZkStateReader();
- TimeOut timeOut = new TimeOut(10, TimeUnit.SECONDS,
- cluster.getJettySolrRunner(0).getCoreContainer().getZkController().getSolrCloudManager().getTimeSource());
+ TimeOut timeOut = new TimeOut(10, TimeUnit.SECONDS, TimeSource.NANO_TIME);
while (reader.getClusterState().getLiveNodes().contains(lostNodeName)) {
Thread.sleep(100);
if (timeOut.hasTimedOut()) fail("Wait for " + lostNodeName + " to leave failed!");
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
index cd279bb..f7e1a62 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
@@ -67,7 +67,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.solr.cloud.autoscaling.AutoScalingHandlerTest.createAutoScalingRequest;
-import static org.apache.solr.cloud.autoscaling.ScheduledTriggers.DEFAULT_SCHEDULED_TRIGGER_DELAY_MS;
+import static org.apache.solr.cloud.autoscaling.ScheduledTriggers.DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS;
import static org.apache.solr.common.cloud.ZkStateReader.SOLR_AUTOSCALING_CONF_PATH;
/**
@@ -120,7 +120,7 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
@Before
public void setupTest() throws Exception {
- throttlingDelayMs.set(ScheduledTriggers.DEFAULT_MIN_BETWEEN_ACTIONS_MS);
+ throttlingDelayMs.set(TimeUnit.SECONDS.toMillis(ScheduledTriggers.DEFAULT_ACTION_THROTTLE_PERIOD_SECONDS));
waitForSeconds = 1 + random().nextInt(3);
actionConstructorCalled = new CountDownLatch(1);
actionInitCalled = new CountDownLatch(1);
@@ -257,7 +257,7 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
}
static AtomicLong lastActionExecutedAt = new AtomicLong(0);
- static AtomicLong throttlingDelayMs = new AtomicLong(ScheduledTriggers.DEFAULT_MIN_BETWEEN_ACTIONS_MS);
+ static AtomicLong throttlingDelayMs = new AtomicLong(TimeUnit.SECONDS.toMillis(ScheduledTriggers.DEFAULT_ACTION_THROTTLE_PERIOD_SECONDS));
static ReentrantLock lock = new ReentrantLock();
public static class ThrottlingTesterAction extends TestTriggerAction {
// nanos are very precise so we need a delta for comparison with ms
@@ -334,7 +334,7 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
cluster.stopJettySolrRunner(index);
// ensure that the old trigger sees the stopped node, todo find a better way to do this
- Thread.sleep(500 + DEFAULT_SCHEDULED_TRIGGER_DELAY_MS);
+ Thread.sleep(500 + TimeUnit.SECONDS.toMillis(DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS));
waitForSeconds = 0;
setTriggerCommand = "{" +
@@ -392,7 +392,7 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
JettySolrRunner newNode = cluster.startJettySolrRunner();
// ensure that the old trigger sees the new node, todo find a better way to do this
- Thread.sleep(500 + DEFAULT_SCHEDULED_TRIGGER_DELAY_MS);
+ Thread.sleep(500 + TimeUnit.SECONDS.toMillis(DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS));
waitForSeconds = 0;
setTriggerCommand = "{" +
@@ -1194,7 +1194,7 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
assertEquals(ev.toString(), TriggerEventProcessorStage.SUCCEEDED, ev.stage);
// the difference between timestamps of the first SUCCEEDED and the last SUCCEEDED
// must be larger than cooldown period
- assertTrue("timestamp delta is less than default cooldown period", ev.timestamp - prevTimestamp > TimeUnit.MILLISECONDS.toNanos(ScheduledTriggers.DEFAULT_COOLDOWN_PERIOD_MS));
+ assertTrue("timestamp delta is less than default cooldown period", ev.timestamp - prevTimestamp > TimeUnit.SECONDS.toNanos(ScheduledTriggers.DEFAULT_COOLDOWN_PERIOD_SECONDS));
prevTimestamp = ev.timestamp;
long modifiedCooldownPeriodSeconds = 7;
@@ -1267,7 +1267,7 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
}
});
assertTrue(getTriggerFiredLatch().await(4, TimeUnit.SECONDS));
- assertTrue(diff.get() - TimeUnit.MILLISECONDS.toNanos(ScheduledTriggers.DEFAULT_SCHEDULED_TRIGGER_DELAY_MS) >= 0);
+ assertTrue(diff.get() - TimeUnit.SECONDS.toNanos(ScheduledTriggers.DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS) >= 0);
// change schedule delay
config = config.withProperties(Collections.singletonMap(AutoScalingParams.TRIGGER_SCHEDULE_DELAY_SECONDS, 4));
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
index b5b91a2..1f1788b 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
@@ -154,6 +154,10 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
}
+ public void simResetLeaderThrottle() {
+ leaderThrottle.reset();
+ }
+
public String simGetRandomNode(Random random) {
if (liveNodes.isEmpty()) {
return null;
@@ -222,7 +226,12 @@ public class SimClusterStateProvider implements ClusterStateProvider {
public void simAddReplica(ZkNodeProps message, NamedList results) throws Exception {
AtomicLong policyVersionAfter = new AtomicLong(-1);
- message = AddReplicaCmd.assignReplicaDetails(cloudManager, getClusterState(), message, policyVersionAfter);
+ ClusterState clusterState = getClusterState();
+ DocCollection coll = clusterState.getCollection(message.getStr(ZkStateReader.COLLECTION_PROP));
+ message = AddReplicaCmd.assignReplicaDetails(cloudManager, clusterState, message, policyVersionAfter);
+ if (message.getStr(CoreAdminParams.CORE_NODE_NAME) == null) {
+ message = message.plus(CoreAdminParams.CORE_NODE_NAME, Assign.assignCoreNodeName(stateManager, coll));
+ }
ReplicaInfo ri = new ReplicaInfo(
message.getStr(CoreAdminParams.CORE_NODE_NAME),
message.getStr(CoreAdminParams.NAME),
@@ -248,6 +257,24 @@ public class SimClusterStateProvider implements ClusterStateProvider {
if (!liveNodes.contains(nodeId)) {
throw new Exception("Target node " + nodeId + " is not live: " + liveNodes);
}
+ // verify info
+ if (replicaInfo.getCore() == null) {
+ throw new Exception("Missing core: " + replicaInfo);
+ }
+// if (replicaInfo.getShard() == null) {
+// throw new Exception("Missing shard: " + replicaInfo);
+// }
+ replicaInfo.getVariables().remove(ZkStateReader.SHARD_ID_PROP);
+ if (replicaInfo.getName() == null) {
+ throw new Exception("Missing name: " + replicaInfo);
+ }
+ if (replicaInfo.getNode() == null) {
+ throw new Exception("Missing node: " + replicaInfo);
+ }
+ if (!replicaInfo.getNode().equals(nodeId)) {
+ throw new Exception("Wrong node (not " + nodeId + "): " + replicaInfo);
+ }
+
lock.lock();
try {
@@ -362,14 +389,16 @@ public class SimClusterStateProvider implements ClusterStateProvider {
if (ri == null) {
throw new IllegalStateException("-- could not find ReplicaInfo for replica " + r);
}
- if (ri.getVariables().remove(ZkStateReader.LEADER_PROP) != null) {
- stateChanged.set(true);
- }
- if (r.isActive(liveNodes)) {
- active.add(ri);
- } else { // if it's on a node that is not live mark it down
- if (!liveNodes.contains(r.getNodeName())) {
- ri.getVariables().put(ZkStateReader.STATE_PROP, Replica.State.DOWN.toString());
+ synchronized (ri) {
+ if (ri.getVariables().remove(ZkStateReader.LEADER_PROP) != null) {
+ stateChanged.set(true);
+ }
+ if (r.isActive(liveNodes)) {
+ active.add(ri);
+ } else { // if it's on a node that is not live mark it down
+ if (!liveNodes.contains(r.getNodeName())) {
+ ri.getVariables().put(ZkStateReader.STATE_PROP, Replica.State.DOWN.toString());
+ }
}
}
});
@@ -389,7 +418,9 @@ public class SimClusterStateProvider implements ClusterStateProvider {
LOG.warn("-- can't find any suitable replica type for " + dc.getName() + " / " + s.getName());
return;
}
- ri.getVariables().put(ZkStateReader.LEADER_PROP, "true");
+ synchronized (ri) {
+ ri.getVariables().put(ZkStateReader.LEADER_PROP, "true");
+ }
stateChanged.set(true);
LOG.info("-- elected new leader for " + dc.getName() + " / " + s.getName() + ": " + ri);
} else {
@@ -425,7 +456,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
AtomicInteger replicaNum = new AtomicInteger(1);
replicaPositions.forEach(pos -> {
Map<String, Object> replicaProps = new HashMap<>();
- replicaProps.put(ZkStateReader.SHARD_ID_PROP, pos.shard);
+ //replicaProps.put(ZkStateReader.SHARD_ID_PROP, pos.shard);
replicaProps.put(ZkStateReader.NODE_NAME_PROP, pos.node);
replicaProps.put(ZkStateReader.REPLICA_TYPE, pos.type.toString());
String coreName = String.format(Locale.ROOT, "%s_%s_replica_%s%s", collectionName, pos.shard, pos.type.name().substring(0,1).toLowerCase(Locale.ROOT),
@@ -542,7 +573,8 @@ public class SimClusterStateProvider implements ClusterStateProvider {
String newSolrCoreName = Assign.buildSolrCoreName(stateManager, coll, slice.getName(), replica.getType());
String coreNodeName = Assign.assignCoreNodeName(stateManager, coll);
- ReplicaInfo newReplica = new ReplicaInfo(coreNodeName, newSolrCoreName, collection, slice.getName(), Replica.Type.NRT, targetNode, null);
+ ReplicaInfo newReplica = new ReplicaInfo(coreNodeName, newSolrCoreName, collection, slice.getName(), replica.getType(), targetNode, null);
+ LOG.debug("-- new replica: " + newReplica);
// xxx should run leader election here already?
simAddReplica(targetNode, newReplica, false);
// this will trigger leader election
@@ -608,7 +640,8 @@ public class SimClusterStateProvider implements ClusterStateProvider {
public void simSplitShard(ZkNodeProps message, NamedList results) throws Exception {
String collectionName = message.getStr(COLLECTION_PROP);
- String sliceName = message.getStr(SHARD_ID_PROP);
+ AtomicReference<String> sliceName = new AtomicReference<>();
+ sliceName.set(message.getStr(SHARD_ID_PROP));
String splitKey = message.getStr("split.key");
ClusterState clusterState = getClusterState();
DocCollection collection = clusterState.getCollection(collectionName);
@@ -622,7 +655,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
SplitShardCmd.fillRanges(cloudManager, message, collection, parentSlice, subRanges, subSlices, subShardNames);
// mark the old slice as inactive
sliceProperties.computeIfAbsent(collectionName, c -> new ConcurrentHashMap<>())
- .computeIfAbsent(sliceName, s -> new ConcurrentHashMap<>())
+ .computeIfAbsent(sliceName.get(), s -> new ConcurrentHashMap<>())
.put(ZkStateReader.SHARD_STATE_PROP, Slice.State.INACTIVE.toString());
// add slice props
for (int i = 0; i < subRanges.size(); i++) {
@@ -631,7 +664,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
Map<String, Object> sliceProps = sliceProperties.computeIfAbsent(collectionName, c -> new ConcurrentHashMap<>())
.computeIfAbsent(subSlice, ss -> new ConcurrentHashMap<>());
sliceProps.put(Slice.RANGE, range);
- sliceProps.put(Slice.PARENT, sliceName);
+ sliceProps.put(Slice.PARENT, sliceName.get());
sliceProps.put(ZkStateReader.SHARD_STATE_PROP, Slice.State.ACTIVE.toString());
}
// add replicas for new subShards
@@ -860,8 +893,15 @@ public class SimClusterStateProvider implements ClusterStateProvider {
Map<String, Map<String, Map<String, Replica>>> collMap = new HashMap<>();
nodeReplicaMap.forEach((n, replicas) -> {
replicas.forEach(ri -> {
- Map<String, Object> props = new HashMap<>(ri.getVariables());
+ Map<String, Object> props;
+ synchronized (ri) {
+ props = new HashMap<>(ri.getVariables());
+ }
props.put(ZkStateReader.NODE_NAME_PROP, n);
+ //props.put(ZkStateReader.SHARD_ID_PROP, ri.getShard());
+ props.put(ZkStateReader.CORE_NAME_PROP, ri.getCore());
+ props.put(ZkStateReader.REPLICA_TYPE, ri.getType().toString());
+ props.put(ZkStateReader.STATE_PROP, ri.getState().toString());
Replica r = new Replica(ri.getName(), props);
collMap.computeIfAbsent(ri.getCollection(), c -> new HashMap<>())
.computeIfAbsent(ri.getShard(), s -> new HashMap<>())
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
index d7e075b..9324be1 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
@@ -60,6 +60,14 @@ public class SimSolrCloudTestCase extends SolrTestCaseJ4 {
cluster = null;
}
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ if (cluster != null) {
+ cluster.getSimClusterStateProvider().simResetLeaderThrottle();
+ }
+ }
+
@Before
public void checkClusterConfiguration() {
if (cluster == null)
@@ -110,9 +118,8 @@ public class SimSolrCloudTestCase extends SolrTestCaseJ4 {
return predicate.matches(n, c);
});
} catch (Exception e) {
- fail(message + "\n" + e.toString() + "\nLive Nodes: " + liveNodesLastSeen.get() + "\nLast available state: " + state.get());
+ throw new AssertionError(message + "\n" + "Live Nodes: " + liveNodesLastSeen.get() + "\nLast available state: " + state.get(), e);
}
- return 0;
}
/**
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestComputePlanAction.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestComputePlanAction.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestComputePlanAction.java
index e10dfc4..5ce0af3 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestComputePlanAction.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestComputePlanAction.java
@@ -132,7 +132,7 @@ public class TestComputePlanAction extends SimSolrCloudTestCase {
rsp = cluster.request(req);
response = rsp.getResponse();
assertEquals(response.get("result").toString(), "success");
- cluster.getTimeSource().sleep(ScheduledTriggers.DEFAULT_COOLDOWN_PERIOD_MS);
+ cluster.getTimeSource().sleep(TimeUnit.SECONDS.toMillis(ScheduledTriggers.DEFAULT_COOLDOWN_PERIOD_SECONDS));
}
@After
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
index 637f2d6..bbc1317 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
@@ -166,7 +166,7 @@ public class TestExecutePlanAction extends SimSolrCloudTestCase {
assertNotNull(response.get("success"));
}
- log.info("Collection ready after " + waitForState(collectionName, 120, TimeUnit.SECONDS, clusterShape(1, 2)) + "ms");
+ log.info("Collection ready after " + waitForState(collectionName, 300, TimeUnit.SECONDS, clusterShape(1, 2)) + "ms");
}
@Test
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/fb408731/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestTriggerIntegration.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestTriggerIntegration.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestTriggerIntegration.java
index f95ef0a..4926e26 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestTriggerIntegration.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestTriggerIntegration.java
@@ -57,7 +57,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.solr.cloud.autoscaling.AutoScalingHandlerTest.createAutoScalingRequest;
-import static org.apache.solr.cloud.autoscaling.ScheduledTriggers.DEFAULT_SCHEDULED_TRIGGER_DELAY_MS;
+import static org.apache.solr.cloud.autoscaling.ScheduledTriggers.DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS;
/**
* An end-to-end integration test for triggers
@@ -239,7 +239,8 @@ public class TestTriggerIntegration extends SimSolrCloudTestCase {
try {
if (lastActionExecutedAt.get() != 0) {
log.info("last action at " + lastActionExecutedAt.get() + " time = " + cluster.getTimeSource().getTime());
- if (TimeUnit.MILLISECONDS.convert(cluster.getTimeSource().getTime() - lastActionExecutedAt.get(), TimeUnit.NANOSECONDS) < ScheduledTriggers.DEFAULT_MIN_BETWEEN_ACTIONS_MS - DELTA_MS) {
+ if (TimeUnit.NANOSECONDS.toMillis(cluster.getTimeSource().getTime() - lastActionExecutedAt.get()) <
+ TimeUnit.SECONDS.toMillis(ScheduledTriggers.DEFAULT_ACTION_THROTTLE_PERIOD_SECONDS) - DELTA_MS) {
log.info("action executed again before minimum wait time from {}", event.getSource());
fail("TriggerListener was fired before the throttling period");
}
@@ -289,7 +290,7 @@ public class TestTriggerIntegration extends SimSolrCloudTestCase {
cluster.simRemoveNode(nodeName);
// ensure that the old trigger sees the stopped node, todo find a better way to do this
- timeOut.sleep(500 + DEFAULT_SCHEDULED_TRIGGER_DELAY_MS);
+ timeOut.sleep(500 + TimeUnit.SECONDS.toMillis(DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS));
waitForSeconds = 0;
setTriggerCommand = "{" +
@@ -347,7 +348,7 @@ public class TestTriggerIntegration extends SimSolrCloudTestCase {
String newNode = cluster.simAddNode();
// ensure that the old trigger sees the new node, todo find a better way to do this
- cluster.getTimeSource().sleep(500 + DEFAULT_SCHEDULED_TRIGGER_DELAY_MS);
+ cluster.getTimeSource().sleep(500 + TimeUnit.SECONDS.toMillis(DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS));
waitForSeconds = 0;
setTriggerCommand = "{" +
@@ -1148,7 +1149,7 @@ public class TestTriggerIntegration extends SimSolrCloudTestCase {
assertEquals(ev.toString(), TriggerEventProcessorStage.SUCCEEDED, ev.stage);
// the difference between timestamps of the first SUCCEEDED and the last SUCCEEDED
// must be larger than cooldown period
- assertTrue("timestamp delta is less than default cooldown period", ev.timestamp - prevTimestamp > TimeUnit.MILLISECONDS.toNanos(ScheduledTriggers.DEFAULT_COOLDOWN_PERIOD_MS));
+ assertTrue("timestamp delta is less than default cooldown period", ev.timestamp - prevTimestamp > TimeUnit.SECONDS.toNanos(ScheduledTriggers.DEFAULT_COOLDOWN_PERIOD_SECONDS));
}
public static class TestSearchRateAction extends TriggerActionBase {