You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by lx...@apache.org on 2018/03/25 04:19:43 UTC
helix git commit: [HELIX-682] controller should delete obsolete
messages with timeout to unblock state transition
Repository: helix
Updated Branches:
refs/heads/master 1c78fef91 -> 4d652eb9a
[HELIX-682] controller should delete obsolete messages with timeout to unblock state transition
Project: http://git-wip-us.apache.org/repos/asf/helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/helix/commit/4d652eb9
Tree: http://git-wip-us.apache.org/repos/asf/helix/tree/4d652eb9
Diff: http://git-wip-us.apache.org/repos/asf/helix/diff/4d652eb9
Branch: refs/heads/master
Commit: 4d652eb9aabaeaacbcfd2df6eba70b5f8c442094
Parents: 1c78fef
Author: Harry Zhang <zh...@usc.edu>
Authored: Tue Mar 20 18:47:02 2018 -0700
Committer: Harry Zhang <zh...@usc.edu>
Committed: Sat Mar 24 19:20:28 2018 -0700
----------------------------------------------------------------------
.../stages/CurrentStateComputationStage.java | 6 +-
.../controller/stages/CurrentStateOutput.java | 32 +++++++-
.../stages/MessageGenerationPhase.java | 81 ++++++++++++++++++-
.../stages/TestRebalancePipeline.java | 82 +++++++++++++++-----
4 files changed, 177 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/helix/blob/4d652eb9/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
index f8f5a2f..edf2c47 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
@@ -23,7 +23,6 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-
import org.apache.helix.controller.pipeline.AbstractBaseStage;
import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.model.CurrentState;
@@ -158,7 +157,10 @@ public class CurrentStateComputationStage extends AbstractBaseStage {
currentState.getState(partitionName));
currentStateOutput.setRequestedState(resourceName, partition, instanceName,
currentState.getRequestedState(partitionName));
- currentStateOutput.setInfo(resourceName, partition, instanceName, currentState.getInfo(partitionName));
+ currentStateOutput
+ .setInfo(resourceName, partition, instanceName, currentState.getInfo(partitionName));
+ currentStateOutput.setEndTime(resourceName, partition, instanceName,
+ currentState.getEndTime(partitionName));
}
}
}
http://git-wip-us.apache.org/repos/asf/helix/blob/4d652eb9/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java b/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java
index 97420e5..ce0880e 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java
@@ -23,7 +23,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
-
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.Message;
import org.apache.helix.model.Partition;
@@ -40,6 +39,13 @@ public class CurrentStateOutput {
private final Map<String, Map<Partition, Map<String, String>>> _currentStateMap;
private final Map<String, Map<Partition, Map<String, Message>>> _pendingStateMap;
private final Map<String, Map<Partition, Map<String, Message>>> _cancellationStateMap;
+
+ // resourceName -> (Partition -> (instanceName -> endTime))
+ // Note that startTime / endTime in CurrentState marks that of state transition
+ // and therefore endTime is the starting timestamp of the partition being in the
+ // current state
+ private final Map<String, Map<Partition, Map<String, Long>>> _currentStateEndTimeMap;
+
// Contains per-resource maps of partition -> (instance, requested_state). This corresponds to the
// REQUESTED_STATE
// field in the CURRENTSTATES node.
@@ -57,6 +63,7 @@ public class CurrentStateOutput {
_currentStateMap = new HashMap<>();
_pendingStateMap = new HashMap<>();
_cancellationStateMap = new HashMap<>();
+ _currentStateEndTimeMap = new HashMap<>();
_resourceStateModelMap = new HashMap<>();
_curStateMetaMap = new HashMap<>();
_requestedStateMap = new HashMap<>();
@@ -101,6 +108,17 @@ public class CurrentStateOutput {
_currentStateMap.get(resourceName).get(partition).put(instanceName, state);
}
+ public void setEndTime(String resourceName, Partition partition,
+ String instanceName, Long timestamp) {
+ if (!_currentStateEndTimeMap.containsKey(resourceName)) {
+ _currentStateEndTimeMap.put(resourceName, new HashMap<Partition, Map<String, Long>>());
+ }
+ if (!_currentStateEndTimeMap.get(resourceName).containsKey(partition)) {
+ _currentStateEndTimeMap.get(resourceName).put(partition, new HashMap<String, Long>());
+ }
+ _currentStateEndTimeMap.get(resourceName).get(partition).put(instanceName, timestamp);
+ }
+
public void setRequestedState(String resourceName, Partition partition, String instanceName,
String state) {
if (!_requestedStateMap.containsKey(resourceName)) {
@@ -168,6 +186,18 @@ public class CurrentStateOutput {
return null;
}
+ public Long getEndTime(String resourceName, Partition partition,
+ String instanceName) {
+ Map<Partition, Map<String, Long>> partitionInfo = _currentStateEndTimeMap.get(resourceName);
+ if (partitionInfo != null) {
+ Map<String, Long> instanceInfo = partitionInfo.get(partition);
+ if (instanceInfo != null && instanceInfo.get(instanceName) != null) {
+ return instanceInfo.get(instanceName);
+ }
+ }
+ return -1L;
+ }
+
public String getRequestedState(String resourceName, Partition partition, String instanceName) {
Map<Partition, Map<String, String>> map = _requestedStateMap.get(resourceName);
if (map != null) {
http://git-wip-us.apache.org/repos/asf/helix/blob/4d652eb9/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
index d60c20e..a41c7b8 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
@@ -24,7 +24,9 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
-
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.api.config.StateTransitionTimeoutConfig;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
@@ -48,6 +50,13 @@ import org.slf4j.LoggerFactory;
*/
public class MessageGenerationPhase extends AbstractBaseStage {
private final static String NO_DESIRED_STATE = "NoDesiredState";
+
+ // If we see there is any invalid pending message leaving on host, i.e. message
+ // tells participant to change from SLAVE to MASTER, and the participant is already
+ // at MASTER state, we wait for 3 sec and if the message is still not cleaned up by
+ // participant, controller will cleanup them proactively to unblock further state
+ // transition
+ private final static long DEFAULT_OBSELETE_MSG_PURGE_DELAY = 3 * 1000;
private static Logger logger = LoggerFactory.getLogger(MessageGenerationPhase.class);
@Override
@@ -55,6 +64,7 @@ public class MessageGenerationPhase extends AbstractBaseStage {
HelixManager manager = event.getAttribute(AttributeName.helixmanager.name());
ClusterDataCache cache = event.getAttribute(AttributeName.ClusterDataCache.name());
Map<String, Resource> resourceMap = event.getAttribute(AttributeName.RESOURCES_TO_REBALANCE.name());
+ Map<String, List<Message>> pendingMessagesToCleanUp = new HashMap<>();
CurrentStateOutput currentStateOutput =
event.getAttribute(AttributeName.CURRENT_STATE.name());
IntermediateStateOutput intermediateStateOutput =
@@ -121,6 +131,18 @@ public class MessageGenerationPhase extends AbstractBaseStage {
Message message = null;
+ if (shouldCleanUpPendingMessage(pendingMessage, currentState,
+ currentStateOutput.getEndTime(resourceName, partition, instanceName))) {
+ logger.info(
+ "Adding pending message {} on instance {} to clean up. Msg: {}->{}, current state of resource {}:{} is {}",
+ pendingMessage.getMsgId(), instanceName, pendingMessage.getFromState(),
+ pendingMessage.getToState(), resourceName, partition, currentState);
+ if (!pendingMessagesToCleanUp.containsKey(instanceName)) {
+ pendingMessagesToCleanUp.put(instanceName, new ArrayList<Message>());
+ }
+ pendingMessagesToCleanUp.get(instanceName).add(pendingMessage);
+ }
+
if (desiredState.equals(NO_DESIRED_STATE) || desiredState.equalsIgnoreCase(currentState)) {
if (desiredState.equals(NO_DESIRED_STATE) || pendingMessage != null && !currentState
.equalsIgnoreCase(pendingMessage.getToState())) {
@@ -207,9 +229,66 @@ public class MessageGenerationPhase extends AbstractBaseStage {
} // end of for-each-partition
}
+
+ // Asynchronously clean up pending messages if necessary
+ if (!pendingMessagesToCleanUp.isEmpty()) {
+ schedulePendingMessageCleanUp(pendingMessagesToCleanUp, cache.getAsyncTasksThreadPool(),
+ manager.getHelixDataAccessor());
+ }
event.addAttribute(AttributeName.MESSAGES_ALL.name(), output);
}
+ /**
+ * Start a job in worker pool that asynchronously clean up pending message. Since it is possible
+ * that participant failed to clean up message after processing, it is important for controller
+ * to try to clean them up as well to unblock further rebalance
+ *
+ * @param pendingMessagesToPurge key: instance name, value: list of pending message to cleanup
+ * @param workerPool ExecutorService that job can be submitted to
+ * @param accessor Data accessor used to clean up message
+ */
+ private void schedulePendingMessageCleanUp(
+ final Map<String, List<Message>> pendingMessagesToPurge, ExecutorService workerPool,
+ final HelixDataAccessor accessor) {
+ workerPool.submit(new Callable<Object>() {
+ @Override
+ public Object call() {
+ for (Map.Entry<String, List<Message>> entry : pendingMessagesToPurge.entrySet()) {
+ String instanceName = entry.getKey();
+ for (Message msg : entry.getValue()) {
+ if (accessor.removeProperty(msg.getKey(accessor.keyBuilder(), instanceName))) {
+ logger.info("Deleted message {} from instance {}", msg.getMsgId(), instanceName);
+ } else {
+ logger.warn(
+ "Failed to delete message {} from instance {}. Will retry next time if the message is still there",
+ msg.getMsgId(), instanceName);
+ }
+ }
+ }
+ return null;
+ }
+ });
+ }
+
+ private boolean shouldCleanUpPendingMessage(Message pendingMsg, String currentState,
+ Long currentStateTransitionEndTime) {
+ if (pendingMsg == null) {
+ return false;
+ }
+ if (currentState.equalsIgnoreCase(pendingMsg.getToState())) {
+ // If pending message's toState is same as current state, state transition is finished
+ // successfully. In this case, we will wait for a timeout for participant to cleanup
+ // processed message. If participant fail to do so, controller is going to proactively delete
+ // the message as participant does not retry message deletion upon failure.
+ return System.currentTimeMillis() - currentStateTransitionEndTime
+ > DEFAULT_OBSELETE_MSG_PURGE_DELAY;
+ } else {
+ // Partition's current state should be either pending message's fromState or toState or
+ // the message is invalid and can be safely deleted immediately.
+ return !currentState.equalsIgnoreCase(pendingMsg.getFromState());
+ }
+ }
+
private Message createStateTransitionMessage(HelixManager manager, Resource resource, String partitionName,
String instanceName, String currentState, String nextState, String sessionId,
String stateModelDefName) {
http://git-wip-us.apache.org/repos/asf/helix/blob/4d652eb9/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
index 3b67d15..7c4bba6 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
@@ -19,9 +19,10 @@ package org.apache.helix.controller.stages;
* under the License.
*/
+import java.util.ArrayList;
import java.util.Date;
import java.util.List;
-
+import java.util.UUID;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
@@ -38,7 +39,6 @@ import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.Message;
-import org.apache.helix.model.Message.Attributes;
import org.apache.helix.model.Partition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -167,33 +167,67 @@ public class TestRebalancePipeline extends ZkUnitTestBase {
Assert.assertEquals(messages.size(), 1);
// round2: node0 and node1 update current states but not removing messages
- // controller's rebalance pipeline should be triggered but since messages are not
- // removed
- // no new messages will be sent
+ // Since controller's rebalancer pipeline will GC pending messages after timeout, and both hosts
+ // update current states to SLAVE, controller will send out rebalance message to
+ // have one host to become master
setCurrentState(clusterName, "localhost_0", resourceName, resourceName + "_0", "session_0",
- "SLAVE");
+ "SLAVE", true);
setCurrentState(clusterName, "localhost_1", resourceName, resourceName + "_0", "session_1",
+ "SLAVE", true);
+
+ // Controller has 3s timeout, so after 1s, controller should not have GCed message
+ Thread.sleep(1000);
+ Assert.assertEquals(accessor.getChildValues(keyBuilder.messages("localhost_0")).size(), 1);
+ Assert.assertEquals(accessor.getChildValues(keyBuilder.messages("localhost_1")).size(), 1);
+
+ // After another 2 second, controller should cleanup messages and continue to rebalance
+ Thread.sleep(3000);
+ // Manually trigger another rebalance by touching current state
+ setCurrentState(clusterName, "localhost_0", resourceName, resourceName + "_0", "session_0",
"SLAVE");
Thread.sleep(1000);
- messages = accessor.getChildNames(keyBuilder.messages("localhost_0"));
- Assert.assertEquals(messages.size(), 1);
- messages = accessor.getChildNames(keyBuilder.messages("localhost_1"));
- Assert.assertEquals(messages.size(), 1);
+ List<Message> host0Msg = accessor.getChildValues(keyBuilder.messages("localhost_0"));
+ List<Message> host1Msg = accessor.getChildValues(keyBuilder.messages("localhost_1"));
+ List<Message> allMsgs = new ArrayList<>(host0Msg);
+ allMsgs.addAll(host1Msg);
+ Assert.assertEquals(allMsgs.size(), 1);
+ Assert.assertEquals(allMsgs.get(0).getToState(), "MASTER");
+ Assert.assertEquals(allMsgs.get(0).getFromState(), "SLAVE");
- // round3: node0 removes message and controller's rebalance pipeline should be
- // triggered
- // and sends S->M to node0
- messages = accessor.getChildNames(keyBuilder.messages("localhost_0"));
- accessor.removeProperty(keyBuilder.message("localhost_0", messages.get(0)));
+ // round3: node0 changes state to master, but failed to delete message,
+ // controller will clean it up
+ setCurrentState(clusterName, "localhost_0", resourceName, resourceName + "_0", "session_0",
+ "MASTER", true);
+ Thread.sleep(3500);
+ // touch current state to trigger rebalance
+ setCurrentState(clusterName, "localhost_0", resourceName, resourceName + "_0", "session_0",
+ "MASTER", false);
+ Thread.sleep(1000);
+ Assert.assertTrue(accessor.getChildNames(keyBuilder.messages("localhost_0")).isEmpty());
+
+
+ // round4: node0 has duplicated but valid message, i.e. there is a P2P message sent to it
+ // due to error in the triggered pipeline, controller should remove duplicated message
+ // immediately as the partition has became master 3 sec ago (there is already a timeout)
+ Message sourceMsg = allMsgs.get(0);
+ Message dupMsg = new Message(sourceMsg.getMsgType(), UUID.randomUUID().toString());
+ dupMsg.getRecord().setSimpleFields(sourceMsg.getRecord().getSimpleFields());
+ dupMsg.getRecord().setListFields(sourceMsg.getRecord().getListFields());
+ dupMsg.getRecord().setMapFields(sourceMsg.getRecord().getMapFields());
+ accessor.setProperty(dupMsg.getKey(accessor.keyBuilder(), dupMsg.getTgtName()), dupMsg);
Thread.sleep(1000);
messages = accessor.getChildNames(keyBuilder.messages("localhost_0"));
- Assert.assertEquals(messages.size(), 1);
- ZNRecord msg =
- accessor.getProperty(keyBuilder.message("localhost_0", messages.get(0))).getRecord();
- String toState = msg.getSimpleField(Attributes.TO_STATE.toString());
- Assert.assertEquals(toState, "MASTER");
+ Assert.assertTrue(messages.isEmpty());
+
+ // round5: node0 has completely invalid message, controller should immediately delete it
+ dupMsg.setFromState("SLAVE");
+ dupMsg.setToState("OFFLINE");
+ accessor.setProperty(dupMsg.getKey(accessor.keyBuilder(), dupMsg.getTgtName()), dupMsg);
+ Thread.sleep(1000);
+ messages = accessor.getChildNames(keyBuilder.messages("localhost_0"));
+ Assert.assertTrue(messages.isEmpty());
System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis()));
@@ -440,6 +474,11 @@ public class TestRebalancePipeline extends ZkUnitTestBase {
protected void setCurrentState(String clusterName, String instance, String resourceGroupName,
String resourceKey, String sessionId, String state) {
+ setCurrentState(clusterName, instance, resourceGroupName, resourceKey, sessionId, state, false);
+ }
+
+ private void setCurrentState(String clusterName, String instance, String resourceGroupName,
+ String resourceKey, String sessionId, String state, boolean updateTimestamp) {
ZKHelixDataAccessor accessor =
new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
Builder keyBuilder = accessor.keyBuilder();
@@ -448,6 +487,9 @@ public class TestRebalancePipeline extends ZkUnitTestBase {
curState.setState(resourceKey, state);
curState.setSessionId(sessionId);
curState.setStateModelDefRef("MasterSlave");
+ if (updateTimestamp) {
+ curState.setEndTime(resourceKey, System.currentTimeMillis());
+ }
accessor.setProperty(keyBuilder.currentState(instance, sessionId, resourceGroupName), curState);
}