Skip to content

Per Replica Throttle -- 4th: add retraced map; downward-only handling… #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kaisun2000 wants to merge 1 commit into
base: throttle_p3
Choose a base branch
from
Open

Per Replica Throttle -- 4th: add retraced map; downward-only handling… #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 30 additions & 1 deletion helix-core/src/main/java/org/apache/helix/controller/stages/PerReplicaThrottleStage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,17 +271,46 @@ private void throttlePerReplicaMessages(IdealState idealState,
if (throttledRecoveryMessages.contains(message)) {
continue;
}
if (throttledLoadMessages.contains(message)) {
continue;
}
finalPartitionMessages.add(message);
}
outMessagesByPartition.put(partition, finalPartitionMessages);
output.addMessages(resourceName, partition, finalPartitionMessages);
}

// Step 6: constructs all retraced partition state map for the resource
// TODO: next PR
constructRetracedPartitionStateMap(resource, retracedPartitionsStateMap, outMessagesByPartition);

// Step 7: emit metrics
// TODO: next PR
}

private void constructRetracedPartitionStateMap(Resource resource,
Map<Partition, Map<String, String>> retracedPartitionsStateMap,
Map<Partition, List<Message>> outMessagesByPartition) {
for (Partition partition : resource.getPartitions()) {
List<Message> partitionMessages = outMessagesByPartition.get(partition);
if (partitionMessages == null) {
continue;
}
for (Message message : partitionMessages) {
if (!Message.MessageType.STATE_TRANSITION.name().equals(message.getMsgType())) {
// todo: log?
// ignore cancellation message etc.
continue;
}
String toState = message.getToState();
// toIntance may not be in the retracedStateMap as so far it is current state based.
// new instance in best possible not in currentstate would not be in retracedStateMap yet.
String toInstance = message.getTgtName();
Map<String, String> retracedStateMap = retracedPartitionsStateMap.get(partition);
retracedStateMap.put(toInstance, toState);
}
}
}

private void propagateCountsTopDown(
StateModelDefinition stateModelDef,
Map<String, Integer> expectedStateCountMap
Expand Down
212 changes: 211 additions & 1 deletion helix-core/src/test/java/org/apache/helix/controller/stages/TestPerReplicaThrottleStage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -350,7 +350,7 @@ public void testBootupRecoveryAndLoad() {
Assert.assertTrue(msgs.size() == 1);
}

// case 4. N1(S) Pending M, N2(S), N3(O), message N3(O->S) is treated as load
// case pending message: N1(S) Pending M, N2(S), N3(O), message N3(O->S) is treated as load
@Test
public void testUnnecessaryForMinReplicasAsLoadWithPendingMsg() {
String resourcePrefix = "resource";
Expand Down Expand Up @@ -410,6 +410,216 @@ public void testUnnecessaryForMinReplicasAsLoadWithPendingMsg() {
Assert.assertTrue(msg.getId().equals("001"));
}

// Case 4, Downward only throttling due to error partitions
@Test
public void testDownwardOnlyThrottle() {
String resourcePrefix = "resource";
int nResource = 1;
int nPartition = 3;
int nReplica = 3;
String[] resources = new String[nResource];
for (int i = 0; i < nResource; i++) {
resources[i] = resourcePrefix + "-" + i;
}
preSetup(resources, nPartition, nReplica);
setupThrottleConfig(0, 2);

event.addAttribute(AttributeName.RESOURCES_TO_REBALANCE.name(),
getResourceMap(resources, nPartition, "MasterSlave"));

// setup current state; setup message output; setup best possible
CurrentStateOutput currentStateOutput = new CurrentStateOutput();
MessageOutput messageOutput = new MessageOutput();
BestPossibleStateOutput bestPossibleStateOutput = new BestPossibleStateOutput();
for (String resource : resources) {
for (int p = 0; p < nPartition; p++) {
Partition partition = new Partition(resource + "_" + p);
if (p == 0) {
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 0, "SLAVE");
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 1, "MASTER");
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 2, "OFFLINE");
// this upward message should be throttled.
Message msgUp = new Message(Message.MessageType.STATE_TRANSITION, "001");
msgUp.setToState("SLAVE");
msgUp.setFromState("OFFLINE");
msgUp.setTgtName(HOSTNAME_PREFIX + 2);
// this downward message should not be throttled.
Message msgDown = new Message(Message.MessageType.STATE_TRANSITION, "002");
msgDown.setToState("SLAVE");
msgDown.setFromState("MASTER");
msgDown.setTgtName(HOSTNAME_PREFIX + 1);
messageOutput.addMessage(resource, partition, msgUp);
messageOutput.addMessage(resource, partition, msgDown);
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 0, "MASTER");
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 1, "SLAVE");
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 2, "SLAVE");
List<String> list = Arrays.asList(HOSTNAME_PREFIX + 0, HOSTNAME_PREFIX + 1, HOSTNAME_PREFIX + 2);
bestPossibleStateOutput.setPreferenceList(resource, partition.getPartitionName(), list);
} else {
// setup error partition
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 0, "MASTER");
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 1, "SLAVE");
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 2, "ERROR");
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 0, "MASTER");
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 1, "SLAVE");
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 2, "SLAVE");
List<String> list = Arrays.asList(HOSTNAME_PREFIX + 0, HOSTNAME_PREFIX + 1, HOSTNAME_PREFIX + 2);
bestPossibleStateOutput.setPreferenceList(resource, partition.getPartitionName(), list);
}
}
}
event.addAttribute(AttributeName.CURRENT_STATE.name(), currentStateOutput);
event.addAttribute(AttributeName.MESSAGES_SELECTED.name(), messageOutput);
event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), bestPossibleStateOutput);
event.addAttribute(AttributeName.ControllerDataProvider.name(),
new ResourceControllerDataProvider());

