You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by jx...@apache.org on 2019/02/27 04:13:12 UTC
[helix] branch master updated (ce9a662 -> 4ecd923)
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git.
from ce9a662 HELIX instance level validation util for rest API
new db33ae5 [HELIX-806] HELIX: Modify endpoints for instrumenting maintenance with custom fields
new a827f8f [HELIX-807] REST: Add get maintenance signal endpoint
new 70b6355 [HELIX-808] TASK: Fix double-booking of tasks with task CurrentStates
new ae586ac [HELIX-809] TEST: Fix unstable TestClusterInMaintenanceModeWhenReachingMaxPartition
new c1f9af5 [HELIX-810] HELIX: Fix NPE in InstanceMessagesCache
new 7daecd0 [HELIX-811] HELIX: Only log relayMsg if it doesn't exist
new 4ecd923 [HELIX-812] HELIX: Fix maintenance history bug
The 7 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:
.../helix/common/caches/InstanceMessagesCache.java | 143 +++++++++++----------
.../dataproviders/BaseControllerDataProvider.java | 12 ++
.../stages/MaintenanceRecoveryStage.java | 8 ++
.../apache/helix/task/AbstractTaskDispatcher.java | 35 ++++-
.../java/org/apache/helix/task/JobDispatcher.java | 44 +++++--
.../controller/TestClusterMaintenanceMode.java | 1 -
...rInMaintenanceModeWhenReachingMaxPartition.java | 4 +-
.../helix/integration/task/TestNoDoubleAssign.java | 7 +-
.../rest/server/resources/AbstractResource.java | 1 -
.../server/resources/helix/ClusterAccessor.java | 50 ++++---
.../helix/rest/server/TestClusterAccessor.java | 39 +++---
11 files changed, 230 insertions(+), 114 deletions(-)
[helix] 03/07: [HELIX-808] TASK: Fix double-booking of tasks with
task CurrentStates
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 70b63558797d193164a08bc73eb99816cf3a6094
Author: narendly <na...@gmail.com>
AuthorDate: Tue Feb 26 17:24:56 2019 -0800
[HELIX-808] TASK: Fix double-booking of tasks with task CurrentStates
It was observed that TestNoDoubleAssign was failing intermittently. Upon debugging with more detailed logs, there was a race condition between newly starting tasks and dropping tasks. To prevent this, dropping state transitions will be prioritized and prevInstanceToTaskAssignment will be built from CurrentStates. This is needed to make sure the right number of tasks are assigned every task pipeline and dropping transitions happen right away.
Changelist:
1\. Change the logic for generating prevInstToTaskAssignment so that it's based on CurrentState
2\. Add a special check for not updating task partition state upon Participant connection loss
3\. TestNoDoubleAssign passes consistently
4. Fix TestNoDoubleAssign so that there won't be any thread leak
---
.../apache/helix/task/AbstractTaskDispatcher.java | 35 +++++++++++++++--
.../java/org/apache/helix/task/JobDispatcher.java | 44 ++++++++++++++++++----
.../helix/integration/task/TestNoDoubleAssign.java | 7 ++--
3 files changed, 72 insertions(+), 14 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java b/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java
index 1bfab8b..b853198 100644
--- a/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java
+++ b/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java
@@ -47,20 +47,42 @@ public abstract class AbstractTaskDispatcher {
ResourceAssignment prevTaskToInstanceStateAssignment, TaskState jobState,
Map<String, Set<Integer>> assignedPartitions, Set<Integer> partitionsToDropFromIs,
Map<Integer, PartitionAssignment> paMap, TargetState jobTgtState,
- Set<Integer> skippedPartitions, WorkflowControllerDataProvider cache) {
+ Set<Integer> skippedPartitions, WorkflowControllerDataProvider cache,
+ Map<String, Set<Integer>> tasksToDrop) {
// Get AssignableInstanceMap for releasing resources for tasks in terminal states
AssignableInstanceManager assignableInstanceManager = cache.getAssignableInstanceManager();
// Iterate through all instances
for (String instance : prevInstanceToTaskAssignments.keySet()) {
+ assignedPartitions.put(instance, new HashSet<Integer>());
+
+ // Set all dropping transitions first. These are tasks coming from Participant disconnects
+ // that have some active current state (INIT or RUNNING) and the requestedState of DROPPED.
+ // These need to be prioritized over any other state transitions because of the race condition
+ // with the same pId (task) running on other instances. This is because in paMap, we can only
+ // define one transition per pId
+ if (tasksToDrop.containsKey(instance)) {
+ for (int pIdToDrop : tasksToDrop.get(instance)) {
+ paMap.put(pIdToDrop,
+ new PartitionAssignment(instance, TaskPartitionState.DROPPED.name()));
+ assignedPartitions.get(instance).add(pIdToDrop);
+ }
+ }
+
if (excludedInstances.contains(instance)) {
continue;
}
// If not an excluded instance, we must instantiate its entry in assignedPartitions
- assignedPartitions.put(instance, new HashSet<Integer>());
Set<Integer> pSet = prevInstanceToTaskAssignments.get(instance);
+
+ // We need to remove all task pId's to be dropped because we already made an assignment in
+ // paMap above for them to be dropped. The following does this.
+ if (tasksToDrop.containsKey(instance)) {
+ pSet.removeAll(tasksToDrop.get(instance));
+ }
+
// Used to keep track of partitions that are in one of the final states: COMPLETED, TIMED_OUT,
// TASK_ERROR, ERROR.
Set<Integer> donePartitions = new TreeSet<>();
@@ -123,7 +145,14 @@ public abstract class AbstractTaskDispatcher {
continue;
}
- paMap.put(pId, new PartitionAssignment(instance, requestedState.name()));
+ // This contains check is necessary because we have already traversed pIdsToDrop at the
+ // beginning of this method. If we already have a dropping transition, we do not want to
+ // overwrite it. Any other requestedState transitions (for example, INIT to RUNNING or
+ // RUNNING to COMPLETE, can wait without affecting correctness - they will be picked up
+ // in ensuing runs of the Task pipeline)
+ if (!paMap.containsKey(pId)) {
+ paMap.put(pId, new PartitionAssignment(instance, requestedState.name()));
+ }
assignedPartitions.get(instance).add(pId);
if (LOG.isDebugEnabled()) {
LOG.debug(
diff --git a/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java b/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java
index 2ae5b2f..596b54b 100644
--- a/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java
+++ b/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java
@@ -204,9 +204,14 @@ public class JobDispatcher extends AbstractTaskDispatcher {
return new ResourceAssignment(jobResource);
}
+ // This set contains all task pIds that need to be dropped because requestedState is DROPPED
+ // Newer versions of Participants, upon connection reset, sets task requestedStates to DROPPED
+ // These dropping transitions will be prioritized above all task state transition assignments
+ Map<String, Set<Integer>> tasksToDrop = new HashMap<>();
+
Map<String, SortedSet<Integer>> prevInstanceToTaskAssignments =
getPrevInstanceToTaskAssignments(liveInstances, prevTaskToInstanceStateAssignment,
- allPartitions, currStateOutput, jobResource);
+ allPartitions, currStateOutput, jobResource, tasksToDrop);
long currentTime = System.currentTimeMillis();
if (LOG.isDebugEnabled()) {
@@ -217,14 +222,15 @@ public class JobDispatcher extends AbstractTaskDispatcher {
// Release resource for tasks in terminal state
updatePreviousAssignedTasksStatus(prevInstanceToTaskAssignments, excludedInstances, jobResource,
currStateOutput, jobCtx, jobCfg, prevTaskToInstanceStateAssignment, jobState,
- assignedPartitions, partitionsToDropFromIs, paMap, jobTgtState, skippedPartitions, cache);
+ assignedPartitions, partitionsToDropFromIs, paMap, jobTgtState, skippedPartitions, cache,
+ tasksToDrop);
addGiveupPartitions(skippedPartitions, jobCtx, allPartitions, jobCfg);
if (jobState == TaskState.IN_PROGRESS && skippedPartitions.size() > jobCfg.getFailureThreshold()
|| (jobCfg.getTargetResource() != null
- && cache.getIdealState(jobCfg.getTargetResource()) != null
- && !cache.getIdealState(jobCfg.getTargetResource()).isEnabled())) {
+ && cache.getIdealState(jobCfg.getTargetResource()) != null
+ && !cache.getIdealState(jobCfg.getTargetResource()).isEnabled())) {
if (isJobFinished(jobCtx, jobResource, currStateOutput)) {
failJob(jobResource, workflowCtx, jobCtx, workflowConfig, cache.getJobConfigMap(), cache);
return buildEmptyAssignment(jobResource, currStateOutput);
@@ -345,17 +351,20 @@ public class JobDispatcher extends AbstractTaskDispatcher {
* @param currStateOutput currentStates to make sure currentStates copied over expired sessions
* are accounted for
* @param jobName job name
+ * @param tasksToDrop instance -> pId's, to gather all pIds that need to be dropped
* @return instance -> partitionIds from previous assignment, if the instance is still live
*/
private static Map<String, SortedSet<Integer>> getPrevInstanceToTaskAssignments(
Iterable<String> liveInstances, ResourceAssignment prevAssignment,
- Set<Integer> allTaskPartitions, CurrentStateOutput currStateOutput, String jobName) {
+ Set<Integer> allTaskPartitions, CurrentStateOutput currStateOutput, String jobName,
+ Map<String, Set<Integer>> tasksToDrop) {
Map<String, SortedSet<Integer>> result = new HashMap<>();
for (String instance : liveInstances) {
result.put(instance, new TreeSet<Integer>());
}
// First, add all task partitions from JobContext
+ // TODO: Remove this portion to get rid of prevAssignment from Task Framework
for (Partition partition : prevAssignment.getMappedPartitions()) {
int pId = TaskUtil.getPartitionId(partition.getPartitionName());
if (allTaskPartitions.contains(pId)) {
@@ -369,6 +378,8 @@ public class JobDispatcher extends AbstractTaskDispatcher {
}
}
+ // Generate prevInstanceToTaskAssignment with CurrentStateOutput as source of truth
+
// Add all pIds existing in CurrentStateOutput as well because task currentStates copied over
// from previous sessions won't show up in prevInstanceToTaskAssignments
// We need to add these back here in order for these task partitions to be dropped (after a
@@ -381,10 +392,27 @@ public class JobDispatcher extends AbstractTaskDispatcher {
String instance = instanceToCurrState.getKey();
String requestedState =
currStateOutput.getRequestedState(jobName, entry.getKey(), instance);
+ TaskPartitionState currState = TaskPartitionState.valueOf(instanceToCurrState.getValue());
int pId = TaskUtil.getPartitionId(entry.getKey().getPartitionName());
- if (result.containsKey(instance) && requestedState != null
- && requestedState.equals(TaskPartitionState.DROPPED.name())) {
- // Only if this instance is live and requestedState is DROPPED
+
+ // We must add all active task pIds back here because dropping transition could overwrite an
+ // active transition in paMap
+ // Add all task partitions in the following states:
+ // currState = INIT, requestedState = RUNNING (bootstrap)
+ // currState = RUNNING, requestedState = ANY (active)
+ // ** for tasks that are just in INIT state, we do not add them here because old
+ // Participants, upon connection reset, set tasks' currentStates to INIT. We cannot consider
+ // those tasks active **
+ if (result.containsKey(instance) && (currState == TaskPartitionState.INIT
+ && requestedState != null && requestedState.equals(TaskPartitionState.RUNNING.name())
+ || currState == TaskPartitionState.RUNNING)) {
+ // Check if this is a dropping transition
+ if (requestedState != null && requestedState.equals(TaskPartitionState.DROPPED.name())) {
+ if (!tasksToDrop.containsKey(instance)) {
+ tasksToDrop.put(instance, new HashSet<Integer>());
+ }
+ tasksToDrop.get(instance).add(pId);
+ }
result.get(instance).add(pId);
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestNoDoubleAssign.java b/helix-core/src/test/java/org/apache/helix/integration/task/TestNoDoubleAssign.java
index 8fa3f94..88da7fb 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/task/TestNoDoubleAssign.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestNoDoubleAssign.java
@@ -45,7 +45,7 @@ import com.google.common.collect.ImmutableMap;
public class TestNoDoubleAssign extends TaskTestBase {
private static final int THREAD_COUNT = 20;
private static final long CONNECTION_DELAY = 100L;
- private static final long POLL_DELAY = 100L;
+ private static final long POLL_DELAY = 50L;
private static final String TASK_DURATION = "200";
private static final Random RANDOM = new Random();
@@ -102,8 +102,6 @@ public class TestNoDoubleAssign extends TaskTestBase {
pollForDoubleAssign();
_driver.pollForWorkflowState(workflowName, TaskState.COMPLETED);
- Assert.assertFalse(_existsDoubleAssign.get());
-
// Shut down thread pools
_executorServicePoll.shutdown();
_executorServiceConnection.shutdown();
@@ -118,6 +116,9 @@ public class TestNoDoubleAssign extends TaskTestBase {
_executorServicePoll.shutdownNow();
_executorServiceConnection.shutdownNow();
}
+ Thread.sleep(500L);
+
+ Assert.assertFalse(_existsDoubleAssign.get());
}
/**
[helix] 06/07: [HELIX-811] HELIX: Only log relayMsg if it doesn't
exist
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 7daecd030ca6555f4a7db645761122142c9188d1
Author: narendly <na...@gmail.com>
AuthorDate: Tue Feb 26 17:29:40 2019 -0800
[HELIX-811] HELIX: Only log relayMsg if it doesn't exist
This log was flooding our log files. We need to change it so that relay messages only get logged once.
---
.../org/apache/helix/common/caches/InstanceMessagesCache.java | 10 ++++++----
1 file changed, 6 insertions(+), 4 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java b/helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
index 21c35af..d936467 100644
--- a/helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
+++ b/helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
@@ -457,13 +457,15 @@ public class InstanceMessagesCache {
private void cacheRelayMessage(Message relayMessage, Message hostMessage) {
String instanceName = relayMessage.getTgtName();
if (!_relayMessageCache.containsKey(instanceName)) {
- _relayMessageCache.put(instanceName, Maps.<String, Message>newHashMap());
+ _relayMessageCache.put(instanceName, Maps.<String, Message> newHashMap());
+ }
+ if (!_relayMessageCache.get(instanceName).containsKey(relayMessage.getId())) {
+ // Only log if the message doesn't already exist in the cache
+ LOG.info("Add relay message to relay cache " + relayMessage.getMsgId() + ", hosted message "
+ + hostMessage.getMsgId());
}
_relayMessageCache.get(instanceName).put(relayMessage.getId(), relayMessage);
_relayHostMessageCache.put(relayMessage.getMsgId(), hostMessage);
-
- LOG.info("Add relay message to relay cache " + relayMessage.getMsgId() + ", hosted message "
- + hostMessage.getMsgId());
}
@Override public String toString() {
[helix] 02/07: [HELIX-807] REST: Add get maintenance signal endpoint
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit a827f8fabd0eb6971aebe771c2d21c9b53c97214
Author: narendly <na...@gmail.com>
AuthorDate: Tue Feb 26 17:20:46 2019 -0800
[HELIX-807] REST: Add get maintenance signal endpoint
Changelist:
1. Add get maintenance signal endpoint
2. Add a test
---
.../rest/server/resources/AbstractResource.java | 1 -
.../server/resources/helix/ClusterAccessor.java | 22 ++++++++++++++++++--
.../helix/rest/server/TestClusterAccessor.java | 24 +++++++++++++++++-----
3 files changed, 39 insertions(+), 8 deletions(-)
diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/AbstractResource.java b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/AbstractResource.java
index f5734a2..3b7d995 100644
--- a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/AbstractResource.java
+++ b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/AbstractResource.java
@@ -46,7 +46,6 @@ public class AbstractResource {
id,
disabled,
history,
- maintenanceHistory,
count,
error
}
diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java
index c81f467..4427d50 100644
--- a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java
+++ b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java
@@ -47,6 +47,7 @@ import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.ControllerHistory;
import org.apache.helix.model.HelixConfigScope;
import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.MaintenanceSignal;
import org.apache.helix.model.Message;
import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
@@ -68,7 +69,10 @@ public class ClusterAccessor extends AbstractHelixResource {
maintenance,
messages,
stateModelDefinitions,
- clusters
+ clusters,
+ maintenanceSignal,
+ maintenanceHistory,
+ clusterName
}
@GET
@@ -342,6 +346,20 @@ public class ClusterAccessor extends AbstractHelixResource {
}
@GET
+ @Path("{clusterId}/controller/maintenanceSignal")
+ public Response getClusterMaintenanceSignal(@PathParam("clusterId") String clusterId) {
+ HelixDataAccessor dataAccessor = getDataAccssor(clusterId);
+ MaintenanceSignal maintenanceSignal =
+ dataAccessor.getProperty(dataAccessor.keyBuilder().maintenance());
+ if (maintenanceSignal != null) {
+ Map<String, String> maintenanceInfo = maintenanceSignal.getRecord().getSimpleFields();
+ maintenanceInfo.put(ClusterProperties.clusterName.name(), clusterId);
+ return JSONRepresentation(maintenanceInfo);
+ }
+ return notFound(String.format("Cluster %s is not in maintenance mode!", clusterId));
+ }
+
+ @GET
@Path("{clusterId}/controller/messages")
public Response getClusterControllerMessages(@PathParam("clusterId") String clusterId) {
HelixDataAccessor dataAccessor = getDataAccssor(clusterId);
@@ -425,7 +443,7 @@ public class ClusterAccessor extends AbstractHelixResource {
historyRecord != null ? historyRecord.getHistoryList() : Collections.emptyList());
break;
case MAINTENANCE:
- history.put(Properties.maintenanceHistory.name(),
+ history.put(ClusterProperties.maintenanceHistory.name(),
historyRecord != null ? historyRecord.getMaintenanceHistoryList()
: Collections.emptyList());
break;
diff --git a/helix-rest/src/test/java/org/apache/helix/rest/server/TestClusterAccessor.java b/helix-rest/src/test/java/org/apache/helix/rest/server/TestClusterAccessor.java
index 5506509..3d7577a 100644
--- a/helix-rest/src/test/java/org/apache/helix/rest/server/TestClusterAccessor.java
+++ b/helix-rest/src/test/java/org/apache/helix/rest/server/TestClusterAccessor.java
@@ -261,16 +261,29 @@ public class TestClusterAccessor extends AbstractTestClass {
node.get(ClusterAccessor.ClusterProperties.maintenance.name()).getBooleanValue();
Assert.assertTrue(maintenance);
+ // Check that we could retrieve maintenance signal correctly
+ String signal = get("clusters/" + cluster + "/controller/maintenanceSignal",
+ Response.Status.OK.getStatusCode(), true);
+ Map<String, Object> maintenanceSignalMap =
+ OBJECT_MAPPER.readValue(signal, new TypeReference<HashMap<String, Object>>() {
+ });
+ Assert.assertEquals(maintenanceSignalMap.get("TRIGGERED_BY"), "USER");
+ Assert.assertEquals(maintenanceSignalMap.get("REASON"), reason);
+ Assert.assertNotNull(maintenanceSignalMap.get("TIMESTAMP"));
+ Assert.assertEquals(maintenanceSignalMap.get("clusterName"), cluster);
+
// disable maintenance mode
post("clusters/" + cluster, ImmutableMap.of("command", "disableMaintenanceMode"),
- Entity.entity("", MediaType.APPLICATION_JSON_TYPE),
- Response.Status.OK.getStatusCode());
+ Entity.entity("", MediaType.APPLICATION_JSON_TYPE), Response.Status.OK.getStatusCode());
// verify no longer in maintenance mode
body = get("clusters/" + cluster + "/maintenance", Response.Status.OK.getStatusCode(), true);
node = OBJECT_MAPPER.readTree(body);
Assert.assertFalse(
node.get(ClusterAccessor.ClusterProperties.maintenance.name()).getBooleanValue());
+
+ get("clusters/" + cluster + "/controller/maintenanceSignal",
+ Response.Status.NOT_FOUND.getStatusCode(), false);
}
@Test
@@ -317,12 +330,13 @@ public class TestClusterAccessor extends AbstractTestClass {
Entity.entity(reason, MediaType.APPLICATION_JSON_TYPE), Response.Status.OK.getStatusCode());
// Get the maintenance history JSON's last entry
- String maintenanceHistory =
- get("clusters/" + cluster + "/controller/maintenanceHistory", Response.Status.OK.getStatusCode(), true);
+ String maintenanceHistory = get("clusters/" + cluster + "/controller/maintenanceHistory",
+ Response.Status.OK.getStatusCode(), true);
Map<String, Object> maintenanceHistoryMap =
OBJECT_MAPPER.readValue(maintenanceHistory, new TypeReference<HashMap<String, Object>>() {
});
- Object maintenanceHistoryList = maintenanceHistoryMap.get(AbstractResource.Properties.maintenanceHistory.name());
+ Object maintenanceHistoryList =
+ maintenanceHistoryMap.get(ClusterAccessor.ClusterProperties.maintenanceHistory.name());
Assert.assertNotNull(maintenanceHistoryList);
List<?> list = (List<?>) maintenanceHistoryList;
Assert.assertTrue(list.size() > 0);
[helix] 04/07: [HELIX-809] TEST: Fix unstable
TestClusterInMaintenanceModeWhenReachingMaxPartition
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit ae586ac6f511ebed7bb01e0bec61f526b7d4b178
Author: narendly <na...@gmail.com>
AuthorDate: Tue Feb 26 17:26:00 2019 -0800
[HELIX-809] TEST: Fix unstable TestClusterInMaintenanceModeWhenReachingMaxPartition
The pause for this was too short so the test was occasionally failing. This RB fixes this.
---
.../TestClusterInMaintenanceModeWhenReachingMaxPartition.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestClusterInMaintenanceModeWhenReachingMaxPartition.java b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestClusterInMaintenanceModeWhenReachingMaxPartition.java
index 9257ae3..7cd0235 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestClusterInMaintenanceModeWhenReachingMaxPartition.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestClusterInMaintenanceModeWhenReachingMaxPartition.java
@@ -102,7 +102,7 @@ public class TestClusterInMaintenanceModeWhenReachingMaxPartition extends ZkTest
_replica, -1);
_testDBs.add(db);
}
- Thread.sleep(100);
+ Thread.sleep(100L);
Assert.assertTrue(_clusterVerifier.verifyByPolling());
MaintenanceSignal maintenanceSignal = _dataAccessor.getProperty(_dataAccessor.keyBuilder().maintenance());
@@ -112,7 +112,7 @@ public class TestClusterInMaintenanceModeWhenReachingMaxPartition extends ZkTest
_participants.get(i).syncStop();
}
- Thread.sleep(500);
+ Thread.sleep(1000L);
maintenanceSignal = _dataAccessor.getProperty(_dataAccessor.keyBuilder().maintenance());
Assert.assertNotNull(maintenanceSignal);
Assert.assertNotNull(maintenanceSignal.getReason());
[helix] 01/07: [HELIX-806] HELIX: Modify endpoints for
instrumenting maintenance with custom fields
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit db33ae5dbe5c50c0db62d3abc828530bc114b0ea
Author: narendly <na...@gmail.com>
AuthorDate: Tue Feb 26 17:19:14 2019 -0800
[HELIX-806] HELIX: Modify endpoints for instrumenting maintenance with custom fields
We want to use content for users to input their KV mappings as a JSON string.
Changelist:
1. Modify enable/disableMaintenanceMode endpoint logic
2. Modify tests
---
.../server/resources/helix/ClusterAccessor.java | 28 +++++++++++-----------
.../helix/rest/server/TestClusterAccessor.java | 15 ++++--------
2 files changed, 19 insertions(+), 24 deletions(-)
diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java
index 51dde95..c81f467 100644
--- a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java
+++ b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java
@@ -158,7 +158,7 @@ public class ClusterAccessor extends AbstractHelixResource {
@Path("{clusterId}")
public Response updateCluster(@PathParam("clusterId") String clusterId,
@QueryParam("command") String commandStr, @QueryParam("superCluster") String superCluster,
- @QueryParam("customFields") String customFields, String content) {
+ String content) {
Command command;
try {
command = getCommand(commandStr);
@@ -211,20 +211,21 @@ public class ClusterAccessor extends AbstractHelixResource {
case enableMaintenanceMode:
case disableMaintenanceMode:
+ // Try to parse the content string. If parseable, use it as a KV mapping. Otherwise, treat it
+ // as a REASON String
+ Map<String, String> customFieldsMap = null;
try {
- Map<String, String> customFieldsMap = new HashMap<>();
- if (customFields != null && !customFields.isEmpty()) {
- customFieldsMap = OBJECT_MAPPER.readValue(customFields,
- new TypeReference<HashMap<String, String>>() {
- });
- }
- helixAdmin.manuallyEnableMaintenanceMode(clusterId,
- command == Command.enableMaintenanceMode, content, customFieldsMap);
- } catch (Exception ex) {
- _logger.error("Failed to disable maintenance mode for cluster {}. Exception: {}", clusterId,
- ex);
- return serverError(ex);
+ // Try to parse content
+ customFieldsMap =
+ OBJECT_MAPPER.readValue(content, new TypeReference<HashMap<String, String>>() {
+ });
+ // content is given as a KV mapping. Nullify content
+ content = null;
+ } catch (Exception e) {
+ // NOP
}
+ helixAdmin.manuallyEnableMaintenanceMode(clusterId, command == Command.enableMaintenanceMode,
+ content, customFieldsMap);
break;
default:
@@ -234,7 +235,6 @@ public class ClusterAccessor extends AbstractHelixResource {
return OK();
}
-
@GET
@Path("{clusterId}/configs")
public Response getClusterConfig(@PathParam("clusterId") String clusterId) {
diff --git a/helix-rest/src/test/java/org/apache/helix/rest/server/TestClusterAccessor.java b/helix-rest/src/test/java/org/apache/helix/rest/server/TestClusterAccessor.java
index 604c682..5506509 100644
--- a/helix-rest/src/test/java/org/apache/helix/rest/server/TestClusterAccessor.java
+++ b/helix-rest/src/test/java/org/apache/helix/rest/server/TestClusterAccessor.java
@@ -20,7 +20,6 @@ package org.apache.helix.rest.server;
*/
import java.io.IOException;
-import java.net.URLEncoder;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
@@ -337,20 +336,16 @@ public class TestClusterAccessor extends AbstractTestClass {
public void testEnableDisableMaintenanceModeWithCustomFields() {
System.out.println("Start test :" + TestHelper.getTestMethodName());
String cluster = _clusters.iterator().next();
- String reason = "Test reason";
HelixDataAccessor accessor = new ZKHelixDataAccessor(cluster, _baseAccessor);
- String customFields = "{\"key1\":\"value1\",\"key2\":\"value2\"}";
- // Note that URLEncoder.encode has to be used due to a Jersey bug
- // See https://github.com/Mercateo/rest-schemagen/issues/51
- post("clusters/" + cluster,
- ImmutableMap.of("command", "enableMaintenanceMode", "customFields",
- URLEncoder.encode(customFields)),
- Entity.entity(reason, MediaType.APPLICATION_JSON_TYPE), Response.Status.OK.getStatusCode());
+ String content = "{\"key1\":\"value1\",\"key2\":\"value2\"}";
+ post("clusters/" + cluster, ImmutableMap.of("command", "enableMaintenanceMode"),
+ Entity.entity(content, MediaType.APPLICATION_JSON_TYPE),
+ Response.Status.OK.getStatusCode());
MaintenanceSignal signal = accessor.getProperty(accessor.keyBuilder().maintenance());
Assert.assertNotNull(signal);
- Assert.assertEquals(reason, signal.getReason());
+ Assert.assertNull(signal.getReason());
Assert.assertEquals(signal.getTriggeringEntity(), MaintenanceSignal.TriggeringEntity.USER);
Map<String, String> simpleFields = signal.getRecord().getSimpleFields();
Assert.assertEquals(simpleFields.get("key1"), "value1");
[helix] 07/07: [HELIX-812] HELIX: Fix maintenance history bug
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 4ecd923f279a53554fc66e9a71356af2f125c451
Author: narendly <na...@gmail.com>
AuthorDate: Tue Feb 26 17:30:47 2019 -0800
[HELIX-812] HELIX: Fix maintenance history bug
There was a bug in maintenance history where when the cluster exits maintenance mode automatically, it would record the exit action twice in history. This is because each pipeline is designed to run MaintenanceRecoveryStage twice.
Changelist:
1. Add a flag so that if maintenanceSignal has been changed, just return from MaintenanceRecoveryStage
---
.../controller/dataproviders/BaseControllerDataProvider.java | 12 ++++++++++++
.../helix/controller/stages/MaintenanceRecoveryStage.java | 8 ++++++++
.../integration/controller/TestClusterMaintenanceMode.java | 1 -
3 files changed, 20 insertions(+), 1 deletion(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java b/helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
index 8d2a4a5..da91f97 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
@@ -78,6 +78,7 @@ public class BaseControllerDataProvider implements ControlContextProvider {
private boolean _updateInstanceOfflineTime = true;
private MaintenanceSignal _maintenanceSignal;
private boolean _isMaintenanceModeEnabled;
+ private boolean _hasMaintenanceSignalChanged;
private ExecutorService _asyncTasksThreadPool;
// A map recording what data has changed
@@ -271,6 +272,9 @@ public class BaseControllerDataProvider implements ControlContextProvider {
private void updateMaintenanceInfo(final HelixDataAccessor accessor) {
_maintenanceSignal = accessor.getProperty(accessor.keyBuilder().maintenance());
_isMaintenanceModeEnabled = _maintenanceSignal != null;
+ // The following flag is to guarantee that there's only one update per pineline run because we
+ // check for whether maintenance recovery could happen twice every pipeline
+ _hasMaintenanceSignalChanged = false;
}
private void updateIdealRuleMap() {
@@ -690,6 +694,14 @@ public class BaseControllerDataProvider implements ControlContextProvider {
return _isMaintenanceModeEnabled;
}
+ public boolean hasMaintenanceSignalChanged() {
+ return _hasMaintenanceSignalChanged;
+ }
+
+ public void setMaintenanceSignalChanged() {
+ _hasMaintenanceSignalChanged = true;
+ }
+
public MaintenanceSignal getMaintenanceSignal() {
return _maintenanceSignal;
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MaintenanceRecoveryStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MaintenanceRecoveryStage.java
index e366ffa..d64459a 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/MaintenanceRecoveryStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MaintenanceRecoveryStage.java
@@ -52,6 +52,13 @@ public class MaintenanceRecoveryStage extends AbstractAsyncBaseStage {
return;
}
+ // Check if the maintenance signal has been changed during this pipeline run
+ // If true, skip this stage because the Controller already changed the signal
+ // The flag will be flipped in the next ReadClusterDataStage()
+ if (cache.hasMaintenanceSignalChanged()) {
+ return;
+ }
+
// Check for the maintenance signal
// If it was entered manually or the signal is null (which shouldn't happen), skip this stage
MaintenanceSignal maintenanceSignal = cache.getMaintenanceSignal();
@@ -108,6 +115,7 @@ public class MaintenanceRecoveryStage extends AbstractAsyncBaseStage {
// MaintenanceSignal. AutoTriggerReason won't be recorded
manager.getClusterManagmentTool().autoEnableMaintenanceMode(manager.getClusterName(), false,
reason, internalReason);
+ cache.setMaintenanceSignalChanged(); // Set the flag so we do not double enable/disable
LogUtil.logInfo(LOG, _eventId, reason);
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterMaintenanceMode.java b/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterMaintenanceMode.java
index 806af2f..2e4f440 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterMaintenanceMode.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterMaintenanceMode.java
@@ -343,7 +343,6 @@ public class TestClusterMaintenanceMode extends TaskTestBase {
ClusterConfig clusterConfig = _manager.getConfigAccessor().getClusterConfig(CLUSTER_NAME);
clusterConfig.setMaxPartitionsPerInstance(-1);
_manager.getConfigAccessor().setClusterConfig(CLUSTER_NAME, clusterConfig);
- System.out.println("Set clusterconfig");
Thread.sleep(500L);
[helix] 05/07: [HELIX-810] HELIX: Fix NPE in InstanceMessagesCache
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit c1f9af5b4ffb88a4893ac04890690946e97082e4
Author: narendly <na...@gmail.com>
AuthorDate: Tue Feb 26 17:28:38 2019 -0800
[HELIX-810] HELIX: Fix NPE in InstanceMessagesCache
It was observed that InstanceMessagesCache was throwing an NPE when it tries to setRelayTime(). This is likely because some relay messages have target instances that are no longer live (thus not in liveInstanceMap). InstanceMessagesCache must handle this gracefully by skipping the operation. We do not delete these msgs right away because the instance may come back alive. Otherwise, after some time has passed, the msg will get expired by the Controller and be removed.
Changelist;
1. Add a try-catch block
2. Improve logging
---
.../helix/common/caches/InstanceMessagesCache.java | 133 +++++++++++----------
1 file changed, 72 insertions(+), 61 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java b/helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
index b57ffb9..21c35af 100644
--- a/helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
+++ b/helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
@@ -326,84 +326,95 @@ public class InstanceMessagesCache {
private void setRelayTime(Message relayMessage, Map<String, LiveInstance> liveInstanceMap,
Map<String, Map<String, Map<String, CurrentState>>> currentStateMap) {
-
// relay time already set, avoid to reset it to a later time.
if (relayMessage.getRelayTime() > relayMessage.getCreateTimeStamp()) {
return;
}
- Message hostedMessage = _relayHostMessageCache.get(relayMessage.getMsgId());
- String sessionId = hostedMessage.getTgtSessionId();
- String instance = hostedMessage.getTgtName();
- String resourceName = hostedMessage.getResourceName();
- String instanceSessionId = liveInstanceMap.get(instance).getSessionId();
-
- long currentTime = System.currentTimeMillis();
- long expiredTime = currentTime + relayMessage.getExpiryPeriod();
-
- if (!instanceSessionId.equals(sessionId)) {
- LOG.debug(
- "Hosted Instance SessionId {} does not match sessionId {} in hosted message , set relay message {} to be expired at {}, hosted message ",
- instanceSessionId, sessionId, relayMessage.getId(), expiredTime,
- hostedMessage.getMsgId());
- relayMessage.setRelayTime(currentTime);
- return;
- }
-
- Map<String, Map<String, CurrentState>> instanceCurrentStateMap = currentStateMap.get(instance);
- if (instanceCurrentStateMap == null) {
- LOG.debug(
- "No instanceCurrentStateMap found for {} on {}, set relay messages {} to be expired at {}"
- + resourceName, instance, relayMessage.getId(), expiredTime);
- relayMessage.setRelayTime(currentTime);
- return;
- }
+ // Relay time has not been set. Proceed to set the relay time
+ try {
+ long currentTime = System.currentTimeMillis();
+ long expiredTime = currentTime + relayMessage.getExpiryPeriod();
+
+ Message hostedMessage = _relayHostMessageCache.get(relayMessage.getMsgId());
+ String sessionId = hostedMessage.getTgtSessionId();
+ String instance = hostedMessage.getTgtName();
+ String resourceName = hostedMessage.getResourceName();
+ if (!liveInstanceMap.containsKey(instance)) {
+ // It's possible that hostedMsg's target is no longer live. In this case, we just set the
+ // relay time and return so that the message could be cleaned up after a short delay
+ relayMessage.setRelayTime(currentTime);
+ return;
+ }
+ String instanceSessionId = liveInstanceMap.get(instance).getSessionId();
+
+ if (!instanceSessionId.equals(sessionId)) {
+ LOG.debug(
+ "Hosted Instance SessionId {} does not match sessionId {} in hosted message , set relay message {} to be expired at {}, hosted message ",
+ instanceSessionId, sessionId, relayMessage.getId(), expiredTime,
+ hostedMessage.getMsgId());
+ relayMessage.setRelayTime(currentTime);
+ return;
+ }
- Map<String, CurrentState> sessionCurrentStateMap = instanceCurrentStateMap.get(sessionId);
- if (sessionCurrentStateMap == null) {
- LOG.debug("No sessionCurrentStateMap found, set relay messages {} to be expired at {}. ",
- relayMessage.getId(), expiredTime);
- relayMessage.setRelayTime(currentTime);
- return;
- }
+ Map<String, Map<String, CurrentState>> instanceCurrentStateMap =
+ currentStateMap.get(instance);
+ if (instanceCurrentStateMap == null) {
+ LOG.debug(
+ "No instanceCurrentStateMap found for {} on {}, set relay messages {} to be expired at {}"
+ + resourceName,
+ instance, relayMessage.getId(), expiredTime);
+ relayMessage.setRelayTime(currentTime);
+ return;
+ }
- String partitionName = hostedMessage.getPartitionName();
- String targetState = hostedMessage.getToState();
- String fromState = hostedMessage.getFromState();
+ Map<String, CurrentState> sessionCurrentStateMap = instanceCurrentStateMap.get(sessionId);
+ if (sessionCurrentStateMap == null) {
+ LOG.debug("No sessionCurrentStateMap found, set relay messages {} to be expired at {}. ",
+ relayMessage.getId(), expiredTime);
+ relayMessage.setRelayTime(currentTime);
+ return;
+ }
- CurrentState currentState = sessionCurrentStateMap.get(resourceName);
- if (currentState == null) {
- LOG.debug("No currentState found for {} on {}, set relay message {} to be expired at {} ",
- resourceName, instance, relayMessage.getId(),
- (currentTime + relayMessage.getExpiryPeriod()));
- relayMessage.setRelayTime(currentTime);
- return;
- }
+ String partitionName = hostedMessage.getPartitionName();
+ String targetState = hostedMessage.getToState();
+ String fromState = hostedMessage.getFromState();
+
+ CurrentState currentState = sessionCurrentStateMap.get(resourceName);
+ if (currentState == null) {
+ LOG.debug("No currentState found for {} on {}, set relay message {} to be expired at {} ",
+ resourceName, instance, relayMessage.getId(),
+ (currentTime + relayMessage.getExpiryPeriod()));
+ relayMessage.setRelayTime(currentTime);
+ return;
+ }
- if (targetState.equals(currentState.getState(partitionName))) {
- long completeTime = currentState.getEndTime(partitionName);
- if (completeTime < relayMessage.getCreateTimeStamp()) {
- completeTime = currentTime;
+ if (targetState.equals(currentState.getState(partitionName))) {
+ long completeTime = currentState.getEndTime(partitionName);
+ if (completeTime < relayMessage.getCreateTimeStamp()) {
+ completeTime = currentTime;
+ }
+ relayMessage.setRelayTime(completeTime);
+ LOG.debug(
+ "Target state match the hosted message's target state, set relay message {} relay time at {}.",
+ relayMessage.getId(), completeTime);
}
- relayMessage.setRelayTime(completeTime);
- LOG.debug(
- "Target state match the hosted message's target state, set relay message {} relay time at {}.",
- relayMessage.getId(), completeTime);
- }
- if (!fromState.equals(currentState.getState(partitionName))) {
- LOG.debug(
- "Current state does not match hosted message's from state, set relay message {} relay time at {}.",
- relayMessage.getId(), currentTime);
- relayMessage.setRelayTime(currentTime);
+ if (!fromState.equals(currentState.getState(partitionName))) {
+ LOG.debug(
+ "Current state does not match hosted message's from state, set relay message {} relay time at {}.",
+ relayMessage.getId(), currentTime);
+ relayMessage.setRelayTime(currentTime);
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to set the relay time. RelayMsgId: {} Exception: {}", relayMessage.getId(),
+ e);
}
}
/**
* Provides a list of current outstanding pending state transition messages on a given instance.
- *
* @param instanceName
- *
* @return
*/
public Map<String, Message> getMessages(String instanceName) {