From 79d7ae17752b65016c0efafadebb2dc130cf6146 Mon Sep 17 00:00:00 2001 From: Kai Sun Date: Thu, 28 Jan 2021 22:29:13 -0800 Subject: [PATCH] Per Replica Throttle -- 4th: add retraced map; downward-only handling and special handling test cases --- .../stages/PerReplicaThrottleStage.java | 31 ++- .../stages/TestPerReplicaThrottleStage.java | 212 +++++++++++++++++- 2 files changed, 241 insertions(+), 2 deletions(-) diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/PerReplicaThrottleStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/PerReplicaThrottleStage.java index a0bd57d9b2..00b48c9e8c 100644 --- a/helix-core/src/main/java/org/apache/helix/controller/stages/PerReplicaThrottleStage.java +++ b/helix-core/src/main/java/org/apache/helix/controller/stages/PerReplicaThrottleStage.java @@ -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> retracedPartitionsStateMap, + Map> outMessagesByPartition) { + for (Partition partition : resource.getPartitions()) { + List 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 retracedStateMap = retracedPartitionsStateMap.get(partition); + retracedStateMap.put(toInstance, toState); + } + } + } + private void propagateCountsTopDown( StateModelDefinition stateModelDef, Map expectedStateCountMap diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestPerReplicaThrottleStage.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestPerReplicaThrottleStage.java index 9f437bb9e8..6e7328b974 100644 --- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestPerReplicaThrottleStage.java +++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestPerReplicaThrottleStage.java @@ -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"; @@ -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 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 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 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 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 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 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 msgs = msgRecordingStage.getRecoveryThrottledMessages();; + Assert.assertTrue(msgs.size() == 0); + msgs = msgRecordingStage.getLoadThrottledMessages(); + Assert.assertTrue(msgs.size() == 0); + } + protected Map getResourceMap(String[] resources, int partitions, String stateModel) { Map resourceMap = new HashMap();