MsgRecordingPerReplicaThrottleStage msgRecordingStage = new MsgRecordingPerReplicaThrottleStage();
runStage(event, new ReadClusterDataStage());
runStage(event, msgRecordingStage);

List<Message> msgs = msgRecordingStage.getLoadThrottledMessages();
Assert.assertTrue(msgs.size() == 1);
Message msg = msgs.get(0);
Assert.assertTrue(msg.getId().equals("001"));

MessageOutput msgsOut = event.getAttribute(AttributeName.PER_REPLICA_OUTPUT_MESSAGES.name());
msgs = msgsOut.getMessages(resources[0], new Partition(resources[0] + "_" + 0));
Assert.assertTrue(msgs.size() == 1);
msg = msgs.get(0);
Assert.assertTrue(msg.getId().equals("002"));
}

// Case 5, Disabled instance not limited by throttling
@Test
public void testDisabledPartitionNotThrottled() {
String resourcePrefix = "resource";
int nResource = 1;
int nPartition = 1;
int nReplica = 3;
String[] resources = new String[nResource];
for (int i = 0; i < nResource; i++) {
resources[i] = resourcePrefix + "-" + i;
}

preSetup(resources, nPartition, nReplica);
setupInstances(nReplica);
setupThrottleConfig(0, 0);

// disable partition on instances
for (int i = 0; i < nReplica; i++ ) {
String instanceName = HOSTNAME_PREFIX + i;
PropertyKey key = accessor.keyBuilder().instanceConfig(instanceName);
InstanceConfig instanceConfig = accessor.getProperty(key);
instanceConfig.setInstanceEnabledForPartition(resources[0], resources[0] + "_" + 0, false);
accessor.setProperty(key, instanceConfig);
}

event.addAttribute(AttributeName.RESOURCES_TO_REBALANCE.name(),
getResourceMap(resources, nPartition, "MasterSlave"));
// setup current state; setup message output; setup best possible
CurrentStateOutput currentStateOutput = new CurrentStateOutput();
MessageOutput messageOutput = new MessageOutput();
BestPossibleStateOutput bestPossibleStateOutput = new BestPossibleStateOutput();
for (String resource : resources) {
for (int p = 0; p < nPartition; p++) {
Partition partition = new Partition(resource + "_" + p);
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 0, "MASTER");
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 1, "SLAVE");
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 2, "OFFLINE");
Message msg = new Message(Message.MessageType.STATE_TRANSITION, "001");
msg.setToState("SLAVE");
msg.setFromState("MASTER");
msg.setTgtName(HOSTNAME_PREFIX + 0);
messageOutput.addMessage(resource, partition, msg);
msg = new Message(Message.MessageType.STATE_TRANSITION, "002");
msg.setToState("OFFLINE");
msg.setFromState("SLAVE");
msg.setTgtName(HOSTNAME_PREFIX + 1);
messageOutput.addMessage(resource, partition, msg);
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 0, "OFFLINE");
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 1, "OFFLINE");
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 2, "OFFLINE");
List<String> list = Arrays.asList(HOSTNAME_PREFIX + 0, HOSTNAME_PREFIX + 1, HOSTNAME_PREFIX + 2);
bestPossibleStateOutput.setPreferenceList(resource, partition.getPartitionName(), list);
}
}
event.addAttribute(AttributeName.CURRENT_STATE.name(), currentStateOutput);
event.addAttribute(AttributeName.MESSAGES_SELECTED.name(), messageOutput);
event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), bestPossibleStateOutput);
event.addAttribute(AttributeName.ControllerDataProvider.name(),
new ResourceControllerDataProvider());

MsgRecordingPerReplicaThrottleStage msgRecordingStage = new MsgRecordingPerReplicaThrottleStage();
runStage(event, new ReadClusterDataStage());
runStage(event, msgRecordingStage);

List<Message> msgs = msgRecordingStage.getRecoveryThrottledMessages();
Assert.assertTrue(msgs.size() == 0);
msgs = msgRecordingStage.getLoadThrottledMessages();
Assert.assertTrue(msgs.size() == 0);
}

// Case 6, Drop message not limited by throttling
@Test
public void testDropMessageNoteThrottled() {
String resourcePrefix = "resource";
int nResource = 1;
int nPartition = 1;
int nReplica = 3;
String[] resources = new String[nResource];
for (int i = 0; i < nResource; i++) {
resources[i] = resourcePrefix + "-" + i;
}

preSetup(resources, nPartition, nReplica);
setupThrottleConfig(0, 0);

event.addAttribute(AttributeName.RESOURCES_TO_REBALANCE.name(),
getResourceMap(resources, nPartition, "MasterSlave"));
// setup current state; setup message output; setup best possible
CurrentStateOutput currentStateOutput = new CurrentStateOutput();
MessageOutput messageOutput = new MessageOutput();
BestPossibleStateOutput bestPossibleStateOutput = new BestPossibleStateOutput();
for (String resource : resources) {
for (int p = 0; p < nPartition; p++) {
Partition partition = new Partition(resource + "_" + p);
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 0, "MASTER");
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 1, "SLAVE");
currentStateOutput.setCurrentState(resource, partition, HOSTNAME_PREFIX + 2, "OFFLINE");
Message msg = new Message(Message.MessageType.STATE_TRANSITION, "001");
msg.setToState("ERROR");
msg.setFromState("SLAVE");
msg.setTgtName(HOSTNAME_PREFIX + 1);
messageOutput.addMessage(resource, partition, msg);
msg = new Message(Message.MessageType.STATE_TRANSITION, "002");
msg.setToState("DROPPED");
msg.setFromState("OFFLINE");
msg.setTgtName(HOSTNAME_PREFIX + 2);
messageOutput.addMessage(resource, partition, msg);
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 0, "MASTER");
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 1, "ERROR");
bestPossibleStateOutput.setState(resource, partition, HOSTNAME_PREFIX + 2, "DROPPED");
List<String> list = Arrays.asList(HOSTNAME_PREFIX + 0, HOSTNAME_PREFIX + 1, HOSTNAME_PREFIX + 2);
bestPossibleStateOutput.setPreferenceList(resource, partition.getPartitionName(), list);
}
}
event.addAttribute(AttributeName.CURRENT_STATE.name(), currentStateOutput);
event.addAttribute(AttributeName.MESSAGES_SELECTED.name(), messageOutput);
event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), bestPossibleStateOutput);
event.addAttribute(AttributeName.ControllerDataProvider.name(),
new ResourceControllerDataProvider());

MsgRecordingPerReplicaThrottleStage msgRecordingStage = new MsgRecordingPerReplicaThrottleStage();
runStage(event, new ReadClusterDataStage());
runStage(event, msgRecordingStage);

List<Message> msgs = msgRecordingStage.getRecoveryThrottledMessages();;
Assert.assertTrue(msgs.size() == 0);
msgs = msgRecordingStage.getLoadThrottledMessages();
Assert.assertTrue(msgs.size() == 0);
}

protected Map<String, Resource> getResourceMap(String[] resources, int partitions,
String stateModel) {
Map<String, Resource> resourceMap = new HashMap<String, Resource>();
Expand Down