You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by hz...@apache.org on 2021/07/16 21:03:07 UTC
[helix] branch master updated (f1bd26f -> ac877c5)
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git.
from f1bd26f Stabilize the ZkClient monitor tests. (#1813)
new 113c933 Add java api for enable/disable cluster pause mode (#1740)
new 88dae34 Add model to record history and status of management mode (#1771)
new 7190eb3 Add management mode pipeline registry and switch logic (#1769)
new f9fba2a Move pause and maintenance handling out of controller (#1793)
new c95362e Add message util to create messages (#1796)
new 33ba7f1 Check cluster management mode status (#1798)
new b52bc04 Add REST APIs for management mode (#1807)
new 02c7ac0 Add message generation logic for management pipeline (#1803)
new 0aca85a Implement Participant Freeze Process (#1812)
new 849ab11 Add integration tests for cluster freeze mode (#1816)
new d2f5637 Refactor freeze mode naming and getStatus default value (#1818)
new 5fd9843 Rename live instance status PAUSED to FROZEN (#1819)
new ac877c5 Rename property CLUSTER_PAUSE to CLUSTER_FREEZE in PauseSignal (#1820)
The 13 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:
.../src/main/java/org/apache/helix/HelixAdmin.java | 23 ++
.../main/java/org/apache/helix/PropertyKey.java | 12 +-
.../java/org/apache/helix/PropertyPathBuilder.java | 8 +
.../main/java/org/apache/helix/PropertyType.java | 1 +
...sException.java => HelixConflictException.java} | 18 +-
.../helix/api/status/ClusterManagementMode.java | 100 ++++++
.../api/status/ClusterManagementModeRequest.java | 112 ++++++
.../helix/controller/GenericHelixController.java | 209 +++++++-----
.../dataproviders/BaseControllerDataProvider.java | 21 +-
.../ManagementControllerDataProvider.java} | 32 +-
.../apache/helix/controller/pipeline/Pipeline.java | 9 +-
...Exception.java => PipelineSwitchException.java} | 12 +-
.../helix/controller/stages/AttributeName.java | 3 +
.../helix/controller/stages/ClusterEventType.java | 1 +
.../stages/ManagementMessageDispatchStage.java | 62 ++++
.../stages/ManagementMessageGenerationPhase.java | 119 +++++++
.../controller/stages/ManagementModeStage.java | 194 +++++++++++
.../controller/stages/MessageGenerationPhase.java | 97 +-----
.../helix/controller/stages/MessageOutput.java | 11 +-
.../controller/stages/ResourceValidationStage.java | 29 ++
.../helix/manager/zk/CurStateCarryOverUpdater.java | 19 +-
.../helix/manager/zk/ParticipantManager.java | 79 +++--
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 87 ++++-
.../apache/helix/manager/zk/ZKHelixManager.java | 47 ++-
.../messaging/handling/HelixTaskExecutor.java | 116 ++++++-
.../messaging/handling/MessageHandlerFactory.java | 10 +
.../java/org/apache/helix/model/ClusterStatus.java | 83 +++++
.../org/apache/helix/model/ControllerHistory.java | 116 +++++--
.../java/org/apache/helix/model/LiveInstance.java | 31 +-
.../main/java/org/apache/helix/model/Message.java | 5 +
.../java/org/apache/helix/model/PauseSignal.java | 42 ++-
.../helix/participant/HelixStateMachineEngine.java | 27 +-
.../helix/participant/statemachine/StateModel.java | 7 +
.../BestPossibleExternalViewVerifier.java | 2 +-
.../main/java/org/apache/helix/util/HelixUtil.java | 33 ++
.../java/org/apache/helix/util/MessageUtil.java | 139 ++++++++
.../java/org/apache/helix/util/RebalanceUtil.java | 49 +++
.../org/apache/helix/TestPropertyPathBuilder.java | 3 +
...n.java => TestManagementMessageGeneration.java} | 123 ++-----
.../controller/stages/TestManagementModeStage.java | 144 ++++++++
.../controller/TestClusterFreezeMode.java | 376 +++++++++++++++++++++
.../paticipant/TestParticipantFreeze.java | 294 ++++++++++++++++
.../TestStateTransitionAppFailureHandling.java | 5 +-
.../apache/helix/manager/zk/TestZkHelixAdmin.java | 70 ++++
.../java/org/apache/helix/mock/MockHelixAdmin.java | 12 +
.../helix/model/TestControllerHistoryModel.java | 93 +++++
.../org/apache/helix/model/TestLiveInstance.java | 10 +
.../server/resources/helix/ClusterAccessor.java | 115 +++++++
.../helix/rest/server/TestClusterAccessor.java | 71 ++++
49 files changed, 2877 insertions(+), 404 deletions(-)
copy helix-core/src/main/java/org/apache/helix/api/exceptions/{HelixMetaDataAccessException.java => HelixConflictException.java} (65%)
create mode 100644 helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
create mode 100644 helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementModeRequest.java
copy helix-core/src/main/java/org/apache/helix/{tools/ClusterLiveNodesVerifier.java => controller/dataproviders/ManagementControllerDataProvider.java} (52%)
copy helix-core/src/main/java/org/apache/helix/controller/pipeline/{StageException.java => PipelineSwitchException.java} (81%)
create mode 100644 helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageDispatchStage.java
create mode 100644 helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
create mode 100644 helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
create mode 100644 helix-core/src/main/java/org/apache/helix/model/ClusterStatus.java
create mode 100644 helix-core/src/main/java/org/apache/helix/util/MessageUtil.java
copy helix-core/src/test/java/org/apache/helix/controller/stages/{TestCancellationMessageGeneration.java => TestManagementMessageGeneration.java} (51%)
create mode 100644 helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
create mode 100644 helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
create mode 100644 helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java
create mode 100644 helix-core/src/test/java/org/apache/helix/model/TestControllerHistoryModel.java
[helix] 06/13: Check cluster management mode status (#1798)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 33ba7f191437aecf353e6419e7b783334a6899c8
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Tue Jun 22 11:22:49 2021 -0700
Check cluster management mode status (#1798)
Controller needs to know the participant freeze status so it can send freeze/unfreeze messages for entering/exiting freeze mode. The status check is done in management mode stage.
This commit adds methods to check cluster management mode status, and update the status and history accordingly.
---
.../main/java/org/apache/helix/PropertyKey.java | 2 +-
.../java/org/apache/helix/PropertyPathBuilder.java | 1 +
.../pipeline/PipelineSwitchException.java | 29 +++++
.../controller/stages/ManagementModeStage.java | 134 +++++++++++++++++++
.../controller/stages/ResourceValidationStage.java | 6 +-
.../java/org/apache/helix/model/ClusterStatus.java | 5 +
.../org/apache/helix/model/ControllerHistory.java | 8 +-
.../controller/stages/TestManagementModeStage.java | 142 +++++++++++++++++++++
8 files changed, 322 insertions(+), 5 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/PropertyKey.java b/helix-core/src/main/java/org/apache/helix/PropertyKey.java
index 254cb95..eb5d5c1 100644
--- a/helix-core/src/main/java/org/apache/helix/PropertyKey.java
+++ b/helix-core/src/main/java/org/apache/helix/PropertyKey.java
@@ -247,7 +247,7 @@ public class PropertyKey {
* @return {@link PropertyKey}
*/
public PropertyKey clusterStatus() {
- return new PropertyKey(PropertyType.STATUS, ClusterStatus.class, _clusterName);
+ return new PropertyKey(PropertyType.STATUS, ClusterStatus.class, _clusterName, _clusterName);
}
/**
diff --git a/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java b/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java
index 34efd29..5c63304 100644
--- a/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java
+++ b/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java
@@ -89,6 +89,7 @@ public class PropertyPathBuilder {
addEntry(PropertyType.CUSTOMIZEDVIEW, 2, "/{clusterName}/CUSTOMIZEDVIEW/{customizedStateType}");
addEntry(PropertyType.CUSTOMIZEDVIEW, 3, "/{clusterName}/CUSTOMIZEDVIEW/{customizedStateType}/{resourceName}");
addEntry(PropertyType.STATUS, 1, "/{clusterName}/STATUS");
+ addEntry(PropertyType.STATUS, 2, "/{clusterName}/STATUS/{clusterName}");
addEntry(PropertyType.TARGETEXTERNALVIEW, 1, "/{clusterName}/TARGETEXTERNALVIEW");
addEntry(PropertyType.TARGETEXTERNALVIEW, 2,
diff --git a/helix-core/src/main/java/org/apache/helix/controller/pipeline/PipelineSwitchException.java b/helix-core/src/main/java/org/apache/helix/controller/pipeline/PipelineSwitchException.java
new file mode 100644
index 0000000..1584708
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/controller/pipeline/PipelineSwitchException.java
@@ -0,0 +1,29 @@
+package org.apache.helix.controller.pipeline;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Used to exit the current controller pipeline and switch to another pipeline.
+ */
+public class PipelineSwitchException extends StageException {
+ public PipelineSwitchException(String message) {
+ super(message);
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
index 042aa14..94ff1d3 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
@@ -19,11 +19,31 @@ package org.apache.helix.controller.stages;
* under the License.
*/
+import java.time.Instant;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.helix.AccessOption;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.PropertyType;
+import org.apache.helix.api.status.ClusterManagementMode;
import org.apache.helix.controller.LogUtil;
import org.apache.helix.controller.dataproviders.ManagementControllerDataProvider;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
+import org.apache.helix.controller.pipeline.StageException;
+import org.apache.helix.model.ClusterStatus;
+import org.apache.helix.model.ControllerHistory;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.LiveInstance.LiveInstanceStatus;
+import org.apache.helix.model.Message;
+import org.apache.helix.model.Message.MessageType;
+import org.apache.helix.model.PauseSignal;
import org.apache.helix.util.HelixUtil;
import org.apache.helix.util.RebalanceUtil;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,9 +58,23 @@ public class ManagementModeStage extends AbstractBaseStage {
// TODO: implement the stage
_eventId = event.getEventId();
String clusterName = event.getClusterName();
+ HelixManager manager = event.getAttribute(AttributeName.helixmanager.name());
+ if (manager == null) {
+ throw new StageException("HelixManager attribute value is null");
+ }
+
+ HelixDataAccessor accessor = manager.getHelixDataAccessor();
ManagementControllerDataProvider cache =
event.getAttribute(AttributeName.ControllerDataProvider.name());
+ ClusterManagementMode managementMode =
+ checkClusterFreezeStatus(cache.getEnabledLiveInstances(), cache.getLiveInstances(),
+ cache.getAllInstancesMessages(), cache.getPauseSignal());
+
+ recordClusterStatus(managementMode, accessor);
+ recordManagementModeHistory(managementMode, cache.getPauseSignal(), manager.getInstanceName(),
+ accessor);
+
// TODO: move to the last stage of management pipeline
checkInManagementMode(clusterName, cache);
}
@@ -53,4 +87,104 @@ public class ManagementModeStage extends AbstractBaseStage {
RebalanceUtil.enableManagementMode(clusterName, false);
}
}
+
+ // Checks cluster freeze, controller pause mode and status.
+ private ClusterManagementMode checkClusterFreezeStatus(
+ Set<String> enabledLiveInstances,
+ Map<String, LiveInstance> liveInstanceMap,
+ Map<String, Collection<Message>> allInstanceMessages,
+ PauseSignal pauseSignal) {
+ ClusterManagementMode.Type type;
+ ClusterManagementMode.Status status = ClusterManagementMode.Status.COMPLETED;
+ if (pauseSignal == null) {
+ // TODO: Should check maintenance mode after it's moved to management pipeline.
+ type = ClusterManagementMode.Type.NORMAL;
+ if (HelixUtil.inManagementMode(pauseSignal, liveInstanceMap, enabledLiveInstances,
+ allInstanceMessages)) {
+ status = ClusterManagementMode.Status.IN_PROGRESS;
+ }
+ } else if (pauseSignal.isClusterPause()) {
+ type = ClusterManagementMode.Type.CLUSTER_PAUSE;
+ if (!instancesFullyFrozen(enabledLiveInstances, liveInstanceMap, allInstanceMessages)) {
+ status = ClusterManagementMode.Status.IN_PROGRESS;
+ }
+ } else {
+ type = ClusterManagementMode.Type.CONTROLLER_PAUSE;
+ }
+
+ return new ClusterManagementMode(type, status);
+ }
+
+ private boolean instancesFullyFrozen(Set<String> enabledLiveInstances,
+ Map<String, LiveInstance> liveInstanceMap,
+ Map<String, Collection<Message>> allInstanceMessages) {
+ // 1. All live instances are frozen
+ // 2. No pending participant status change message.
+ return enabledLiveInstances.stream().noneMatch(
+ instance -> !LiveInstanceStatus.PAUSED.equals(liveInstanceMap.get(instance).getStatus())
+ || hasPendingMessage(allInstanceMessages.get(instance),
+ MessageType.PARTICIPANT_STATUS_CHANGE));
+ }
+
+ private boolean hasPendingMessage(Collection<Message> messages, MessageType type) {
+ return messages != null && messages.stream()
+ .anyMatch(message -> type.name().equals(message.getMsgType()));
+ }
+
+ private void recordClusterStatus(ClusterManagementMode mode, HelixDataAccessor accessor) {
+ // update cluster status
+ PropertyKey statusPropertyKey = accessor.keyBuilder().clusterStatus();
+ ClusterStatus clusterStatus = accessor.getProperty(statusPropertyKey);
+ if (clusterStatus == null) {
+ clusterStatus = new ClusterStatus();
+ }
+
+ ClusterManagementMode.Type recordedType = clusterStatus.getManagementMode();
+ ClusterManagementMode.Status recordedStatus = clusterStatus.getManagementModeStatus();
+
+ // If there is any pending message sent by users, status could be computed as in progress.
+ // Skip recording status change to avoid confusion after cluster is already fully frozen.
+ if (ClusterManagementMode.Type.CLUSTER_PAUSE.equals(recordedType)
+ && ClusterManagementMode.Status.COMPLETED.equals(recordedStatus)
+ && ClusterManagementMode.Type.CLUSTER_PAUSE.equals(mode.getMode())
+ && ClusterManagementMode.Status.IN_PROGRESS.equals(mode.getStatus())) {
+ LOG.info("Skip recording status mode={}, status={}, because cluster is fully frozen",
+ mode.getMode(), mode.getStatus());
+ return;
+ }
+
+ if (!mode.getMode().equals(recordedType) || !mode.getStatus().equals(recordedStatus)) {
+ // Only update status when it's different with metadata store
+ clusterStatus.setManagementMode(mode.getMode());
+ clusterStatus.setManagementModeStatus(mode.getStatus());
+ if (!accessor.updateProperty(statusPropertyKey, clusterStatus)) {
+ LOG.error("Failed to update cluster status {}", clusterStatus);
+ }
+ }
+ }
+
+ private void recordManagementModeHistory(ClusterManagementMode mode, PauseSignal pauseSignal,
+ String controllerName, HelixDataAccessor accessor) {
+ // Only record completed status
+ if (!ClusterManagementMode.Status.COMPLETED.equals(mode.getStatus())) {
+ return;
+ }
+
+ // Record a management mode history in controller history
+ String path = accessor.keyBuilder().controllerLeaderHistory().getPath();
+ long timestamp = Instant.now().toEpochMilli();
+ String fromHost = (pauseSignal == null ? null : pauseSignal.getFromHost());
+ String reason = (pauseSignal == null ? null : pauseSignal.getReason());
+
+ // Need the updater to avoid race condition with controller/maintenance history updates.
+ if (!accessor.getBaseDataAccessor().update(path, oldRecord -> {
+ if (oldRecord == null) {
+ oldRecord = new ZNRecord(PropertyType.HISTORY.toString());
+ }
+ return new ControllerHistory(oldRecord)
+ .updateManagementModeHistory(controllerName, mode, fromHost, timestamp, reason);
+ }, AccessOption.PERSISTENT)) {
+ LOG.error("Failed to write management mode history to ZK!");
+ }
+ }
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java
index 613ce2e..696506a 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java
@@ -24,6 +24,7 @@ import java.util.Map;
import org.apache.helix.controller.LogUtil;
import org.apache.helix.controller.dataproviders.BaseControllerDataProvider;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
+import org.apache.helix.controller.pipeline.PipelineSwitchException;
import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.Resource;
@@ -102,9 +103,10 @@ public class ResourceValidationStage extends AbstractBaseStage {
LogUtil.logInfo(LOG, _eventId,
"Enabling management mode pipeline for cluster " + event.getClusterName());
RebalanceUtil.enableManagementMode(event.getClusterName(), true);
- throw new StageException(
+ // TODO: redesign to terminate and switch pipeline more peacefully
+ throw new PipelineSwitchException(
"Pipeline should not be run because cluster " + event.getClusterName()
- + "is in management mode");
+ + " is in management mode");
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/model/ClusterStatus.java b/helix-core/src/main/java/org/apache/helix/model/ClusterStatus.java
index 6ed354c..a405fe3 100644
--- a/helix-core/src/main/java/org/apache/helix/model/ClusterStatus.java
+++ b/helix-core/src/main/java/org/apache/helix/model/ClusterStatus.java
@@ -22,6 +22,7 @@ package org.apache.helix.model;
import org.apache.helix.HelixProperty;
import org.apache.helix.PropertyType;
import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
/**
* Represents the cluster status. It can have fields for
@@ -32,6 +33,10 @@ public class ClusterStatus extends HelixProperty {
super(PropertyType.STATUS.name());
}
+ public ClusterStatus(ZNRecord record) {
+ super(record);
+ }
+
public enum ClusterStatusProperty {
MANAGEMENT_MODE,
MANAGEMENT_MODE_STATUS
diff --git a/helix-core/src/main/java/org/apache/helix/model/ControllerHistory.java b/helix-core/src/main/java/org/apache/helix/model/ControllerHistory.java
index 4e418c3..47b0958 100644
--- a/helix-core/src/main/java/org/apache/helix/model/ControllerHistory.java
+++ b/helix-core/src/main/java/org/apache/helix/model/ControllerHistory.java
@@ -162,8 +162,12 @@ public class ControllerHistory extends HelixProperty {
historyEntry.put(ConfigProperty.TIME.name(), Instant.ofEpochMilli(time).toString());
historyEntry.put(ManagementModeConfigKey.MODE.name(), mode.getMode().name());
historyEntry.put(ManagementModeConfigKey.STATUS.name(), mode.getStatus().name());
- historyEntry.put(PauseSignal.PauseSignalProperty.FROM_HOST.name(), fromHost);
- historyEntry.put(PauseSignal.PauseSignalProperty.REASON.name(), reason);
+ if (fromHost != null) {
+ historyEntry.put(PauseSignal.PauseSignalProperty.FROM_HOST.name(), fromHost);
+ }
+ if (reason != null) {
+ historyEntry.put(PauseSignal.PauseSignalProperty.REASON.name(), reason);
+ }
return populateHistoryEntries(HistoryType.MANAGEMENT_MODE, historyEntry.toString());
}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
new file mode 100644
index 0000000..28ca524
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
@@ -0,0 +1,142 @@
+package org.apache.helix.controller.stages;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.List;
+
+import org.apache.helix.HelixConstants;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.TestHelper;
+import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.api.status.ClusterManagementModeRequest;
+import org.apache.helix.common.ZkTestBase;
+import org.apache.helix.controller.dataproviders.ManagementControllerDataProvider;
+import org.apache.helix.controller.pipeline.Pipeline;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.ClusterStatus;
+import org.apache.helix.model.LiveInstance;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+public class TestManagementModeStage extends ZkTestBase {
+ HelixManager _manager;
+ HelixDataAccessor _accessor;
+ String _clusterName;
+
+ @BeforeClass
+ public void beforeClass() {
+ _clusterName = "CLUSTER_" + TestHelper.getTestClassName();
+ _accessor = new ZKHelixDataAccessor(_clusterName, new ZkBaseDataAccessor<>(_gZkClient));
+ _manager = new DummyClusterManager(_clusterName, _accessor);
+ }
+
+ @AfterClass
+ public void afterClass() {
+ deleteLiveInstances(_clusterName);
+ deleteCluster(_clusterName);
+ }
+
+ @Test
+ public void testClusterFreezeStatus() throws Exception {
+ // ideal state: node0 is MASTER, node1 is SLAVE
+ // replica=2 means 1 master and 1 slave
+ setupIdealState(_clusterName, new int[]{0, 1}, new String[]{"TestDB"}, 1, 2);
+ List<LiveInstance> liveInstances = setupLiveInstances(_clusterName, new int[]{0, 1});
+ setupStateModel(_clusterName);
+
+ ClusterEvent event = new ClusterEvent(_clusterName, ClusterEventType.Unknown);
+ ManagementControllerDataProvider cache = new ManagementControllerDataProvider(_clusterName,
+ Pipeline.Type.MANAGEMENT_MODE.name());
+ event.addAttribute(AttributeName.helixmanager.name(), _manager);
+ event.addAttribute(AttributeName.ControllerDataProvider.name(), cache);
+
+ // Freeze cluster
+ ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
+ .withClusterName(_clusterName)
+ .withMode(ClusterManagementMode.Type.CLUSTER_PAUSE)
+ .withReason("test")
+ .build();
+ _gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
+
+ Pipeline dataRefresh = new Pipeline();
+ dataRefresh.addStage(new ReadClusterDataStage());
+ runPipeline(event, dataRefresh, false);
+ ManagementModeStage managementModeStage = new ManagementModeStage();
+ managementModeStage.process(event);
+
+ // In frozen mode
+ ClusterStatus clusterStatus = _accessor.getProperty(_accessor.keyBuilder().clusterStatus());
+ Assert.assertEquals(clusterStatus.getManagementMode(), ClusterManagementMode.Type.CLUSTER_PAUSE);
+
+
+ // Mark a live instance to be pause state
+ LiveInstance liveInstance = liveInstances.get(0);
+ liveInstance.setStatus(LiveInstance.LiveInstanceStatus.PAUSED);
+ PropertyKey liveInstanceKey =
+ _accessor.keyBuilder().liveInstance(liveInstance.getInstanceName());
+ _accessor.updateProperty(liveInstanceKey, liveInstance);
+ // Require cache refresh
+ cache.notifyDataChange(HelixConstants.ChangeType.LIVE_INSTANCE);
+
+ // Unfreeze cluster
+ request = ClusterManagementModeRequest.newBuilder()
+ .withClusterName(_clusterName)
+ .withMode(ClusterManagementMode.Type.NORMAL)
+ .withReason("test")
+ .build();
+ _gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
+ runPipeline(event, dataRefresh, false);
+ managementModeStage.process(event);
+ clusterStatus = _accessor.getProperty(_accessor.keyBuilder().clusterStatus());
+
+ Assert.assertEquals(clusterStatus.getManagementMode(), ClusterManagementMode.Type.NORMAL);
+ // In progress because a live instance is still frozen
+ Assert.assertEquals(clusterStatus.getManagementModeStatus(),
+ ClusterManagementMode.Status.IN_PROGRESS);
+
+ // remove froze status to mark the live instance to be normal status
+ liveInstance = _accessor.getProperty(liveInstanceKey);
+ liveInstance.getRecord().getSimpleFields()
+ .remove(LiveInstance.LiveInstanceProperty.STATUS.name());
+ _accessor.setProperty(liveInstanceKey, liveInstance);
+ // Require cache refresh
+ cache.notifyDataChange(HelixConstants.ChangeType.LIVE_INSTANCE);
+ runPipeline(event, dataRefresh, false);
+ try {
+ managementModeStage.process(event);
+ } catch (HelixException expected) {
+ // It's expected because controller does not set for cluster.
+ Assert.assertTrue(expected.getMessage()
+ .startsWith("Failed to switch management mode pipeline, enabled=false"));
+ }
+ clusterStatus = _accessor.getProperty(_accessor.keyBuilder().clusterStatus());
+
+ // Fully existed frozen mode
+ Assert.assertEquals(clusterStatus.getManagementMode(), ClusterManagementMode.Type.NORMAL);
+ Assert.assertEquals(clusterStatus.getManagementModeStatus(),
+ ClusterManagementMode.Status.COMPLETED);
+ }
+}
[helix] 01/13: Add java api for enable/disable cluster pause mode
(#1740)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 113c9335761fbe34f10914003c1e529c91d68527
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Wed May 19 15:45:11 2021 -0700
Add java api for enable/disable cluster pause mode (#1740)
Cluster pause mode feature is going to be added. This commit adds java api to set cluster pause mode: void setClusterManagementMode(ClusterManagementModeRequest request);
---
.../src/main/java/org/apache/helix/HelixAdmin.java | 14 +++
.../api/exceptions/HelixConflictException.java | 33 +++++++
.../helix/api/status/ClusterManagementMode.java | 70 ++++++++++++++
.../api/status/ClusterManagementModeRequest.java | 102 +++++++++++++++++++++
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 80 +++++++++++++++-
.../java/org/apache/helix/model/LiveInstance.java | 28 +++++-
.../java/org/apache/helix/model/PauseSignal.java | 42 ++++++++-
.../apache/helix/manager/zk/TestZkHelixAdmin.java | 70 ++++++++++++++
.../java/org/apache/helix/mock/MockHelixAdmin.java | 6 ++
.../org/apache/helix/model/TestLiveInstance.java | 8 ++
10 files changed, 449 insertions(+), 4 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/HelixAdmin.java b/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
index 3c456bb..d403571 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
@@ -37,6 +37,8 @@ import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.MaintenanceSignal;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
+import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.api.status.ClusterManagementModeRequest;
/*
* Helix cluster management
@@ -368,6 +370,18 @@ public interface HelixAdmin {
boolean isInMaintenanceMode(String clusterName);
/**
+ * Requests to put a cluster into a management mode
+ * {@link ClusterManagementMode.Type}. When this method returns,
+ * it means the signal has been successfully sent, but it does not mean the cluster has
+ * fully entered the mode. Because the cluster can take some time to complete the request.
+ * <p>
+ * To check the cluster management mode status, call {@link #getClusterManagementMode(String)}.
+ *
+ * @param request request to set the cluster management mode. {@link ClusterManagementModeRequest}
+ */
+ void setClusterManagementMode(ClusterManagementModeRequest request);
+
+ /**
* Reset a list of partitions in error state for an instance
* The partitions are assume to be in error state and reset will bring them from error
* to initial state. An error to initial state transition is required for reset.
diff --git a/helix-core/src/main/java/org/apache/helix/api/exceptions/HelixConflictException.java b/helix-core/src/main/java/org/apache/helix/api/exceptions/HelixConflictException.java
new file mode 100644
index 0000000..c626155
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/api/exceptions/HelixConflictException.java
@@ -0,0 +1,33 @@
+package org.apache.helix.api.exceptions;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.helix.HelixException;
+
+/**
+ * Occurs when a conflict with a previous successful write is detected. This generally occurs when
+ * a write request is conflicted with the existing data. Eg. if a cluster is already in cluster
+ * pause mode, a request of enabling maintenance mode is a conflict.
+ */
+public class HelixConflictException extends HelixException {
+ public HelixConflictException(String message) {
+ super(message);
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
new file mode 100644
index 0000000..d7a1637
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
@@ -0,0 +1,70 @@
+package org.apache.helix.api.status;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Represents the management mode of the cluster:
+ * 1. what type of mode it targets to be;
+ * 2. what progress status it is now.
+ */
+public class ClusterManagementMode {
+ /** Represents */
+ public enum Type {
+ /** Cluster is not in any pause or maintenance mode */
+ NORMAL,
+
+ /**
+ * Puts a cluster into pause mode, which will pause controller and participants.
+ * This can be used to retain the cluster state.
+ */
+ CLUSTER_PAUSE,
+
+ /** Pause controller only, but not participants. */
+ CONTROLLER_PAUSE,
+
+ /** Put cluster into maintenance mode. */
+ MAINTENANCE
+ }
+
+ /** Current status of the cluster mode */
+ public enum Status {
+ /** Cluster is in progress to the target {@link Type} of mode */
+ IN_PROGRESS,
+
+ /** Cluster is fully stable in the target {@link Type} of mode */
+ COMPLETED
+ }
+
+ private final Type mode;
+ private final Status status;
+
+ public ClusterManagementMode(Type mode, Status status) {
+ this.mode = mode;
+ this.status = status;
+ }
+
+ public Status getStatus() {
+ return status;
+ }
+
+ public Type getMode() {
+ return mode;
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementModeRequest.java b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementModeRequest.java
new file mode 100644
index 0000000..dd2fe58
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementModeRequest.java
@@ -0,0 +1,102 @@
+package org.apache.helix.api.status;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Represents a request to set the cluster management mode {@link ClusterManagementMode}
+ */
+public class ClusterManagementModeRequest {
+ private final ClusterManagementMode.Type _mode;
+ private final String _clusterName;
+ private final String _reason;
+ private final boolean _cancelPendingST;
+
+ public ClusterManagementMode.Type getMode() {
+ return _mode;
+ }
+
+ public String getClusterName() {
+ return _clusterName;
+ }
+
+ public String getReason() {
+ return _reason;
+ }
+
+ public boolean isCancelPendingST() {
+ return _cancelPendingST;
+ }
+
+ public ClusterManagementModeRequest(Builder builder) {
+ _mode = builder.mode;
+ _clusterName = builder.clusterName;
+ _reason = builder.reason;
+ _cancelPendingST = builder.cancelPendingST;
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+ private ClusterManagementMode.Type mode;
+ private String clusterName;
+ private String reason = "";
+ private boolean cancelPendingST;
+
+ public Builder withMode(ClusterManagementMode.Type mode) {
+ this.mode = mode;
+ return this;
+ }
+
+ public Builder withClusterName(String clusterName) {
+ this.clusterName = clusterName;
+ return this;
+ }
+
+ public Builder withReason(String reason) {
+ this.reason = reason;
+ return this;
+ }
+
+ /**
+ * If mode is not CLUSTER_PAUSE, this should not be set to true.
+ *
+ * @param cancelPendingST whether or not cancel pending ST for CLUSTER_PAUSE mode.
+ * @return {@link Builder}
+ */
+ public Builder withCancelPendingST(boolean cancelPendingST) {
+ this.cancelPendingST = cancelPendingST;
+ return this;
+ }
+
+ public ClusterManagementModeRequest build() {
+ validate();
+ return new ClusterManagementModeRequest(this);
+ }
+
+ private void validate() {
+ Preconditions.checkNotNull(mode, "Mode not set");
+ Preconditions.checkNotNull(clusterName, "Cluster name not set");
+ }
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
index 62840b6..2545139 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
@@ -25,6 +25,7 @@ import java.io.FileInputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -51,9 +52,9 @@ import org.apache.helix.PropertyKey;
import org.apache.helix.PropertyPathBuilder;
import org.apache.helix.PropertyType;
import org.apache.helix.SystemPropertyKeys;
+import org.apache.helix.api.exceptions.HelixConflictException;
+import org.apache.helix.api.status.ClusterManagementMode;
import org.apache.helix.api.topology.ClusterTopology;
-import org.apache.helix.controller.rebalancer.DelayedAutoRebalancer;
-import org.apache.helix.controller.rebalancer.strategy.CrushEdRebalanceStrategy;
import org.apache.helix.controller.rebalancer.strategy.RebalanceStrategy;
import org.apache.helix.controller.rebalancer.util.WagedValidationUtil;
import org.apache.helix.controller.rebalancer.waged.WagedRebalancer;
@@ -80,6 +81,7 @@ import org.apache.helix.model.ParticipantHistory;
import org.apache.helix.model.PauseSignal;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
+import org.apache.helix.api.status.ClusterManagementModeRequest;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.tools.DefaultIdealStateCalculator;
import org.apache.helix.util.HelixUtil;
@@ -93,6 +95,7 @@ import org.apache.helix.zookeeper.impl.client.FederatedZkClient;
import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
+import org.apache.helix.zookeeper.zkclient.NetworkUtil;
import org.apache.helix.zookeeper.zkclient.exception.ZkException;
import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
import org.apache.zookeeper.KeeperException;
@@ -498,6 +501,79 @@ public class ZKHelixAdmin implements HelixAdmin {
}
@Override
+ public void setClusterManagementMode(ClusterManagementModeRequest request) {
+ ClusterManagementMode.Type mode = request.getMode();
+ String clusterName = request.getClusterName();
+ String reason = request.getReason();
+
+ // TODO: support other modes
+ switch (mode) {
+ case CLUSTER_PAUSE:
+ enableClusterPauseMode(clusterName, request.isCancelPendingST(), reason);
+ break;
+ case NORMAL:
+ // If from other modes, should check what mode it is in and call the api accordingly.
+ // If we put all mode config in one znode, one generic method is good enough.
+ disableClusterPauseMode(clusterName);
+ break;
+ default:
+ throw new IllegalArgumentException("ClusterManagementMode " + mode + " is not supported");
+ }
+ }
+
+ private void enableClusterPauseMode(String clusterName, boolean cancelPendingST, String reason) {
+ String hostname = NetworkUtil.getLocalhostName();
+ logger.info(
+ "Enable cluster pause mode for cluster: {}. CancelPendingST: {}. Reason: {}. From Host: {}",
+ clusterName, cancelPendingST, reason, hostname);
+
+ BaseDataAccessor<ZNRecord> baseDataAccessor = new ZkBaseDataAccessor<>(_zkClient);
+ HelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, baseDataAccessor);
+
+ if (baseDataAccessor.exists(accessor.keyBuilder().pause().getPath(), AccessOption.PERSISTENT)) {
+ throw new HelixConflictException(clusterName + " pause signal already exists");
+ }
+ if (baseDataAccessor.exists(accessor.keyBuilder().maintenance().getPath(), AccessOption.PERSISTENT)) {
+ throw new HelixConflictException(clusterName + " maintenance signal already exists");
+ }
+
+ // check whether cancellation is enabled
+ ClusterConfig config = accessor.getProperty(accessor.keyBuilder().clusterConfig());
+ if (cancelPendingST && !config.isStateTransitionCancelEnabled()) {
+ throw new HelixConflictException(
+ "State transition cancellation not enabled in " + clusterName);
+ }
+
+ PauseSignal pauseSignal = new PauseSignal();
+ pauseSignal.setClusterPause(true);
+ pauseSignal.setCancelPendingST(cancelPendingST);
+ pauseSignal.setFromHost(hostname);
+ pauseSignal.setTriggerTime(Instant.now().toEpochMilli());
+ if (reason != null && !reason.isEmpty()) {
+ pauseSignal.setReason(reason);
+ }
+ // TODO: merge management status signal into one znode to avoid race condition
+ if (!accessor.createPause(pauseSignal)) {
+ throw new HelixException("Failed to create pause signal");
+ }
+ }
+
+ private void disableClusterPauseMode(String clusterName) {
+ logger.info("Disable cluster pause mode for cluster: {}", clusterName);
+ HelixDataAccessor accessor =
+ new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<>(_zkClient));
+ PropertyKey pausePropertyKey = accessor.keyBuilder().pause();
+ PauseSignal pauseSignal = accessor.getProperty(pausePropertyKey);
+ if (pauseSignal == null || !pauseSignal.isClusterPause()) {
+ throw new HelixException("Cluster pause mode is not enabled for cluster " + clusterName);
+ }
+
+ if (!accessor.removeProperty(pausePropertyKey)) {
+ throw new HelixException("Failed to disable cluster pause mode for cluster: " + clusterName);
+ }
+ }
+
+ @Override
@Deprecated
public void enableMaintenanceMode(String clusterName, boolean enabled, String reason) {
manuallyEnableMaintenanceMode(clusterName, enabled, reason, null);
diff --git a/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java b/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
index 0adccdd..7a9671d 100644
--- a/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
+++ b/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
@@ -40,7 +40,17 @@ public class LiveInstance extends HelixProperty {
LIVE_INSTANCE,
ZKPROPERTYTRANSFERURL,
RESOURCE_CAPACITY,
- CURRENT_TASK_THREAD_POOL_SIZE
+ CURRENT_TASK_THREAD_POOL_SIZE,
+
+ /** Represents the status of live instance, eg. PAUSED */
+ STATUS
+ }
+
+ /**
+ * Saved values for the {@link LiveInstanceProperty#STATUS} field
+ */
+ public enum LiveInstanceStatus {
+ PAUSED
}
/**
@@ -131,6 +141,22 @@ public class LiveInstance extends HelixProperty {
}
/**
+ * Gets the live instance's status. Returns null if the status field is not set.
+ */
+ public LiveInstanceStatus getStatus() {
+ return _record.getEnumField(LiveInstanceProperty.STATUS.name(), LiveInstanceStatus.class, null);
+ }
+
+ /**
+ * Sets the status in simple field.
+ *
+ * @param status status value
+ */
+ public void setStatus(LiveInstanceStatus status) {
+ _record.setEnumField(LiveInstanceProperty.STATUS.name(), status);
+ }
+
+ /**
* Get an identifier that represents the instance and where it is located
* @return identifier, e.g. process_id@host
*/
diff --git a/helix-core/src/main/java/org/apache/helix/model/PauseSignal.java b/helix-core/src/main/java/org/apache/helix/model/PauseSignal.java
index e68345d..bcfd17d 100644
--- a/helix-core/src/main/java/org/apache/helix/model/PauseSignal.java
+++ b/helix-core/src/main/java/org/apache/helix/model/PauseSignal.java
@@ -19,6 +19,8 @@ package org.apache.helix.model;
* under the License.
*/
+import java.time.Instant;
+
import org.apache.helix.HelixProperty;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
@@ -26,9 +28,18 @@ import org.apache.helix.zookeeper.datamodel.ZNRecord;
* Represent a pause in the cluster
*/
public class PauseSignal extends HelixProperty {
+ private static final String DEFAULT_PAUSE_ID = "pause";
public enum PauseSignalProperty {
- REASON
+ REASON,
+ CLUSTER_PAUSE,
+ FROM_HOST,
+ CANCEL_PENDING_ST,
+ TRIGGER_TIME
+ }
+
+ public PauseSignal() {
+ this(DEFAULT_PAUSE_ID);
}
/**
@@ -63,4 +74,33 @@ public class PauseSignal extends HelixProperty {
public boolean isValid() {
return true;
}
+
+ public void setClusterPause(boolean pause) {
+ _record.setBooleanField(PauseSignalProperty.CLUSTER_PAUSE.name(), pause);
+ }
+
+ public boolean isClusterPause() {
+ return _record.getBooleanField(PauseSignalProperty.CLUSTER_PAUSE.name(), false);
+ }
+
+ public void setFromHost(String host) {
+ _record.setSimpleField(PauseSignalProperty.FROM_HOST.name(), host);
+ }
+
+ public String getFromHost() {
+ return _record.getSimpleField(PauseSignalProperty.FROM_HOST.name());
+ }
+
+ public void setCancelPendingST(boolean cancel) {
+ _record.setBooleanField(PauseSignalProperty.CANCEL_PENDING_ST.name(), cancel);
+ }
+
+ public boolean getCancelPendingST() {
+ return _record.getBooleanField(PauseSignalProperty.CANCEL_PENDING_ST.name(), false);
+ }
+
+ public void setTriggerTime(long time) {
+ _record.setSimpleField(PauseSignalProperty.TRIGGER_TIME.name(),
+ Instant.ofEpochMilli(time).toString());
+ }
}
diff --git a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java
index ce6ea9d..7ddf677 100644
--- a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java
+++ b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java
@@ -45,10 +45,13 @@ import org.apache.helix.PropertyPathBuilder;
import org.apache.helix.PropertyType;
import org.apache.helix.TestHelper;
import org.apache.helix.ZkUnitTestBase;
+import org.apache.helix.api.exceptions.HelixConflictException;
+import org.apache.helix.api.status.ClusterManagementMode;
import org.apache.helix.api.topology.ClusterTopology;
import org.apache.helix.cloud.constants.CloudProvider;
import org.apache.helix.controller.rebalancer.waged.WagedRebalancer;
import org.apache.helix.examples.MasterSlaveStateModelFactory;
+import org.apache.helix.integration.manager.ClusterControllerManager;
import org.apache.helix.integration.manager.MockParticipantManager;
import org.apache.helix.model.CloudConfig;
import org.apache.helix.model.ClusterConfig;
@@ -65,15 +68,18 @@ import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.MasterSlaveSMD;
+import org.apache.helix.model.PauseSignal;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.model.builder.ConstraintItemBuilder;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
+import org.apache.helix.api.status.ClusterManagementModeRequest;
import org.apache.helix.participant.StateMachineEngine;
import org.apache.helix.tools.StateModelConfigGenerator;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.exception.ZkClientException;
+import org.apache.helix.zookeeper.zkclient.NetworkUtil;
import org.apache.helix.zookeeper.zkclient.exception.ZkException;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
@@ -1060,4 +1066,68 @@ public class TestZkHelixAdmin extends ZkUnitTestBase {
tool.dropCluster(clusterName);
System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis()));
}
+
+ /*
+ * Tests 2 APIs: enable and disable cluster pause mode.
+ */
+ @Test
+ public void testEnableDisableClusterPauseMode() {
+ String className = TestHelper.getTestClassName();
+ String methodName = TestHelper.getTestMethodName();
+ String clusterName = className + "_" + methodName;
+
+ _gSetupTool.setupTestCluster(clusterName);
+ ClusterControllerManager controller =
+ new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0");
+ controller.syncStart();
+ _gSetupTool.activateCluster(clusterName, controller.getClusterName(), true);
+
+ try {
+ // Should not create pause with pending cancel ST enabled because cancellation is not enabled
+ try {
+ ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
+ .withClusterName(clusterName)
+ .withMode(ClusterManagementMode.Type.CLUSTER_PAUSE)
+ .withCancelPendingST(true)
+ .withReason(methodName)
+ .build();
+ _gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
+ Assert.fail("Should not create pause with pending cancel ST enabled because "
+ + "cancellation is not enabled");
+ } catch (HelixConflictException e) {
+ Assert.assertTrue(e.getMessage().startsWith("State transition cancellation not enabled"));
+ }
+
+ ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
+ .withClusterName(clusterName)
+ .withMode(ClusterManagementMode.Type.CLUSTER_PAUSE)
+ .withReason(methodName)
+ .build();
+ _gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
+ HelixDataAccessor dataAccessor = new ZKHelixDataAccessor(clusterName, _baseAccessor);
+ PauseSignal pauseSignal = dataAccessor.getProperty(dataAccessor.keyBuilder().pause());
+
+ // Verify pause signal is correctly written
+ Assert.assertNotNull(pauseSignal);
+ Assert.assertTrue(pauseSignal.isClusterPause());
+ Assert.assertFalse(pauseSignal.getCancelPendingST());
+ Assert.assertEquals(pauseSignal.getFromHost(), NetworkUtil.getLocalhostName());
+ Assert.assertEquals(pauseSignal.getReason(), methodName);
+
+ // Disable pause mode
+ request = ClusterManagementModeRequest.newBuilder()
+ .withClusterName(clusterName)
+ .withMode(ClusterManagementMode.Type.NORMAL)
+ .build();
+ _gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
+ pauseSignal = dataAccessor.getProperty(dataAccessor.keyBuilder().pause());
+
+ // Verify pause signal has been deleted.
+ Assert.assertNull(pauseSignal);
+ } finally {
+ _gSetupTool.activateCluster(clusterName, controller.getClusterName(), false);
+ controller.syncStop();
+ _gSetupTool.deleteCluster(clusterName);
+ }
+ }
}
diff --git a/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java b/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java
index 42633c1..2bacc74 100644
--- a/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java
+++ b/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java
@@ -44,6 +44,7 @@ import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.MaintenanceSignal;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
+import org.apache.helix.api.status.ClusterManagementModeRequest;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
public class MockHelixAdmin implements HelixAdmin {
@@ -325,6 +326,11 @@ public class MockHelixAdmin implements HelixAdmin {
return false;
}
+ @Override
+ public void setClusterManagementMode(ClusterManagementModeRequest request) {
+
+ }
+
@Override public void resetPartition(String clusterName, String instanceName, String resourceName,
List<String> partitionNames) {
diff --git a/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java b/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java
index 39f0a57..18285f5 100644
--- a/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java
+++ b/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java
@@ -146,4 +146,12 @@ public class TestLiveInstance extends ZkUnitTestBase {
Assert.assertEquals(testLiveInstance.getCurrentTaskThreadPoolSize(), 100);
}
+
+ @Test
+ public void testLiveInstanceStatus() {
+ LiveInstance testLiveInstance = new LiveInstance("testLiveInstanceStatus");
+ Assert.assertNull(testLiveInstance.getStatus());
+ testLiveInstance.setStatus(LiveInstance.LiveInstanceStatus.PAUSED);
+ Assert.assertEquals(testLiveInstance.getStatus(), LiveInstance.LiveInstanceStatus.PAUSED);
+ }
}
[helix] 12/13: Rename live instance status PAUSED to FROZEN (#1819)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 5fd984353c42091b4dc7750b6c34c97c6a2b9972
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Thu Jul 15 14:08:08 2021 -0700
Rename live instance status PAUSED to FROZEN (#1819)
Since cluster freeze mode is naming as "FREEZE", live instance status should be named to "FROZEN" accordingly.
This commit renames the enum live instance status PAUSED to FROZEN. No logic change.
---
.../java/org/apache/helix/api/status/ClusterManagementMode.java | 2 +-
.../org/apache/helix/controller/stages/ManagementModeStage.java | 2 +-
.../main/java/org/apache/helix/manager/zk/ParticipantManager.java | 2 +-
.../src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java | 4 ++--
.../org/apache/helix/messaging/handling/HelixTaskExecutor.java | 2 +-
helix-core/src/main/java/org/apache/helix/model/LiveInstance.java | 2 +-
helix-core/src/main/java/org/apache/helix/util/HelixUtil.java | 2 +-
.../apache/helix/controller/stages/TestManagementModeStage.java | 2 +-
.../helix/integration/controller/TestClusterFreezeMode.java | 6 +++---
.../helix/integration/paticipant/TestParticipantFreeze.java | 8 ++++----
.../src/test/java/org/apache/helix/model/TestLiveInstance.java | 4 ++--
.../apache/helix/rest/server/resources/helix/ClusterAccessor.java | 4 ++--
12 files changed, 20 insertions(+), 20 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
index a342a8c..daf121c 100644
--- a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
+++ b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
@@ -89,7 +89,7 @@ public class ClusterManagementMode {
public LiveInstance.LiveInstanceStatus getDesiredParticipantStatus() {
switch (mode) {
case CLUSTER_FREEZE:
- return LiveInstance.LiveInstanceStatus.PAUSED;
+ return LiveInstance.LiveInstanceStatus.FROZEN;
case NORMAL:
return LiveInstance.LiveInstanceStatus.NORMAL;
default:
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
index 55a320a..a281e08 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
@@ -119,7 +119,7 @@ public class ManagementModeStage extends AbstractBaseStage {
// 1. All live instances are frozen
// 2. No pending participant status change message.
return enabledLiveInstances.stream().noneMatch(
- instance -> !LiveInstanceStatus.PAUSED.equals(liveInstanceMap.get(instance).getStatus())
+ instance -> !LiveInstanceStatus.FROZEN.equals(liveInstanceMap.get(instance).getStatus())
|| hasPendingMessage(allInstanceMessages.get(instance),
MessageType.PARTICIPANT_STATUS_CHANGE));
}
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
index b7a0014..09d994f 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
@@ -178,7 +178,7 @@ public class ParticipantManager {
String status = _liveInstanceInfoProvider.getAdditionalLiveInstanceInfo()
.getSimpleField(LiveInstance.LiveInstanceProperty.STATUS.name());
// If frozen, no carry-over
- return !LiveInstance.LiveInstanceStatus.PAUSED.name().equals(status);
+ return !LiveInstance.LiveInstanceStatus.FROZEN.name().equals(status);
}
private void joinCluster() {
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
index 4b88f74..b5f6174 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
@@ -1320,7 +1320,7 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
// Will only create live instance
LiveInstance.LiveInstanceStatus liveInstanceStatus =
_messagingService.getExecutor().getLiveInstanceStatus();
- if (LiveInstance.LiveInstanceStatus.PAUSED
+ if (LiveInstance.LiveInstanceStatus.FROZEN
.equals(liveInstanceStatus)) {
handleNewSessionInManagementMode(sessionId, liveInstanceStatus);
return;
@@ -1387,7 +1387,7 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
private void handleNewSessionInManagementMode(String sessionId,
LiveInstance.LiveInstanceStatus liveInstanceStatus) throws Exception {
- LOG.info("Skip reset because instance is in {} status", LiveInstance.LiveInstanceStatus.PAUSED);
+ LOG.info("Skip reset because instance is in {} status", LiveInstance.LiveInstanceStatus.FROZEN);
if (!InstanceType.PARTICIPANT.equals(_instanceType)
&& !InstanceType.CONTROLLER_PARTICIPANT.equals(_instanceType)) {
return;
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
index 11badac..b981a98 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
@@ -1352,7 +1352,7 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
boolean success = false;
switch (toStatus) {
- case PAUSED:
+ case FROZEN:
_freezeSessionId = sessionId;
_liveInstanceStatus = toStatus;
// Entering freeze mode, update live instance status.
diff --git a/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java b/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
index e711fea..ba9cc00 100644
--- a/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
+++ b/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
@@ -50,7 +50,7 @@ public class LiveInstance extends HelixProperty {
* Saved values for the {@link LiveInstanceProperty#STATUS} field
*/
public enum LiveInstanceStatus {
- PAUSED,
+ FROZEN,
NORMAL
}
diff --git a/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java b/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java
index c770583..75ca302 100644
--- a/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java
@@ -576,7 +576,7 @@ public final class HelixUtil {
Map<String, LiveInstance> liveInstanceMap,
Map<String, Collection<Message>> instancesMessages) {
// Check live instance status and participant status change message
- return LiveInstance.LiveInstanceStatus.PAUSED.equals(liveInstanceMap.get(instance).getStatus())
+ return LiveInstance.LiveInstanceStatus.FROZEN.equals(liveInstanceMap.get(instance).getStatus())
|| (instancesMessages.getOrDefault(instance, Collections.emptyList()).stream()
.anyMatch(Message::isParticipantStatusChangeType));
}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
index 1be5c6b..11a0c60 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
@@ -96,7 +96,7 @@ public class TestManagementModeStage extends ZkTestBase {
// Mark a live instance to be pause state
LiveInstance liveInstance = liveInstances.get(0);
- liveInstance.setStatus(LiveInstance.LiveInstanceStatus.PAUSED);
+ liveInstance.setStatus(LiveInstance.LiveInstanceStatus.FROZEN);
PropertyKey liveInstanceKey =
_accessor.keyBuilder().liveInstance(liveInstance.getInstanceName());
_accessor.updateProperty(liveInstanceKey, liveInstance);
diff --git a/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java b/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
index afaea91..09de4dc 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
@@ -172,7 +172,7 @@ public class TestClusterFreezeMode extends ZkTestBase {
latch.countDown();
// Verify live instance status and cluster status
- verifyLiveInstanceStatus(_participants, LiveInstance.LiveInstanceStatus.PAUSED);
+ verifyLiveInstanceStatus(_participants, LiveInstance.LiveInstanceStatus.FROZEN);
expectedClusterStatus = new ClusterStatus();
expectedClusterStatus.setManagementMode(ClusterManagementMode.Type.CLUSTER_FREEZE);
@@ -204,7 +204,7 @@ public class TestClusterFreezeMode extends ZkTestBase {
// The new participant/live instance should be frozen by controller
verifyLiveInstanceStatus(new MockParticipantManager[]{newParticipant},
- LiveInstance.LiveInstanceStatus.PAUSED);
+ LiveInstance.LiveInstanceStatus.FROZEN);
newParticipant.syncStop();
}
@@ -236,7 +236,7 @@ public class TestClusterFreezeMode extends ZkTestBase {
Assert.assertEquals(liveInstance.getEphemeralOwner(), _participants[1].getSessionId());
// Status is frozen because controller sends a freeze message.
verifyLiveInstanceStatus(new MockParticipantManager[]{_participants[1]},
- LiveInstance.LiveInstanceStatus.PAUSED);
+ LiveInstance.LiveInstanceStatus.FROZEN);
// Old session current state is deleted because of current state carry-over
Assert.assertTrue(TestHelper.verify(
diff --git a/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java b/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java
index 1eb760e..334ebb1 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java
@@ -199,7 +199,7 @@ public class TestParticipantFreeze extends ZkTestBase {
// New live instance ephemeral node with FROZEN status
Assert.assertFalse(_originSession.equals(liveInstance.getEphemeralOwner()));
- Assert.assertEquals(liveInstance.getStatus(), LiveInstance.LiveInstanceStatus.PAUSED);
+ Assert.assertEquals(liveInstance.getStatus(), LiveInstance.LiveInstanceStatus.FROZEN);
// New session path does not exist since no current state carry over for the current session.
Assert.assertFalse(
@@ -212,7 +212,7 @@ public class TestParticipantFreeze extends ZkTestBase {
@Test(dependsOnMethods = "testHandleNewSessionWhenFrozen")
public void testUnfreezeParticipant() throws Exception {
Message unfreezeMessage = MessageUtil
- .createStatusChangeMessage(LiveInstance.LiveInstanceStatus.PAUSED,
+ .createStatusChangeMessage(LiveInstance.LiveInstanceStatus.FROZEN,
LiveInstance.LiveInstanceStatus.NORMAL, _manager.getInstanceName(),
_manager.getSessionId(), _instanceName, _participants[0].getSessionId());
List<PropertyKey> keys = Collections
@@ -275,7 +275,7 @@ public class TestParticipantFreeze extends ZkTestBase {
private void freezeParticipant(MockParticipantManager participant) throws Exception {
Message freezeMessage = MessageUtil
.createStatusChangeMessage(LiveInstance.LiveInstanceStatus.NORMAL,
- LiveInstance.LiveInstanceStatus.PAUSED, _manager.getInstanceName(),
+ LiveInstance.LiveInstanceStatus.FROZEN, _manager.getInstanceName(),
_manager.getSessionId(), participant.getInstanceName(), participant.getSessionId());
List<PropertyKey> keys = Collections
@@ -285,7 +285,7 @@ public class TestParticipantFreeze extends ZkTestBase {
Assert.assertTrue(success[0]);
// Live instance status is frozen in both memory and zk
- verifyLiveInstanceStatus(participant, LiveInstance.LiveInstanceStatus.PAUSED);
+ verifyLiveInstanceStatus(participant, LiveInstance.LiveInstanceStatus.FROZEN);
// Freeze message is correctly deleted
Assert.assertTrue(TestHelper.verify(() -> !_gZkClient.exists(
_keyBuilder.message(participant.getInstanceName(), freezeMessage.getId()).getPath()),
diff --git a/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java b/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java
index b8024bb..84d4b45 100644
--- a/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java
+++ b/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java
@@ -153,7 +153,7 @@ public class TestLiveInstance extends ZkUnitTestBase {
Assert.assertNull(testLiveInstance.getRecord()
.getSimpleField(LiveInstance.LiveInstanceProperty.STATUS.name()));
Assert.assertEquals(testLiveInstance.getStatus(), LiveInstance.LiveInstanceStatus.NORMAL);
- testLiveInstance.setStatus(LiveInstance.LiveInstanceStatus.PAUSED);
- Assert.assertEquals(testLiveInstance.getStatus(), LiveInstance.LiveInstanceStatus.PAUSED);
+ testLiveInstance.setStatus(LiveInstance.LiveInstanceStatus.FROZEN);
+ Assert.assertEquals(testLiveInstance.getStatus(), LiveInstance.LiveInstanceStatus.FROZEN);
}
}
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 552a530..cf2457c 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
@@ -348,7 +348,7 @@ public class ClusterAccessor extends AbstractHelixResource {
// Entering cluster freeze mode, check live instance freeze status and pending ST
for (LiveInstance liveInstance : liveInstances) {
String instanceName = liveInstance.getInstanceName();
- if (!LiveInstance.LiveInstanceStatus.PAUSED.equals(liveInstance.getStatus())) {
+ if (!LiveInstance.LiveInstanceStatus.FROZEN.equals(liveInstance.getStatus())) {
liveInstancesInProgress.add(instanceName);
}
Stat stat = baseAccessor
@@ -361,7 +361,7 @@ public class ClusterAccessor extends AbstractHelixResource {
} else if (ClusterManagementMode.Type.NORMAL.equals(mode.getMode())) {
// Exiting freeze mode, check live instance unfreeze status
for (LiveInstance liveInstance : liveInstances) {
- if (LiveInstance.LiveInstanceStatus.PAUSED.equals(liveInstance.getStatus())) {
+ if (LiveInstance.LiveInstanceStatus.FROZEN.equals(liveInstance.getStatus())) {
liveInstancesInProgress.add(liveInstance.getInstanceName());
}
}
[helix] 04/13: Move pause and maintenance handling out of
controller (#1793)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit f9fba2ac59ecdd6baeaeb005c4d0e428ab76c3c6
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Sat Jun 12 01:00:25 2021 -0700
Move pause and maintenance handling out of controller (#1793)
With management mode pipeline, the pause and maintenance signals handling logic should be moved out of the onControllerChange() and moved to the management mode pipeline.
This commit handles pause/maintenance signals enable/disable and update cluster status accordingly.
---
.../helix/controller/GenericHelixController.java | 80 +++-------------------
.../dataproviders/BaseControllerDataProvider.java | 21 +++++-
.../ManagementControllerDataProvider.java | 24 ++++++-
.../controller/stages/ManagementModeStage.java | 16 +++--
.../controller/stages/ResourceValidationStage.java | 29 ++++++--
.../main/java/org/apache/helix/model/Message.java | 5 ++
.../BestPossibleExternalViewVerifier.java | 2 +-
.../main/java/org/apache/helix/util/HelixUtil.java | 35 ++++++++--
.../java/org/apache/helix/util/RebalanceUtil.java | 4 +-
9 files changed, 119 insertions(+), 97 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
index 7da6ef0..6cc1e5f 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
@@ -103,9 +103,7 @@ import org.apache.helix.model.CustomizedStateConfig;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
-import org.apache.helix.model.MaintenanceSignal;
import org.apache.helix.model.Message;
-import org.apache.helix.model.PauseSignal;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.monitoring.mbeans.ClusterEventMonitor;
import org.apache.helix.monitoring.mbeans.ClusterStatusMonitor;
@@ -177,13 +175,6 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
private long _continuousTaskRebalanceFailureCount = 0;
/**
- * The _paused flag is checked by function handleEvent(), while if the flag is set handleEvent()
- * will be no-op. Other event handling logic keeps the same when the flag is set.
- */
- private boolean _paused;
- private boolean _inMaintenanceMode;
-
- /**
* The executors that can periodically run the rebalancing pipeline. A
* SingleThreadScheduledExecutor will start if there is resource group that has the config to do
* periodically rebalance.
@@ -837,18 +828,16 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
// have this instanceof clauses
List<Pipeline> pipelines;
boolean isTaskFrameworkPipeline = false;
- Pipeline.Type pipelineType;
+ boolean isManagementPipeline = false;
if (dataProvider instanceof ResourceControllerDataProvider) {
pipelines = _registry.getPipelinesForEvent(event.getEventType());
- pipelineType = Pipeline.Type.DEFAULT;
} else if (dataProvider instanceof WorkflowControllerDataProvider) {
pipelines = _taskRegistry.getPipelinesForEvent(event.getEventType());
isTaskFrameworkPipeline = true;
- pipelineType = Pipeline.Type.TASK;
} else if (dataProvider instanceof ManagementControllerDataProvider) {
pipelines = _managementModeRegistry.getPipelinesForEvent(event.getEventType());
- pipelineType = Pipeline.Type.MANAGEMENT_MODE;
+ isManagementPipeline = true;
} else {
logger.warn(String
.format("No %s pipeline to run for event: %s::%s", dataProvider.getPipelineName(),
@@ -857,11 +846,10 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
}
// Should not run management mode and default/task pipelines at the same time.
- if ((_inManagementMode && !Pipeline.Type.MANAGEMENT_MODE.equals(pipelineType))
- || (!_inManagementMode && Pipeline.Type.MANAGEMENT_MODE.equals(pipelineType))) {
+ if (_inManagementMode != isManagementPipeline) {
logger.info("Should not run management mode and default/task pipelines at the same time. "
- + "cluster={}, inManagementMode={}, pipelineType={}. Ignoring the event: {}",
- manager.getClusterName(), _inManagementMode, pipelineType, event.getEventType());
+ + "cluster={}, inManagementMode={}, isManagementPipeline={}. Ignoring the event: {}",
+ manager.getClusterName(), _inManagementMode, isManagementPipeline, event.getEventType());
return;
}
@@ -881,6 +869,8 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
checkRebalancingTimer(manager, Collections.<IdealState>emptyList(), dataProvider.getClusterConfig());
}
if (_isMonitoring) {
+ _clusterStatusMonitor.setEnabled(!_inManagementMode);
+ _clusterStatusMonitor.setPaused(_inManagementMode);
event.addAttribute(AttributeName.clusterStatusMonitor.name(), _clusterStatusMonitor);
}
}
@@ -1335,25 +1325,8 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
}
if (controllerIsLeader) {
- HelixManager manager = changeContext.getManager();
- HelixDataAccessor accessor = manager.getHelixDataAccessor();
- Builder keyBuilder = accessor.keyBuilder();
-
- PauseSignal pauseSignal = accessor.getProperty(keyBuilder.pause());
- MaintenanceSignal maintenanceSignal = accessor.getProperty(keyBuilder.maintenance());
- boolean prevPaused = _paused;
- boolean prevInMaintenanceMode = _inMaintenanceMode;
- _paused = updateControllerState(pauseSignal, _paused);
- _inMaintenanceMode = updateControllerState(maintenanceSignal, _inMaintenanceMode);
- // TODO: remove triggerResumeEvent when moving pause/maintenance to management pipeline
- if (!triggerResumeEvent(changeContext, prevPaused, prevInMaintenanceMode)) {
- pushToEventQueues(ClusterEventType.ControllerChange, changeContext, Collections.emptyMap());
- }
-
enableClusterStatusMonitor(true);
- _clusterStatusMonitor.setEnabled(!_paused);
- _clusterStatusMonitor.setPaused(_paused);
- _clusterStatusMonitor.setMaintenance(_inMaintenanceMode);
+ pushToEventQueues(ClusterEventType.ControllerChange, changeContext, Collections.emptyMap());
} else {
enableClusterStatusMonitor(false);
// Note that onControllerChange is executed in parallel with the event processing thread. It
@@ -1543,43 +1516,6 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
}
}
- private boolean updateControllerState(PauseSignal signal, boolean statusFlag) {
- if (signal != null) {
- if (!statusFlag) {
- statusFlag = true;
- // This log is recorded for the first time entering PAUSE/MAINTENANCE mode
- logger.info(String.format("controller is now %s",
- (signal instanceof MaintenanceSignal) ? "in maintenance mode" : "paused"));
- }
- } else {
- statusFlag = false;
- }
- return statusFlag;
- }
-
- /**
- * Trigger a Resume Event if the cluster is back to activated.
- * @param changeContext
- * @param prevPaused the previous paused status.
- * @param prevInMaintenanceMode the previous in maintenance mode status.
- */
- private boolean triggerResumeEvent(NotificationContext changeContext, boolean prevPaused,
- boolean prevInMaintenanceMode) {
- /**
- * WARNING: the logic here is tricky.
- * 1. Only resume if not paused. So if the Maintenance mode is removed but the cluster is still
- * paused, the resume event should not be sent.
- * 2. Only send resume event if the status is changed back to active. So we don't send multiple
- * event unnecessarily.
- */
- if (!_paused && (prevPaused || (prevInMaintenanceMode && !_inMaintenanceMode))) {
- pushToEventQueues(ClusterEventType.Resume, changeContext, Collections.EMPTY_MAP);
- logger.info("controller is now resumed from paused/maintenance state");
- return true;
- }
- return false;
- }
-
// TODO: refactor this to use common/ClusterEventProcessor.
@Deprecated
private class ClusterEventProcessor extends Thread {
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 6ce25e2..3c705f4 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
@@ -57,6 +57,7 @@ import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.MaintenanceSignal;
import org.apache.helix.model.Message;
import org.apache.helix.model.ParticipantHistory;
+import org.apache.helix.model.PauseSignal;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.task.TaskConstants;
@@ -87,6 +88,7 @@ public class BaseControllerDataProvider implements ControlContextProvider {
private boolean _updateInstanceOfflineTime = true;
private MaintenanceSignal _maintenanceSignal;
+ private PauseSignal _pauseSignal;
private boolean _isMaintenanceModeEnabled;
private boolean _hasMaintenanceSignalChanged;
private ExecutorService _asyncTasksThreadPool;
@@ -300,8 +302,9 @@ public class BaseControllerDataProvider implements ControlContextProvider {
}
}
- private void updateMaintenanceInfo(final HelixDataAccessor accessor) {
+ private void refreshManagementSignals(final HelixDataAccessor accessor) {
_maintenanceSignal = accessor.getProperty(accessor.keyBuilder().maintenance());
+ _pauseSignal = accessor.getProperty(accessor.keyBuilder().pause());
_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
@@ -373,7 +376,7 @@ public class BaseControllerDataProvider implements ControlContextProvider {
refreshResourceConfig(accessor, refreshedTypes);
_stateModelDefinitionCache.refresh(accessor);
_clusterConstraintsCache.refresh(accessor);
- updateMaintenanceInfo(accessor);
+ refreshManagementSignals(accessor);
timeoutNodesDuringMaintenance(accessor, _clusterConfig, _isMaintenanceModeEnabled);
// TODO: once controller gets split, only one controller should update offline instance history
@@ -622,6 +625,16 @@ public class BaseControllerDataProvider implements ControlContextProvider {
}
/**
+ * Gets all messages for each instance.
+ *
+ * @return Map of {instanceName -> Collection of Message}.
+ */
+ public Map<String, Collection<Message>> getAllInstancesMessages() {
+ return getAllInstances().stream().collect(
+ Collectors.toMap(instance -> instance, instance -> getMessages(instance).values()));
+ }
+
+ /**
* This function is supposed to be only used by testing purpose for safety. For "get" usage,
* please use getStaleMessagesByInstance.
*/
@@ -968,6 +981,10 @@ public class BaseControllerDataProvider implements ControlContextProvider {
return _maintenanceSignal;
}
+ public PauseSignal getPauseSignal() {
+ return _pauseSignal;
+ }
+
protected StringBuilder genCacheContentStringBuilder() {
StringBuilder sb = new StringBuilder();
sb.append(String.format("liveInstaceMap: %s", _liveInstanceCache.getPropertyMap())).append("\n");
diff --git a/helix-core/src/main/java/org/apache/helix/controller/dataproviders/ManagementControllerDataProvider.java b/helix-core/src/main/java/org/apache/helix/controller/dataproviders/ManagementControllerDataProvider.java
index d178ca5..fe940ff 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/dataproviders/ManagementControllerDataProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/dataproviders/ManagementControllerDataProvider.java
@@ -19,9 +19,27 @@ package org.apache.helix.controller.dataproviders;
* under the License.
*/
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.helix.HelixConstants;
+
+/**
+ * Data provider for controller management mode pipeline.
+ */
public class ManagementControllerDataProvider extends BaseControllerDataProvider {
- // TODO: implement this class to only refresh required event types
- public ManagementControllerDataProvider(String clusterName, String name) {
- super(clusterName, name);
+ // Only these types of properties are refreshed for the full refresh request.
+ private static final List<HelixConstants.ChangeType> FULL_REFRESH_PROPERTIES =
+ Arrays.asList(HelixConstants.ChangeType.LIVE_INSTANCE, HelixConstants.ChangeType.MESSAGE);
+
+ public ManagementControllerDataProvider(String clusterName, String pipelineName) {
+ super(clusterName, pipelineName);
+ }
+
+ @Override
+ public void requireFullRefresh() {
+ for (HelixConstants.ChangeType type : FULL_REFRESH_PROPERTIES) {
+ _propertyDataChangedMap.get(type).set(true);
+ }
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
index 512224d..042aa14 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
@@ -19,9 +19,9 @@ package org.apache.helix.controller.stages;
* under the License.
*/
+import org.apache.helix.controller.LogUtil;
import org.apache.helix.controller.dataproviders.ManagementControllerDataProvider;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
-import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.util.HelixUtil;
import org.apache.helix.util.RebalanceUtil;
import org.slf4j.Logger;
@@ -36,13 +36,21 @@ public class ManagementModeStage extends AbstractBaseStage {
@Override
public void process(ClusterEvent event) throws Exception {
// TODO: implement the stage
+ _eventId = event.getEventId();
String clusterName = event.getClusterName();
ManagementControllerDataProvider cache =
event.getAttribute(AttributeName.ControllerDataProvider.name());
- if (!HelixUtil.inManagementMode(cache)) {
- LOG.info("Exiting management mode pipeline for cluster {}", clusterName);
+
+ // TODO: move to the last stage of management pipeline
+ checkInManagementMode(clusterName, cache);
+ }
+
+ private void checkInManagementMode(String clusterName, ManagementControllerDataProvider cache) {
+ // Should exit management mode
+ if (!HelixUtil.inManagementMode(cache.getPauseSignal(), cache.getLiveInstances(),
+ cache.getEnabledLiveInstances(), cache.getAllInstancesMessages())) {
+ LogUtil.logInfo(LOG, _eventId, "Exiting management mode pipeline for cluster " + clusterName);
RebalanceUtil.enableManagementMode(clusterName, false);
- throw new StageException("Exiting management mode pipeline for cluster " + clusterName);
}
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java
index a4d4783..613ce2e 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java
@@ -28,6 +28,7 @@ import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.Resource;
import org.apache.helix.model.StateModelDefinition;
+import org.apache.helix.monitoring.mbeans.ClusterStatusMonitor;
import org.apache.helix.util.HelixUtil;
import org.apache.helix.util.RebalanceUtil;
import org.slf4j.Logger;
@@ -44,12 +45,7 @@ public class ResourceValidationStage extends AbstractBaseStage {
throw new StageException("Missing attributes in event:" + event + ". Requires DataCache");
}
- // Check if cluster is still in management mode. Eg. there exists any frozen live instance.
- if (HelixUtil.inManagementMode(cache)) {
- // Trigger an immediate management mode pipeline.
- RebalanceUtil.enableManagementMode(event.getClusterName(), true);
- throw new StageException("Pipeline should not be run because cluster is in management mode");
- }
+ processManagementMode(event, cache);
Map<String, Resource> resourceMap = event.getAttribute(AttributeName.RESOURCES.name());
if (resourceMap == null) {
@@ -91,6 +87,27 @@ public class ResourceValidationStage extends AbstractBaseStage {
}
}
+ private void processManagementMode(ClusterEvent event, BaseControllerDataProvider cache)
+ throws StageException {
+ // Set cluster status monitor for maintenance mode
+ ClusterStatusMonitor monitor = event.getAttribute(AttributeName.clusterStatusMonitor.name());
+ if (monitor != null) {
+ monitor.setMaintenance(cache.isMaintenanceModeEnabled());
+ }
+
+ // Check if cluster is still in management mode. Eg. there exists any frozen live instance.
+ if (HelixUtil.inManagementMode(cache.getPauseSignal(), cache.getLiveInstances(),
+ cache.getEnabledLiveInstances(), cache.getAllInstancesMessages())) {
+ // Trigger an immediate management mode pipeline.
+ LogUtil.logInfo(LOG, _eventId,
+ "Enabling management mode pipeline for cluster " + event.getClusterName());
+ RebalanceUtil.enableManagementMode(event.getClusterName(), true);
+ throw new StageException(
+ "Pipeline should not be run because cluster " + event.getClusterName()
+ + "is in management mode");
+ }
+ }
+
/**
* Check if the ideal state adheres to a rule
* @param idealState the ideal state to check
diff --git a/helix-core/src/main/java/org/apache/helix/model/Message.java b/helix-core/src/main/java/org/apache/helix/model/Message.java
index b509506..6222df5 100644
--- a/helix-core/src/main/java/org/apache/helix/model/Message.java
+++ b/helix-core/src/main/java/org/apache/helix/model/Message.java
@@ -59,6 +59,7 @@ public class Message extends HelixProperty {
NO_OP,
PARTICIPANT_ERROR_REPORT,
PARTICIPANT_SESSION_CHANGE,
+ PARTICIPANT_STATUS_CHANGE,
CHAINED_MESSAGE, // this is a message subtype
RELAYED_MESSAGE
}
@@ -927,6 +928,10 @@ public class Message extends HelixProperty {
return getTgtName().equalsIgnoreCase(InstanceType.CONTROLLER.name());
}
+ public boolean isParticipantStatusChangeType() {
+ return MessageType.PARTICIPANT_STATUS_CHANGE.name().equalsIgnoreCase(getMsgType());
+ }
+
/**
* Get the {@link PropertyKey} for this message
* @param keyBuilder PropertyKey Builder
diff --git a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java
index eaab1a2..3b133a1 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java
@@ -416,7 +416,7 @@ public class BestPossibleExternalViewVerifier extends ZkHelixClusterVerifier {
*/
private BestPossibleStateOutput calcBestPossState(ResourceControllerDataProvider cache, Set<String> resources)
throws Exception {
- ClusterEvent event = new ClusterEvent(ClusterEventType.StateVerifier);
+ ClusterEvent event = new ClusterEvent(_clusterName, ClusterEventType.StateVerifier);
event.addAttribute(AttributeName.ControllerDataProvider.name(), cache);
RebalanceUtil.runStage(event, new ResourceComputationStage());
diff --git a/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java b/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java
index 3151716..c770583 100644
--- a/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java
@@ -21,6 +21,7 @@ package org.apache.helix.util;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -38,7 +39,6 @@ import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.PropertyType;
import org.apache.helix.controller.common.PartitionStateMap;
-import org.apache.helix.controller.dataproviders.BaseControllerDataProvider;
import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
import org.apache.helix.controller.rebalancer.AbstractRebalancer;
import org.apache.helix.controller.rebalancer.strategy.RebalanceStrategy;
@@ -60,6 +60,7 @@ import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.Message;
import org.apache.helix.model.Partition;
+import org.apache.helix.model.PauseSignal;
import org.apache.helix.model.ResourceAssignment;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
@@ -550,13 +551,33 @@ public final class HelixUtil {
}
/**
- * Checks whether or not the cluster is in management mode.
+ * Checks whether or not the cluster is in management mode. It checks:
+ * - pause signal
+ * - live instances: whether any live instance is not in normal status, eg. frozen.
+ * - messages: whether live instance has a participant status change message
*
- * @param cache
- * @return
+ * @param pauseSignal pause signal
+ * @param liveInstanceMap map of live instances
+ * @param enabledLiveInstances set of enabled live instance names. They should be all included
+ * in the liveInstanceMap.
+ * @param instancesMessages a map of all instances' messages.
+ * @return true if cluster is in management mode; otherwise, false
*/
- public static boolean inManagementMode(BaseControllerDataProvider cache) {
- // TODO: implement the logic. Parameters can also change
- return true;
+ public static boolean inManagementMode(PauseSignal pauseSignal,
+ Map<String, LiveInstance> liveInstanceMap, Set<String> enabledLiveInstances,
+ Map<String, Collection<Message>> instancesMessages) {
+ // Check pause signal and abnormal live instances (eg. in freeze mode)
+ // TODO: should check maintenance signal when moving maintenance to management pipeline
+ return pauseSignal != null || enabledLiveInstances.stream().anyMatch(
+ instance -> isInstanceInManagementMode(instance, liveInstanceMap, instancesMessages));
+ }
+
+ private static boolean isInstanceInManagementMode(String instance,
+ Map<String, LiveInstance> liveInstanceMap,
+ Map<String, Collection<Message>> instancesMessages) {
+ // Check live instance status and participant status change message
+ return LiveInstance.LiveInstanceStatus.PAUSED.equals(liveInstanceMap.get(instance).getStatus())
+ || (instancesMessages.getOrDefault(instance, Collections.emptyList()).stream()
+ .anyMatch(Message::isParticipantStatusChangeType));
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java b/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java
index db2b76f..f74b98f 100644
--- a/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java
@@ -159,8 +159,8 @@ public class RebalanceUtil {
enabled);
leaderController.setInManagementMode(enabled);
} else {
- LOG.error("Failed to switch management mode pipeline, enabled={}. "
- + "Controller for cluster {} does not exist", clusterName, enabled);
+ throw new HelixException(String.format("Failed to switch management mode pipeline, "
+ + "enabled=%s. Controller for cluster %s does not exist", enabled, clusterName));
}
// Triggers an event to immediately run the pipeline
[helix] 02/13: Add model to record history and status of management
mode (#1771)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 88dae34522e3dc75497ea6d8cb5c19b288ee3212
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Wed Jun 9 14:54:14 2021 -0700
Add model to record history and status of management mode (#1771)
Management mode operation history needs to be persisted to the controller history znode. The status of IN_PROGRESS or COMPLETED is recorded in the temporary status znode: /{clusterName}/STATUS/CLUSTER/{clusterName}.
This commit adds data model and methods to record the status and history for management mode.
---
.../main/java/org/apache/helix/PropertyKey.java | 12 ++-
.../java/org/apache/helix/PropertyPathBuilder.java | 7 ++
.../main/java/org/apache/helix/PropertyType.java | 1 +
.../java/org/apache/helix/model/ClusterStatus.java | 78 ++++++++++++++
.../org/apache/helix/model/ControllerHistory.java | 112 +++++++++++++++------
.../org/apache/helix/TestPropertyPathBuilder.java | 3 +
.../helix/model/TestControllerHistoryModel.java | 93 +++++++++++++++++
7 files changed, 276 insertions(+), 30 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/PropertyKey.java b/helix-core/src/main/java/org/apache/helix/PropertyKey.java
index 2cf1168..254cb95 100644
--- a/helix-core/src/main/java/org/apache/helix/PropertyKey.java
+++ b/helix-core/src/main/java/org/apache/helix/PropertyKey.java
@@ -25,10 +25,11 @@ import java.util.Objects;
import org.apache.helix.model.CloudConfig;
import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.ClusterConstraints;
+import org.apache.helix.model.ClusterStatus;
import org.apache.helix.model.ControllerHistory;
import org.apache.helix.model.CurrentState;
-import org.apache.helix.model.CustomizedStateConfig;
import org.apache.helix.model.CustomizedState;
+import org.apache.helix.model.CustomizedStateConfig;
import org.apache.helix.model.CustomizedView;
import org.apache.helix.model.Error;
import org.apache.helix.model.ExternalView;
@@ -56,10 +57,10 @@ import static org.apache.helix.PropertyType.CONFIGS;
import static org.apache.helix.PropertyType.CONTROLLER;
import static org.apache.helix.PropertyType.CURRENTSTATES;
import static org.apache.helix.PropertyType.CUSTOMIZEDSTATES;
+import static org.apache.helix.PropertyType.CUSTOMIZEDVIEW;
import static org.apache.helix.PropertyType.ERRORS;
import static org.apache.helix.PropertyType.ERRORS_CONTROLLER;
import static org.apache.helix.PropertyType.EXTERNALVIEW;
-import static org.apache.helix.PropertyType.CUSTOMIZEDVIEW;
import static org.apache.helix.PropertyType.HISTORY;
import static org.apache.helix.PropertyType.IDEALSTATES;
import static org.apache.helix.PropertyType.INSTANCE_HISTORY;
@@ -241,6 +242,13 @@ public class PropertyKey {
_clusterName, ConfigScopeProperty.CLUSTER.toString(), _clusterName);
}
+ /**
+ * Get a property key associated with this cluster status
+ * @return {@link PropertyKey}
+ */
+ public PropertyKey clusterStatus() {
+ return new PropertyKey(PropertyType.STATUS, ClusterStatus.class, _clusterName);
+ }
/**
* Get a property key associated with this Cloud configuration
diff --git a/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java b/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java
index 2ba1ebd..34efd29 100644
--- a/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java
+++ b/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java
@@ -25,6 +25,7 @@ import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import org.apache.helix.model.ClusterStatus;
import org.apache.helix.model.ControllerHistory;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.CustomizedView;
@@ -66,6 +67,7 @@ public class PropertyPathBuilder {
typeToClassMapping.put(PropertyType.HISTORY, ControllerHistory.class);
typeToClassMapping.put(PropertyType.PAUSE, PauseSignal.class);
typeToClassMapping.put(PropertyType.MAINTENANCE, MaintenanceSignal.class);
+ typeToClassMapping.put(PropertyType.STATUS, ClusterStatus.class);
// TODO: Below must handle the case for future versions of Task Framework with a different path
// structure
typeToClassMapping.put(PropertyType.WORKFLOWCONTEXT, WorkflowContext.class);
@@ -86,6 +88,7 @@ public class PropertyPathBuilder {
addEntry(PropertyType.CUSTOMIZEDVIEW, 1, "/{clusterName}/CUSTOMIZEDVIEW");
addEntry(PropertyType.CUSTOMIZEDVIEW, 2, "/{clusterName}/CUSTOMIZEDVIEW/{customizedStateType}");
addEntry(PropertyType.CUSTOMIZEDVIEW, 3, "/{clusterName}/CUSTOMIZEDVIEW/{customizedStateType}/{resourceName}");
+ addEntry(PropertyType.STATUS, 1, "/{clusterName}/STATUS");
addEntry(PropertyType.TARGETEXTERNALVIEW, 1, "/{clusterName}/TARGETEXTERNALVIEW");
addEntry(PropertyType.TARGETEXTERNALVIEW, 2,
@@ -452,4 +455,8 @@ public class PropertyPathBuilder {
public static String maintenance(String clusterName) {
return String.format("/%s/CONTROLLER/MAINTENANCE", clusterName);
}
+
+ public static String clusterStatus(String clusterName) {
+ return String.format("/%s/STATUS/CLUSTER/%s", clusterName, clusterName);
+ }
}
diff --git a/helix-core/src/main/java/org/apache/helix/PropertyType.java b/helix-core/src/main/java/org/apache/helix/PropertyType.java
index bedf79e..474ea05 100644
--- a/helix-core/src/main/java/org/apache/helix/PropertyType.java
+++ b/helix-core/src/main/java/org/apache/helix/PropertyType.java
@@ -48,6 +48,7 @@ public enum PropertyType {
STATEMODELDEFS(Type.CLUSTER, true, false, false, false, true),
CONTROLLER(Type.CLUSTER, true, false),
PROPERTYSTORE(Type.CLUSTER, true, false),
+ STATUS(Type.CLUSTER, true, false, true),
// INSTANCE PROPERTIES
MESSAGES(Type.INSTANCE, true, true, true),
diff --git a/helix-core/src/main/java/org/apache/helix/model/ClusterStatus.java b/helix-core/src/main/java/org/apache/helix/model/ClusterStatus.java
new file mode 100644
index 0000000..6ed354c
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/model/ClusterStatus.java
@@ -0,0 +1,78 @@
+package org.apache.helix.model;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.helix.HelixProperty;
+import org.apache.helix.PropertyType;
+import org.apache.helix.api.status.ClusterManagementMode;
+
+/**
+ * Represents the cluster status. It can have fields for
+ * {@link ClusterManagementMode} type and status.
+ */
+public class ClusterStatus extends HelixProperty {
+ public ClusterStatus() {
+ super(PropertyType.STATUS.name());
+ }
+
+ public enum ClusterStatusProperty {
+ MANAGEMENT_MODE,
+ MANAGEMENT_MODE_STATUS
+ }
+
+ /**
+ * Sets the type of management mode
+ *
+ * @param mode {@link ClusterManagementMode.Type}
+ */
+ public void setManagementMode(ClusterManagementMode.Type mode) {
+ _record.setEnumField(ClusterStatusProperty.MANAGEMENT_MODE.name(), mode);
+ }
+
+ /**
+ * Gets the type of management mode
+ *
+ * @return {@link ClusterManagementMode.Type}
+ */
+ public ClusterManagementMode.Type getManagementMode() {
+ return _record.getEnumField(ClusterStatusProperty.MANAGEMENT_MODE.name(),
+ ClusterManagementMode.Type.class, null);
+ }
+
+ /**
+ * Sets the cluster management mode status.
+ *
+ * @param status {@link ClusterManagementMode.Status}
+ */
+ public void setManagementModeStatus(ClusterManagementMode.Status status) {
+ _record.setEnumField(ClusterStatusProperty.MANAGEMENT_MODE_STATUS.name(), status);
+ }
+
+ /**
+ * Gets the {@link ClusterManagementMode.Status} of cluster management mode.
+ *
+ * @return {@link ClusterManagementMode.Status} if status is valid; otherwise, return {@code
+ * null}.
+ */
+ public ClusterManagementMode.Status getManagementModeStatus() {
+ return _record.getEnumField(ClusterStatusProperty.MANAGEMENT_MODE_STATUS.name(),
+ ClusterManagementMode.Status.class, null);
+ }
+}
\ No newline at end of file
diff --git a/helix-core/src/main/java/org/apache/helix/model/ControllerHistory.java b/helix-core/src/main/java/org/apache/helix/model/ControllerHistory.java
index cc0da7f..4e418c3 100644
--- a/helix-core/src/main/java/org/apache/helix/model/ControllerHistory.java
+++ b/helix-core/src/main/java/org/apache/helix/model/ControllerHistory.java
@@ -22,7 +22,9 @@ package org.apache.helix.model;
import java.io.IOException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
+import java.time.Instant;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
@@ -30,10 +32,11 @@ import java.util.Map;
import java.util.TimeZone;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.helix.HelixException;
import org.apache.helix.HelixProperty;
+import org.apache.helix.api.status.ClusterManagementMode;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
-
/**
* The history of instances that have served as the leader controller
*/
@@ -57,6 +60,12 @@ public class ControllerHistory extends HelixProperty {
}
+ private enum ManagementModeConfigKey {
+ MANAGEMENT_MODE_HISTORY,
+ MODE,
+ STATUS
+ }
+
private enum OperationType {
// The following are options for OPERATION_TYPE in MaintenanceConfigKey
ENTER,
@@ -65,7 +74,8 @@ public class ControllerHistory extends HelixProperty {
public enum HistoryType {
CONTROLLER_LEADERSHIP,
- MAINTENANCE
+ MAINTENANCE,
+ MANAGEMENT_MODE
}
public ControllerHistory(String id) {
@@ -96,17 +106,6 @@ public class ControllerHistory extends HelixProperty {
list.add(instanceName);
// TODO: remove above in future when we confirmed no one consumes it */
- List<String> historyList = _record.getListField(ConfigProperty.HISTORY.name());
- if (historyList == null) {
- historyList = new ArrayList<>();
- _record.setListField(ConfigProperty.HISTORY.name(), historyList);
- }
-
- // Keep only the last HISTORY_SIZE entries
- while (historyList.size() >= HISTORY_SIZE) {
- historyList.remove(0);
- }
-
Map<String, String> historyEntry = new HashMap<>();
long currentTime = System.currentTimeMillis();
@@ -119,8 +118,7 @@ public class ControllerHistory extends HelixProperty {
historyEntry.put(ConfigProperty.DATE.name(), dateTime);
historyEntry.put(ConfigProperty.VERSION.name(), version);
- historyList.add(historyEntry.toString());
- return _record;
+ return populateHistoryEntries(HistoryType.CONTROLLER_LEADERSHIP, historyEntry.toString());
}
/**
@@ -137,6 +135,40 @@ public class ControllerHistory extends HelixProperty {
}
/**
+ * Gets the management mode history.
+ *
+ * @return List of history strings.
+ */
+ public List<String> getManagementModeHistory() {
+ List<String> history =
+ _record.getListField(ManagementModeConfigKey.MANAGEMENT_MODE_HISTORY.name());
+ return history == null ? Collections.emptyList() : history;
+ }
+
+ /**
+ * Updates management mode and status history to controller history in FIFO order.
+ *
+ * @param controller controller name
+ * @param mode cluster management mode {@link ClusterManagementMode}
+ * @param fromHost the hostname that creates the management mode signal
+ * @param time time in millis
+ * @param reason reason to put the cluster in management mode
+ * @return updated history znrecord
+ */
+ public ZNRecord updateManagementModeHistory(String controller, ClusterManagementMode mode,
+ String fromHost, long time, String reason) {
+ Map<String, String> historyEntry = new HashMap<>();
+ historyEntry.put(ConfigProperty.CONTROLLER.name(), controller);
+ historyEntry.put(ConfigProperty.TIME.name(), Instant.ofEpochMilli(time).toString());
+ historyEntry.put(ManagementModeConfigKey.MODE.name(), mode.getMode().name());
+ historyEntry.put(ManagementModeConfigKey.STATUS.name(), mode.getStatus().name());
+ historyEntry.put(PauseSignal.PauseSignalProperty.FROM_HOST.name(), fromHost);
+ historyEntry.put(PauseSignal.PauseSignalProperty.REASON.name(), reason);
+
+ return populateHistoryEntries(HistoryType.MANAGEMENT_MODE, historyEntry.toString());
+ }
+
+ /**
* Record up to MAINTENANCE_HISTORY_SIZE number of changes to MaintenanceSignal in FIFO order.
* @param enabled
* @param reason
@@ -148,18 +180,6 @@ public class ControllerHistory extends HelixProperty {
public ZNRecord updateMaintenanceHistory(boolean enabled, String reason, long currentTime,
MaintenanceSignal.AutoTriggerReason internalReason, Map<String, String> customFields,
MaintenanceSignal.TriggeringEntity triggeringEntity) throws IOException {
- List<String> maintenanceHistoryList =
- _record.getListField(MaintenanceConfigKey.MAINTENANCE_HISTORY.name());
- if (maintenanceHistoryList == null) {
- maintenanceHistoryList = new ArrayList<>();
- _record.setListField(MaintenanceConfigKey.MAINTENANCE_HISTORY.name(), maintenanceHistoryList);
- }
-
- // Keep only the last MAINTENANCE_HISTORY_SIZE entries
- while (maintenanceHistoryList.size() >= MAINTENANCE_HISTORY_SIZE) {
- maintenanceHistoryList.remove(0);
- }
-
DateFormat df = new SimpleDateFormat("yyyy-MM-dd-HH:" + "mm:ss");
df.setTimeZone(TimeZone.getTimeZone("UTC"));
String dateTime = df.format(new Date(currentTime));
@@ -189,7 +209,43 @@ public class ControllerHistory extends HelixProperty {
}
}
}
- maintenanceHistoryList.add(new ObjectMapper().writeValueAsString(maintenanceEntry));
+
+ return populateHistoryEntries(HistoryType.MAINTENANCE,
+ new ObjectMapper().writeValueAsString(maintenanceEntry));
+ }
+
+ private ZNRecord populateHistoryEntries(HistoryType type, String entry) {
+ String configKey;
+ int historySize;
+ switch (type) {
+ case CONTROLLER_LEADERSHIP:
+ configKey = ConfigProperty.HISTORY.name();
+ historySize = HISTORY_SIZE;
+ break;
+ case MAINTENANCE:
+ configKey = MaintenanceConfigKey.MAINTENANCE_HISTORY.name();
+ historySize = MAINTENANCE_HISTORY_SIZE;
+ break;
+ case MANAGEMENT_MODE:
+ configKey = ManagementModeConfigKey.MANAGEMENT_MODE_HISTORY.name();
+ historySize = HISTORY_SIZE;
+ break;
+ default:
+ throw new HelixException("Unknown history type " + type.name());
+ }
+
+ List<String> historyList = _record.getListField(configKey);
+ if (historyList == null) {
+ historyList = new ArrayList<>();
+ _record.setListField(configKey, historyList);
+ }
+
+ while (historyList.size() >= historySize) {
+ historyList.remove(0);
+ }
+
+ historyList.add(entry);
+
return _record;
}
diff --git a/helix-core/src/test/java/org/apache/helix/TestPropertyPathBuilder.java b/helix-core/src/test/java/org/apache/helix/TestPropertyPathBuilder.java
index 422fb9c..9212568 100644
--- a/helix-core/src/test/java/org/apache/helix/TestPropertyPathBuilder.java
+++ b/helix-core/src/test/java/org/apache/helix/TestPropertyPathBuilder.java
@@ -19,6 +19,7 @@ package org.apache.helix;
* under the License.
*/
+import org.testng.Assert;
import org.testng.AssertJUnit;
import org.testng.annotations.Test;
@@ -56,5 +57,7 @@ public class TestPropertyPathBuilder {
actual = PropertyPathBuilder.controllerMessage("test_cluster");
AssertJUnit.assertEquals(actual, "/test_cluster/CONTROLLER/MESSAGES");
+ actual = PropertyPathBuilder.clusterStatus("test_cluster");
+ Assert.assertEquals(actual, "/test_cluster/STATUS/CLUSTER/test_cluster");
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/model/TestControllerHistoryModel.java b/helix-core/src/test/java/org/apache/helix/model/TestControllerHistoryModel.java
new file mode 100644
index 0000000..2e0bab3
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/model/TestControllerHistoryModel.java
@@ -0,0 +1,93 @@
+package org.apache.helix.model;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.google.common.base.Splitter;
+import org.apache.helix.TestHelper;
+import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.zookeeper.zkclient.NetworkUtil;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class TestControllerHistoryModel {
+ @Test
+ public void testManagementModeHistory() {
+ ControllerHistory controllerHistory = new ControllerHistory("HISTORY");
+ String controller = "controller-0";
+ ClusterManagementMode mode = new ClusterManagementMode(ClusterManagementMode.Type.CLUSTER_PAUSE,
+ ClusterManagementMode.Status.COMPLETED);
+ long time = System.currentTimeMillis();
+ String fromHost = NetworkUtil.getLocalhostName();
+ String reason = TestHelper.getTestMethodName();
+ controllerHistory.updateManagementModeHistory(controller, mode, fromHost, time, reason);
+
+ List<String> historyList = controllerHistory.getManagementModeHistory();
+ String lastHistory = historyList.get(historyList.size() - 1);
+ Map<String, String> historyMap = stringToMap(lastHistory);
+
+ Map<String, String> expectedMap = new HashMap<>();
+ expectedMap.put("CONTROLLER", controller);
+ expectedMap.put("TIME", Instant.ofEpochMilli(time).toString());
+ expectedMap.put("MODE", mode.getMode().name());
+ expectedMap.put("STATUS", mode.getStatus().name());
+ expectedMap.put(PauseSignal.PauseSignalProperty.FROM_HOST.name(), fromHost);
+ expectedMap.put(PauseSignal.PauseSignalProperty.REASON.name(), reason);
+
+ Assert.assertEquals(historyMap, expectedMap);
+
+ // Add more than 10 entries, it should only keep the latest 10.
+ List<String> reasonList = new ArrayList<>();
+ for (int i = 0; i < 15; i++) {
+ String reasonI = reason + "-" + i;
+ controllerHistory.updateManagementModeHistory(controller, mode, fromHost, time, reasonI);
+ reasonList.add(reasonI);
+ }
+
+ historyList = controllerHistory.getManagementModeHistory();
+
+ Assert.assertEquals(historyList.size(), 10);
+
+ // Assert the history is the latest 10 entries.
+ int i = 5;
+ for (String entry : historyList) {
+ Map<String, String> actual = stringToMap(entry);
+ Assert.assertEquals(actual.get(PauseSignal.PauseSignalProperty.REASON.name()),
+ reasonList.get(i++));
+ }
+ }
+
+ /**
+ * Performs conversion from a map string into a map. The string was converted by map's toString().
+ *
+ * @param mapAsString A string that is converted by map's toString() method.
+ * Example: "{k1=v1, k2=v2}"
+ * @return Map<String, String>
+ */
+ private static Map<String, String> stringToMap(String mapAsString) {
+ return Splitter.on(", ").withKeyValueSeparator('=')
+ .split(mapAsString.substring(1, mapAsString.length() - 1));
+ }
+}
[helix] 07/13: Add REST APIs for management mode (#1807)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit b52bc048ddd22ed9770540ff71024ecec5f59bc1
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Mon Jun 28 17:28:22 2021 -0700
Add REST APIs for management mode (#1807)
This commit adds JAVA and REST APIs to set cluster freeze mode and get the management mode status
---
.../src/main/java/org/apache/helix/HelixAdmin.java | 13 ++-
.../helix/api/status/ClusterManagementMode.java | 6 ++
.../api/status/ClusterManagementModeRequest.java | 10 ++
.../helix/controller/GenericHelixController.java | 1 +
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 15 ++-
.../java/org/apache/helix/mock/MockHelixAdmin.java | 6 ++
.../server/resources/helix/ClusterAccessor.java | 115 +++++++++++++++++++++
.../helix/rest/server/TestClusterAccessor.java | 70 +++++++++++++
8 files changed, 230 insertions(+), 6 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/HelixAdmin.java b/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
index d403571..05bd60e 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
@@ -23,6 +23,8 @@ import java.io.IOException;
import java.util.List;
import java.util.Map;
+import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.api.status.ClusterManagementModeRequest;
import org.apache.helix.api.topology.ClusterTopology;
import org.apache.helix.model.CloudConfig;
import org.apache.helix.model.ClusterConstraints;
@@ -37,8 +39,6 @@ import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.MaintenanceSignal;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
-import org.apache.helix.api.status.ClusterManagementMode;
-import org.apache.helix.api.status.ClusterManagementModeRequest;
/*
* Helix cluster management
@@ -382,6 +382,15 @@ public interface HelixAdmin {
void setClusterManagementMode(ClusterManagementModeRequest request);
/**
+ * Gets cluster management status {@link ClusterManagementMode}: what mode the cluster is and
+ * whether the cluster has fully reached to that mode.
+ *
+ * @param clusterName cluster name
+ * @return {@link ClusterManagementMode}
+ */
+ ClusterManagementMode getClusterManagementMode(String clusterName);
+
+ /**
* Reset a list of partitions in error state for an instance
* The partitions are assume to be in error state and reset will bring them from error
* to initial state. An error to initial state transition is required for reset.
diff --git a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
index d7a1637..cbd2019 100644
--- a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
+++ b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
@@ -55,6 +55,12 @@ public class ClusterManagementMode {
private final Type mode;
private final Status status;
+ // Default constructor for json deserialization
+ private ClusterManagementMode() {
+ mode = null;
+ status = null;
+ }
+
public ClusterManagementMode(Type mode, Status status) {
this.mode = mode;
this.status = status;
diff --git a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementModeRequest.java b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementModeRequest.java
index dd2fe58..f3d9b07 100644
--- a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementModeRequest.java
+++ b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementModeRequest.java
@@ -19,11 +19,14 @@ package org.apache.helix.api.status;
* under the License.
*/
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonPOJOBuilder;
import com.google.common.base.Preconditions;
/**
* Represents a request to set the cluster management mode {@link ClusterManagementMode}
*/
+@JsonDeserialize(builder = ClusterManagementModeRequest.Builder.class)
public class ClusterManagementModeRequest {
private final ClusterManagementMode.Type _mode;
private final String _clusterName;
@@ -57,6 +60,7 @@ public class ClusterManagementModeRequest {
return new Builder();
}
+ @JsonPOJOBuilder(buildMethodName = "buildFromJson")
public static final class Builder {
private ClusterManagementMode.Type mode;
private String clusterName;
@@ -94,6 +98,12 @@ public class ClusterManagementModeRequest {
return new ClusterManagementModeRequest(this);
}
+ // Used by Json deserializer
+ private ClusterManagementModeRequest buildFromJson() {
+ Preconditions.checkNotNull(mode, "Mode not set");
+ return new ClusterManagementModeRequest((this));
+ }
+
private void validate() {
Preconditions.checkNotNull(mode, "Mode not set");
Preconditions.checkNotNull(clusterName, "Cluster name not set");
diff --git a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
index 6cc1e5f..00bdfcc 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
@@ -83,6 +83,7 @@ import org.apache.helix.controller.stages.ExternalViewComputeStage;
import org.apache.helix.controller.stages.IntermediateStateCalcStage;
import org.apache.helix.controller.stages.MaintenanceRecoveryStage;
import org.apache.helix.controller.stages.ManagementModeStage;
+import org.apache.helix.controller.stages.MessageGenerationPhase;
import org.apache.helix.controller.stages.MessageSelectionStage;
import org.apache.helix.controller.stages.MessageThrottleStage;
import org.apache.helix.controller.stages.PersistAssignmentStage;
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
index 2545139..73c699e 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
@@ -54,6 +54,7 @@ import org.apache.helix.PropertyType;
import org.apache.helix.SystemPropertyKeys;
import org.apache.helix.api.exceptions.HelixConflictException;
import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.api.status.ClusterManagementModeRequest;
import org.apache.helix.api.topology.ClusterTopology;
import org.apache.helix.controller.rebalancer.strategy.RebalanceStrategy;
import org.apache.helix.controller.rebalancer.util.WagedValidationUtil;
@@ -62,6 +63,7 @@ import org.apache.helix.model.CloudConfig;
import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.ClusterConstraints;
import org.apache.helix.model.ClusterConstraints.ConstraintType;
+import org.apache.helix.model.ClusterStatus;
import org.apache.helix.model.ConstraintItem;
import org.apache.helix.model.ControllerHistory;
import org.apache.helix.model.CurrentState;
@@ -81,7 +83,6 @@ import org.apache.helix.model.ParticipantHistory;
import org.apache.helix.model.PauseSignal;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
-import org.apache.helix.api.status.ClusterManagementModeRequest;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.tools.DefaultIdealStateCalculator;
import org.apache.helix.util.HelixUtil;
@@ -521,6 +522,15 @@ public class ZKHelixAdmin implements HelixAdmin {
}
}
+ @Override
+ public ClusterManagementMode getClusterManagementMode(String clusterName) {
+ HelixDataAccessor accessor =
+ new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<>(_zkClient));
+ ClusterStatus status = accessor.getProperty(accessor.keyBuilder().clusterStatus());
+ return status == null ? null
+ : new ClusterManagementMode(status.getManagementMode(), status.getManagementModeStatus());
+ }
+
private void enableClusterPauseMode(String clusterName, boolean cancelPendingST, String reason) {
String hostname = NetworkUtil.getLocalhostName();
logger.info(
@@ -533,9 +543,6 @@ public class ZKHelixAdmin implements HelixAdmin {
if (baseDataAccessor.exists(accessor.keyBuilder().pause().getPath(), AccessOption.PERSISTENT)) {
throw new HelixConflictException(clusterName + " pause signal already exists");
}
- if (baseDataAccessor.exists(accessor.keyBuilder().maintenance().getPath(), AccessOption.PERSISTENT)) {
- throw new HelixConflictException(clusterName + " maintenance signal already exists");
- }
// check whether cancellation is enabled
ClusterConfig config = accessor.getProperty(accessor.keyBuilder().clusterConfig());
diff --git a/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java b/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java
index 2bacc74..f7a1034 100644
--- a/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java
+++ b/helix-core/src/test/java/org/apache/helix/mock/MockHelixAdmin.java
@@ -30,6 +30,7 @@ import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.PropertyPathBuilder;
import org.apache.helix.PropertyType;
+import org.apache.helix.api.status.ClusterManagementMode;
import org.apache.helix.api.topology.ClusterTopology;
import org.apache.helix.model.CloudConfig;
import org.apache.helix.model.ClusterConfig;
@@ -331,6 +332,11 @@ public class MockHelixAdmin implements HelixAdmin {
}
+ @Override
+ public ClusterManagementMode getClusterManagementMode(String clusterName) {
+ return null;
+ }
+
@Override public void resetPartition(String clusterName, String instanceName, String resourceName,
List<String> partitionNames) {
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 78010b9..9efcdc7 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
@@ -22,8 +22,10 @@ package org.apache.helix.rest.server.resources.helix;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import javax.ws.rs.DELETE;
import javax.ws.rs.DefaultValue;
import javax.ws.rs.GET;
@@ -36,17 +38,22 @@ import javax.ws.rs.core.Response;
import com.codahale.metrics.annotation.ResponseMetered;
import com.codahale.metrics.annotation.Timed;
+import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang3.StringUtils;
import org.apache.helix.AccessOption;
+import org.apache.helix.BaseDataAccessor;
import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.PropertyKey;
import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.api.exceptions.HelixConflictException;
+import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.api.status.ClusterManagementModeRequest;
import org.apache.helix.manager.zk.ZKUtil;
import org.apache.helix.model.CloudConfig;
import org.apache.helix.model.ClusterConfig;
@@ -66,6 +73,7 @@ import org.apache.helix.rest.server.service.ClusterServiceImpl;
import org.apache.helix.tools.ClusterSetup;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -300,6 +308,113 @@ public class ClusterAccessor extends AbstractHelixResource {
@ResponseMetered(name = HttpConstants.READ_REQUEST)
@Timed(name = HttpConstants.READ_REQUEST)
@GET
+ @Path("{clusterId}/management-mode")
+ public Response getClusterManagementMode(@PathParam("clusterId") String clusterId,
+ @QueryParam("showDetails") boolean showDetails) {
+ ClusterManagementMode mode = getHelixAdmin().getClusterManagementMode(clusterId);
+ if (mode == null) {
+ return notFound("Cluster " + clusterId + " is not in management mode");
+ }
+
+ Map<String, Object> responseMap = new HashMap<>();
+ responseMap.put("cluster", clusterId);
+ responseMap.put("mode", mode.getMode());
+ responseMap.put("status", mode.getStatus());
+ if (showDetails) {
+ // To show details, query participants that are in progress to management mode.
+ responseMap.put("details", getManagementModeDetails(clusterId, mode));
+ }
+
+ return JSONRepresentation(responseMap);
+ }
+
+ private Map<String, Object> getManagementModeDetails(String clusterId,
+ ClusterManagementMode mode) {
+ Map<String, Object> details = new HashMap<>();
+ Map<String, Object> participantDetails = new HashMap<>();
+ ClusterManagementMode.Status status = mode.getStatus();
+ details.put("cluster", ImmutableMap.of("cluster", clusterId, "status", status.name()));
+
+ boolean hasPendingST = false;
+ Set<String> liveInstancesInProgress = new HashSet<>();
+
+ if (ClusterManagementMode.Status.IN_PROGRESS.equals(status)) {
+ HelixDataAccessor accessor = getDataAccssor(clusterId);
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ List<LiveInstance> liveInstances = accessor.getChildValues(keyBuilder.liveInstances());
+ BaseDataAccessor<ZNRecord> baseAccessor = accessor.getBaseDataAccessor();
+
+ if (ClusterManagementMode.Type.CLUSTER_PAUSE.equals(mode.getMode())) {
+ // Entering cluster freeze mode, check live instance freeze status and pending ST
+ for (LiveInstance liveInstance : liveInstances) {
+ String instanceName = liveInstance.getInstanceName();
+ if (!LiveInstance.LiveInstanceStatus.PAUSED.equals(liveInstance.getStatus())) {
+ liveInstancesInProgress.add(instanceName);
+ }
+ Stat stat = baseAccessor
+ .getStat(keyBuilder.messages(instanceName).getPath(), AccessOption.PERSISTENT);
+ if (stat.getNumChildren() > 0) {
+ hasPendingST = true;
+ liveInstancesInProgress.add(instanceName);
+ }
+ }
+ } else if (ClusterManagementMode.Type.NORMAL.equals(mode.getMode())) {
+ // Exiting freeze mode, check live instance unfreeze status
+ for (LiveInstance liveInstance : liveInstances) {
+ if (LiveInstance.LiveInstanceStatus.PAUSED.equals(liveInstance.getStatus())) {
+ liveInstancesInProgress.add(liveInstance.getInstanceName());
+ }
+ }
+ }
+ }
+
+ participantDetails.put("status", status.name());
+ participantDetails.put("liveInstancesInProgress", liveInstancesInProgress);
+ if (ClusterManagementMode.Type.CLUSTER_PAUSE.equals(mode.getMode())) {
+ // Add pending ST result for cluster freeze mode
+ participantDetails.put("hasPendingStateTransition", hasPendingST);
+ }
+
+ details.put(ClusterProperties.liveInstances.name(), participantDetails);
+ return details;
+ }
+
+ @ResponseMetered(name = HttpConstants.WRITE_REQUEST)
+ @Timed(name = HttpConstants.WRITE_REQUEST)
+ @POST
+ @Path("{clusterId}/management-mode")
+ public Response updateClusterManagementMode(@PathParam("clusterId") String clusterId,
+ @DefaultValue("{}") String content) {
+ ClusterManagementModeRequest request;
+ try {
+ request = OBJECT_MAPPER.readerFor(ClusterManagementModeRequest.class).readValue(content);
+ } catch (JsonProcessingException e) {
+ LOG.warn("Failed to parse json string: {}", content, e);
+ return badRequest("Invalid payload json body: " + content);
+ }
+
+ // Need to add cluster name
+ request = ClusterManagementModeRequest.newBuilder()
+ .withClusterName(clusterId)
+ .withMode(request.getMode())
+ .withCancelPendingST(request.isCancelPendingST())
+ .withReason(request.getReason())
+ .build();
+
+ try {
+ getHelixAdmin().setClusterManagementMode(request);
+ } catch (HelixConflictException e) {
+ return Response.status(Response.Status.CONFLICT).entity(e.getMessage()).build();
+ } catch (HelixException e) {
+ return serverError(e.getMessage());
+ }
+
+ return JSONRepresentation(ImmutableMap.of("acknowledged", true));
+ }
+
+ @ResponseMetered(name = HttpConstants.READ_REQUEST)
+ @Timed(name = HttpConstants.READ_REQUEST)
+ @GET
@Path("{clusterId}/configs")
public Response getClusterConfig(@PathParam("clusterId") String clusterId) {
ConfigAccessor accessor = getConfigAccessor();
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 cca5284..8d2de70 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
@@ -36,16 +36,20 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.collect.ImmutableMap;
import com.sun.research.ws.wadl.HTTPMethods;
+import org.apache.helix.AccessOption;
import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.PropertyKey;
import org.apache.helix.TestHelper;
+import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.api.status.ClusterManagementModeRequest;
import org.apache.helix.cloud.azure.AzureConstants;
import org.apache.helix.cloud.constants.CloudProvider;
import org.apache.helix.controller.rebalancer.waged.WagedRebalancer;
import org.apache.helix.integration.manager.ClusterDistributedController;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.manager.zk.ZKUtil;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.model.CloudConfig;
import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.CustomizedStateConfig;
@@ -54,6 +58,7 @@ import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.MaintenanceSignal;
+import org.apache.helix.model.PauseSignal;
import org.apache.helix.model.RESTConfig;
import org.apache.helix.rest.common.HelixRestNamespace;
import org.apache.helix.rest.server.auditlog.AuditLog;
@@ -1285,6 +1290,71 @@ public class TestClusterAccessor extends AbstractTestClass {
System.out.println("End test :" + TestHelper.getTestMethodName());
}
+ @Test
+ public void testClusterFreezeMode() throws Exception {
+ String cluster = _clusters.iterator().next();
+ HelixDataAccessor dataAccessor =
+ new ZKHelixDataAccessor(cluster, new ZkBaseDataAccessor<>(_gZkClient));
+ // Pause not existed
+ Assert.assertNull(dataAccessor.getProperty(dataAccessor.keyBuilder().pause()));
+
+ String endpoint = "clusters/" + cluster + "/management-mode";
+
+ // Set cluster pause mode
+ ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
+ .withMode(ClusterManagementMode.Type.CLUSTER_PAUSE)
+ .withClusterName(cluster)
+ .build();
+ String payload = OBJECT_MAPPER.writeValueAsString(request);
+ post(endpoint, null, Entity.entity(payload, MediaType.APPLICATION_JSON_TYPE),
+ Response.Status.OK.getStatusCode());
+
+ PauseSignal pauseSignal = dataAccessor.getProperty(dataAccessor.keyBuilder().pause());
+ Assert.assertNotNull(pauseSignal);
+ Assert.assertTrue(pauseSignal.isClusterPause());
+ Assert.assertFalse(pauseSignal.getCancelPendingST());
+
+ // Wait until cluster status is persisted
+ TestHelper.verify(() -> dataAccessor.getBaseDataAccessor()
+ .exists(dataAccessor.keyBuilder().clusterStatus().getPath(), AccessOption.PERSISTENT),
+ TestHelper.WAIT_DURATION);
+
+ // Verify get cluster status
+ String body = get(endpoint, null, Response.Status.OK.getStatusCode(), true);
+ ClusterManagementMode mode =
+ OBJECT_MAPPER.readerFor(ClusterManagementMode.class).readValue(body);
+ Assert.assertEquals(mode.getMode(), ClusterManagementMode.Type.CLUSTER_PAUSE);
+ // Depending on timing, it could IN_PROGRESS or COMPLETED.
+ // It's just to verify the rest response format is correct
+ Assert.assertTrue(ClusterManagementMode.Status.IN_PROGRESS.equals(mode.getStatus())
+ || ClusterManagementMode.Status.COMPLETED.equals(mode.getStatus()));
+
+ body = get(endpoint, ImmutableMap.of("showDetails", "true"), Response.Status.OK.getStatusCode(),
+ true);
+ Map<String, Object> responseMap = OBJECT_MAPPER.readerFor(Map.class).readValue(body);
+ Map<String, Object> detailsMap = (Map<String, Object>) responseMap.get("details");
+
+ Assert.assertEquals(responseMap.get("cluster"), cluster);
+ Assert.assertEquals(responseMap.get("mode"), mode.getMode().name());
+ Assert.assertEquals(responseMap.get("status"), mode.getStatus().name());
+ Assert.assertTrue(responseMap.containsKey("details"));
+ Assert.assertTrue(detailsMap.containsKey("cluster"));
+ Assert.assertTrue(detailsMap.containsKey("liveInstances"));
+
+ // set normal mode
+ request = ClusterManagementModeRequest.newBuilder()
+ .withMode(ClusterManagementMode.Type.NORMAL)
+ .withClusterName(cluster)
+ .build();
+ payload = OBJECT_MAPPER.writeValueAsString(request);
+ post(endpoint, null, Entity.entity(payload, MediaType.APPLICATION_JSON_TYPE),
+ Response.Status.OK.getStatusCode());
+
+ // Pause signal is deleted
+ pauseSignal = dataAccessor.getProperty(dataAccessor.keyBuilder().pause());
+ Assert.assertNull(pauseSignal);
+ }
+
private ClusterConfig getClusterConfigFromRest(String cluster) throws IOException {
String body = get("clusters/" + cluster + "/configs", null, Response.Status.OK.getStatusCode(), true);
[helix] 03/13: Add management mode pipeline registry and switch
logic (#1769)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 7190eb3ee3150a77c53ae768f89ee2b0473cc711
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Wed Jun 9 21:01:30 2021 -0700
Add management mode pipeline registry and switch logic (#1769)
Management Mode Pipeline will help check the cluster status and determine whether the default pipelines can be run.
One use case is, it will help controller to decide when it can exit the cluster freeze mode.
This commit adds management mode pipeline and logic to switch from/to the default resource/task pipelines.
---
.../helix/controller/GenericHelixController.java | 151 ++++++++++++++++-----
.../ManagementControllerDataProvider.java} | 28 +---
.../apache/helix/controller/pipeline/Pipeline.java | 9 +-
.../helix/controller/stages/ClusterEventType.java | 1 +
.../controller/stages/ManagementModeStage.java | 48 +++++++
.../controller/stages/ResourceValidationStage.java | 10 ++
.../main/java/org/apache/helix/util/HelixUtil.java | 12 ++
.../java/org/apache/helix/util/RebalanceUtil.java | 22 +++
8 files changed, 223 insertions(+), 58 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
index 11cd9ee..7da6ef0 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
@@ -63,6 +63,7 @@ import org.apache.helix.api.listeners.TaskCurrentStateChangeListener;
import org.apache.helix.common.ClusterEventBlockingQueue;
import org.apache.helix.common.DedupEventProcessor;
import org.apache.helix.controller.dataproviders.BaseControllerDataProvider;
+import org.apache.helix.controller.dataproviders.ManagementControllerDataProvider;
import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
import org.apache.helix.controller.dataproviders.WorkflowControllerDataProvider;
import org.apache.helix.controller.pipeline.AsyncWorkerType;
@@ -81,7 +82,7 @@ import org.apache.helix.controller.stages.CustomizedViewAggregationStage;
import org.apache.helix.controller.stages.ExternalViewComputeStage;
import org.apache.helix.controller.stages.IntermediateStateCalcStage;
import org.apache.helix.controller.stages.MaintenanceRecoveryStage;
-import org.apache.helix.controller.stages.MessageGenerationPhase;
+import org.apache.helix.controller.stages.ManagementModeStage;
import org.apache.helix.controller.stages.MessageSelectionStage;
import org.apache.helix.controller.stages.MessageThrottleStage;
import org.apache.helix.controller.stages.PersistAssignmentStage;
@@ -141,6 +142,7 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
private static final int ASYNC_TASKS_THREADPOOL_SIZE = 10;
private final PipelineRegistry _registry;
private final PipelineRegistry _taskRegistry;
+ private final PipelineRegistry _managementModeRegistry;
final AtomicReference<Map<String, LiveInstance>> _lastSeenInstances;
final AtomicReference<Map<String, LiveInstance>> _lastSeenSessions;
@@ -163,6 +165,11 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
private final ClusterEventBlockingQueue _taskEventQueue;
private final ClusterEventProcessor _taskEventThread;
+ // Controller will switch to run management mode pipeline when set to true.
+ private boolean _inManagementMode;
+ private final ClusterEventBlockingQueue _managementModeEventQueue;
+ private final ClusterEventProcessor _managementModeEventThread;
+
private final Map<AsyncWorkerType, DedupEventProcessor<String, Runnable>> _asyncFIFOWorkerPool;
private long _continuousRebalanceFailureCount = 0;
@@ -197,6 +204,7 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
*/
private final ResourceControllerDataProvider _resourceControlDataProvider;
private final WorkflowControllerDataProvider _workflowControlDataProvider;
+ private final ManagementControllerDataProvider _managementControllerDataProvider;
private final ScheduledExecutorService _asyncTasksThreadPool;
/**
@@ -285,13 +293,21 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
public GenericHelixController(String clusterName) {
this(createDefaultRegistry(Pipeline.Type.DEFAULT.name()),
- createTaskRegistry(Pipeline.Type.TASK.name()), clusterName,
+ createTaskRegistry(Pipeline.Type.TASK.name()),
+ createManagementModeRegistry(Pipeline.Type.MANAGEMENT_MODE.name()), clusterName,
Sets.newHashSet(Pipeline.Type.TASK, Pipeline.Type.DEFAULT));
}
public GenericHelixController(String clusterName, Set<Pipeline.Type> enabledPipelins) {
this(createDefaultRegistry(Pipeline.Type.DEFAULT.name()),
- createTaskRegistry(Pipeline.Type.TASK.name()), clusterName, enabledPipelins);
+ createTaskRegistry(Pipeline.Type.TASK.name()),
+ createManagementModeRegistry(Pipeline.Type.MANAGEMENT_MODE.name()),
+ clusterName,
+ enabledPipelins);
+ }
+
+ public void setInManagementMode(boolean enabled) {
+ _inManagementMode = enabled;
}
class RebalanceTask extends TimerTask {
@@ -566,6 +582,8 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
registry
.register(ClusterEventType.OnDemandRebalance, dataRefresh, autoExitMaintenancePipeline,
dataPreprocess, externalViewPipeline, rebalancePipeline);
+ registry.register(ClusterEventType.ControllerChange, dataRefresh, autoExitMaintenancePipeline,
+ dataPreprocess, externalViewPipeline, rebalancePipeline);
// TODO: We now include rebalance pipeline in customized state change for correctness.
// However, it is not efficient, and we should improve this by splitting the pipeline or
// controller roles to multiple hosts.
@@ -625,22 +643,49 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
rebalancePipeline);
registry.register(ClusterEventType.OnDemandRebalance, dataRefresh, dataPreprocess,
rebalancePipeline);
+ registry.register(ClusterEventType.ControllerChange, dataRefresh, dataPreprocess,
+ rebalancePipeline);
+ return registry;
+ }
+ }
+
+ private static PipelineRegistry createManagementModeRegistry(String pipelineName) {
+ logger.info("Creating management mode registry");
+ synchronized (GenericHelixController.class) {
+ // cluster data cache refresh
+ Pipeline dataRefresh = new Pipeline(pipelineName);
+ dataRefresh.addStage(new ReadClusterDataStage());
+
+ // cluster management mode process
+ Pipeline managementMode = new Pipeline(pipelineName);
+ managementMode.addStage(new ManagementModeStage());
+
+ PipelineRegistry registry = new PipelineRegistry();
+ Arrays.asList(
+ ClusterEventType.ControllerChange,
+ ClusterEventType.LiveInstanceChange,
+ ClusterEventType.MessageChange,
+ ClusterEventType.OnDemandRebalance,
+ ClusterEventType.PeriodicalRebalance
+ ).forEach(type -> registry.register(type, dataRefresh, managementMode));
+
return registry;
}
}
// TODO: refactor the constructor as providing both registry but only enabling one looks confusing
public GenericHelixController(PipelineRegistry registry, PipelineRegistry taskRegistry) {
- this(registry, taskRegistry, null, Sets.newHashSet(
- Pipeline.Type.TASK, Pipeline.Type.DEFAULT));
+ this(registry, taskRegistry, createManagementModeRegistry(Pipeline.Type.MANAGEMENT_MODE.name()),
+ null, Sets.newHashSet(Pipeline.Type.TASK, Pipeline.Type.DEFAULT));
}
private GenericHelixController(PipelineRegistry registry, PipelineRegistry taskRegistry,
- final String clusterName, Set<Pipeline.Type> enabledPipelineTypes) {
- _paused = false;
+ PipelineRegistry managementModeRegistry, final String clusterName,
+ Set<Pipeline.Type> enabledPipelineTypes) {
_enabledPipelineTypes = enabledPipelineTypes;
_registry = registry;
_taskRegistry = taskRegistry;
+ _managementModeRegistry = managementModeRegistry;
_lastSeenInstances = new AtomicReference<>();
_lastSeenSessions = new AtomicReference<>();
_lastSeenCustomizedStateTypesMapRef = new AtomicReference<>();
@@ -660,6 +705,7 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
_onDemandRebalanceTimer =
new Timer("GenericHelixController_" + _clusterName + "_onDemand_Timer", true);
+ // TODO: refactor to simplify below similar code of the 3 pipelines
// initialize pipelines at the end so we have everything else prepared
if (_enabledPipelineTypes.contains(Pipeline.Type.DEFAULT)) {
logger.info("Initializing {} pipeline", Pipeline.Type.DEFAULT.name());
@@ -689,6 +735,16 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
_taskEventThread = null;
}
+ logger.info("Initializing {} pipeline", Pipeline.Type.MANAGEMENT_MODE.name());
+ _managementControllerDataProvider =
+ new ManagementControllerDataProvider(clusterName, Pipeline.Type.MANAGEMENT_MODE.name());
+ _managementModeEventQueue = new ClusterEventBlockingQueue();
+ _managementModeEventThread =
+ new ClusterEventProcessor(_managementControllerDataProvider, _managementModeEventQueue,
+ Pipeline.Type.MANAGEMENT_MODE.name() + "-" + clusterName);
+ initPipeline(_managementModeEventThread, _managementControllerDataProvider);
+ logger.info("Initialized {} pipeline", Pipeline.Type.MANAGEMENT_MODE.name());
+
addController(this);
}
@@ -776,12 +832,36 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
_helixManager = manager;
- // TODO If init controller with paused = true, it may not take effect immediately
- // _paused is default false. If any events come before controllerChangeEvent, the controller
- // will be excuting in un-paused mode. Which might not be the config in ZK.
- if (_paused) {
- logger.info("Cluster " + manager.getClusterName() + " is paused. Ignoring the event:" + event
- .getEventType());
+ // Prepare ClusterEvent
+ // TODO (harry): this is a temporal workaround - after controller is separated we should not
+ // have this instanceof clauses
+ List<Pipeline> pipelines;
+ boolean isTaskFrameworkPipeline = false;
+ Pipeline.Type pipelineType;
+
+ if (dataProvider instanceof ResourceControllerDataProvider) {
+ pipelines = _registry.getPipelinesForEvent(event.getEventType());
+ pipelineType = Pipeline.Type.DEFAULT;
+ } else if (dataProvider instanceof WorkflowControllerDataProvider) {
+ pipelines = _taskRegistry.getPipelinesForEvent(event.getEventType());
+ isTaskFrameworkPipeline = true;
+ pipelineType = Pipeline.Type.TASK;
+ } else if (dataProvider instanceof ManagementControllerDataProvider) {
+ pipelines = _managementModeRegistry.getPipelinesForEvent(event.getEventType());
+ pipelineType = Pipeline.Type.MANAGEMENT_MODE;
+ } else {
+ logger.warn(String
+ .format("No %s pipeline to run for event: %s::%s", dataProvider.getPipelineName(),
+ event.getEventType(), event.getEventId()));
+ return;
+ }
+
+ // Should not run management mode and default/task pipelines at the same time.
+ if ((_inManagementMode && !Pipeline.Type.MANAGEMENT_MODE.equals(pipelineType))
+ || (!_inManagementMode && Pipeline.Type.MANAGEMENT_MODE.equals(pipelineType))) {
+ logger.info("Should not run management mode and default/task pipelines at the same time. "
+ + "cluster={}, inManagementMode={}, pipelineType={}. Ignoring the event: {}",
+ manager.getClusterName(), _inManagementMode, pipelineType, event.getEventType());
return;
}
@@ -808,26 +888,6 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
dataProvider.setClusterEventId(event.getEventId());
event.addAttribute(AttributeName.LastRebalanceFinishTimeStamp.name(), _lastPipelineEndTimestamp);
-
- // Prepare ClusterEvent
- // TODO (harry): this is a temporal workaround - after controller is separated we should not
- // have this instanceof clauses
- List<Pipeline> pipelines;
- boolean isTaskFrameworkPipeline = false;
-
- if (dataProvider instanceof ResourceControllerDataProvider) {
- pipelines = _registry
- .getPipelinesForEvent(event.getEventType());
- } else if (dataProvider instanceof WorkflowControllerDataProvider) {
- pipelines = _taskRegistry
- .getPipelinesForEvent(event.getEventType());
- isTaskFrameworkPipeline = true;
- } else {
- logger.warn(String
- .format("No %s pipeline to run for event: %s::%s", dataProvider.getPipelineName(),
- event.getEventType(), event.getEventId()));
- return;
- }
event.addAttribute(AttributeName.ControllerDataProvider.name(), dataProvider);
logger.info("START: Invoking {} controller pipeline for cluster: {}. Event type: {}, ID: {}. "
@@ -1202,6 +1262,10 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
if (_workflowControlDataProvider != null) {
_workflowControlDataProvider.notifyDataChange(type, path);
}
+
+ if (_managementControllerDataProvider != null) {
+ _managementControllerDataProvider.notifyDataChange(type, path);
+ }
}
private void requestDataProvidersFullRefresh() {
@@ -1212,6 +1276,10 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
if (_workflowControlDataProvider != null) {
_workflowControlDataProvider.requireFullRefresh();
}
+
+ if (_managementControllerDataProvider != null) {
+ _managementControllerDataProvider.requireFullRefresh();
+ }
}
private void pushToEventQueues(ClusterEventType eventType, NotificationContext changeContext,
@@ -1228,6 +1296,14 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
for (Map.Entry<String, Object> attr : eventAttributes.entrySet()) {
event.addAttribute(attr.getKey(), attr.getValue());
}
+
+ // Management mode event will force management mode pipeline.
+ if (_inManagementMode) {
+ event.setEventId(uid + "_" + Pipeline.Type.MANAGEMENT_MODE.name());
+ enqueueEvent(_managementModeEventQueue, event);
+ return;
+ }
+
enqueueEvent(_eventQueue, event);
enqueueEvent(_taskEventQueue,
event.clone(String.format("%s_%s", uid, Pipeline.Type.TASK.name())));
@@ -1269,7 +1345,10 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
boolean prevInMaintenanceMode = _inMaintenanceMode;
_paused = updateControllerState(pauseSignal, _paused);
_inMaintenanceMode = updateControllerState(maintenanceSignal, _inMaintenanceMode);
- triggerResumeEvent(changeContext, prevPaused, prevInMaintenanceMode);
+ // TODO: remove triggerResumeEvent when moving pause/maintenance to management pipeline
+ if (!triggerResumeEvent(changeContext, prevPaused, prevInMaintenanceMode)) {
+ pushToEventQueues(ClusterEventType.ControllerChange, changeContext, Collections.emptyMap());
+ }
enableClusterStatusMonitor(true);
_clusterStatusMonitor.setEnabled(!_paused);
@@ -1484,7 +1563,7 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
* @param prevPaused the previous paused status.
* @param prevInMaintenanceMode the previous in maintenance mode status.
*/
- private void triggerResumeEvent(NotificationContext changeContext, boolean prevPaused,
+ private boolean triggerResumeEvent(NotificationContext changeContext, boolean prevPaused,
boolean prevInMaintenanceMode) {
/**
* WARNING: the logic here is tricky.
@@ -1496,7 +1575,9 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
if (!_paused && (prevPaused || (prevInMaintenanceMode && !_inMaintenanceMode))) {
pushToEventQueues(ClusterEventType.Resume, changeContext, Collections.EMPTY_MAP);
logger.info("controller is now resumed from paused/maintenance state");
+ return true;
}
+ return false;
}
// TODO: refactor this to use common/ClusterEventProcessor.
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterEventType.java b/helix-core/src/main/java/org/apache/helix/controller/dataproviders/ManagementControllerDataProvider.java
similarity index 62%
copy from helix-core/src/main/java/org/apache/helix/controller/stages/ClusterEventType.java
copy to helix-core/src/main/java/org/apache/helix/controller/dataproviders/ManagementControllerDataProvider.java
index cd0ce60..d178ca5 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterEventType.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/dataproviders/ManagementControllerDataProvider.java
@@ -1,4 +1,4 @@
-package org.apache.helix.controller.stages;
+package org.apache.helix.controller.dataproviders;
/*
* Licensed to the Apache Software Foundation (ASF) under one
@@ -19,25 +19,9 @@ package org.apache.helix.controller.stages;
* under the License.
*/
-public enum ClusterEventType {
- IdealStateChange,
- CurrentStateChange,
- TaskCurrentStateChange,
- CustomizedStateChange,
- ConfigChange,
- ClusterConfigChange,
- ResourceConfigChange,
- InstanceConfigChange,
- CustomizeStateConfigChange,
- LiveInstanceChange,
- MessageChange,
- ExternalViewChange,
- CustomizedViewChange,
- TargetExternalViewChange,
- Resume,
- PeriodicalRebalance,
- OnDemandRebalance,
- RetryRebalance,
- StateVerifier,
- Unknown
+public class ManagementControllerDataProvider extends BaseControllerDataProvider {
+ // TODO: implement this class to only refresh required event types
+ public ManagementControllerDataProvider(String clusterName, String name) {
+ super(clusterName, name);
+ }
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/pipeline/Pipeline.java b/helix-core/src/main/java/org/apache/helix/controller/pipeline/Pipeline.java
index ecf42da..4196e23 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/pipeline/Pipeline.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/pipeline/Pipeline.java
@@ -35,7 +35,14 @@ public class Pipeline {
public enum Type {
DEFAULT,
- TASK
+ TASK,
+
+ /**
+ * A pipeline used to manage the cluster when it is in admin management mode:
+ * cluster freeze mode, controller pause mode, etc. Used by Helix internally,
+ * not meant to be used for Helix external users.
+ */
+ MANAGEMENT_MODE
}
public Pipeline() {
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterEventType.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterEventType.java
index cd0ce60..65f6bb4 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterEventType.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterEventType.java
@@ -37,6 +37,7 @@ public enum ClusterEventType {
Resume,
PeriodicalRebalance,
OnDemandRebalance,
+ ControllerChange,
RetryRebalance,
StateVerifier,
Unknown
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
new file mode 100644
index 0000000..512224d
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
@@ -0,0 +1,48 @@
+package org.apache.helix.controller.stages;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.helix.controller.dataproviders.ManagementControllerDataProvider;
+import org.apache.helix.controller.pipeline.AbstractBaseStage;
+import org.apache.helix.controller.pipeline.StageException;
+import org.apache.helix.util.HelixUtil;
+import org.apache.helix.util.RebalanceUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Checks the cluster status whether the cluster is in management mode.
+ */
+public class ManagementModeStage extends AbstractBaseStage {
+ private static final Logger LOG = LoggerFactory.getLogger(ManagementModeStage.class);
+
+ @Override
+ public void process(ClusterEvent event) throws Exception {
+ // TODO: implement the stage
+ String clusterName = event.getClusterName();
+ ManagementControllerDataProvider cache =
+ event.getAttribute(AttributeName.ControllerDataProvider.name());
+ if (!HelixUtil.inManagementMode(cache)) {
+ LOG.info("Exiting management mode pipeline for cluster {}", clusterName);
+ RebalanceUtil.enableManagementMode(clusterName, false);
+ throw new StageException("Exiting management mode pipeline for cluster " + clusterName);
+ }
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java
index 075c40f..a4d4783 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ResourceValidationStage.java
@@ -28,6 +28,8 @@ import org.apache.helix.controller.pipeline.StageException;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.Resource;
import org.apache.helix.model.StateModelDefinition;
+import org.apache.helix.util.HelixUtil;
+import org.apache.helix.util.RebalanceUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,6 +43,14 @@ public class ResourceValidationStage extends AbstractBaseStage {
if (cache == null) {
throw new StageException("Missing attributes in event:" + event + ". Requires DataCache");
}
+
+ // Check if cluster is still in management mode. Eg. there exists any frozen live instance.
+ if (HelixUtil.inManagementMode(cache)) {
+ // Trigger an immediate management mode pipeline.
+ RebalanceUtil.enableManagementMode(event.getClusterName(), true);
+ throw new StageException("Pipeline should not be run because cluster is in management mode");
+ }
+
Map<String, Resource> resourceMap = event.getAttribute(AttributeName.RESOURCES.name());
if (resourceMap == null) {
throw new StageException("Resources must be computed prior to validation!");
diff --git a/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java b/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java
index ce22c5f..3151716 100644
--- a/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/util/HelixUtil.java
@@ -38,6 +38,7 @@ import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.PropertyType;
import org.apache.helix.controller.common.PartitionStateMap;
+import org.apache.helix.controller.dataproviders.BaseControllerDataProvider;
import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
import org.apache.helix.controller.rebalancer.AbstractRebalancer;
import org.apache.helix.controller.rebalancer.strategy.RebalanceStrategy;
@@ -547,4 +548,15 @@ public final class HelixUtil {
instanceConfig.setInstanceEnabled(true);
return instanceConfig;
}
+
+ /**
+ * Checks whether or not the cluster is in management mode.
+ *
+ * @param cache
+ * @return
+ */
+ public static boolean inManagementMode(BaseControllerDataProvider cache) {
+ // TODO: implement the logic. Parameters can also change
+ return true;
+ }
}
diff --git a/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java b/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java
index 91ec406..db2b76f 100644
--- a/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java
@@ -145,6 +145,28 @@ public class RebalanceUtil {
return result;
}
+ /**
+ * Enables/disables controller to run management mode pipeline.
+ *
+ * @param clusterName target cluster name
+ * @param enabled enable/disable controller to management mode pipeline
+ */
+ public static void enableManagementMode(String clusterName, boolean enabled) {
+ GenericHelixController leaderController =
+ GenericHelixController.getLeaderController(clusterName);
+ if (leaderController != null) {
+ LOG.info("Switching management mode pipeline for cluster={}, enabled={}", clusterName,
+ enabled);
+ leaderController.setInManagementMode(enabled);
+ } else {
+ LOG.error("Failed to switch management mode pipeline, enabled={}. "
+ + "Controller for cluster {} does not exist", clusterName, enabled);
+ }
+
+ // Triggers an event to immediately run the pipeline
+ scheduleOnDemandPipeline(clusterName, 0L);
+ }
+
public static void scheduleOnDemandPipeline(String clusterName, long delay) {
scheduleOnDemandPipeline(clusterName, delay, true);
}
[helix] 11/13: Refactor freeze mode naming and getStatus default
value (#1818)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit d2f5637f9d1a7cce54b024ac714d4552809ff2f9
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Wed Jul 14 19:32:27 2021 -0700
Refactor freeze mode naming and getStatus default value (#1818)
Rename CLUSTER_PAUSE with CLUSTER_FREEZE mode.
Change default return value for LiveInstance#getStatus(): null -> NORMAL. And fix tests accordingly.
---
.../helix/api/status/ClusterManagementMode.java | 6 +++---
.../stages/ManagementMessageGenerationPhase.java | 4 +---
.../helix/controller/stages/ManagementModeStage.java | 8 +++-----
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 2 +-
.../helix/messaging/handling/HelixTaskExecutor.java | 3 ++-
.../java/org/apache/helix/model/LiveInstance.java | 6 ++++--
.../stages/TestManagementMessageGeneration.java | 2 +-
.../controller/stages/TestManagementModeStage.java | 4 ++--
.../integration/controller/TestClusterFreezeMode.java | 10 +++++-----
.../integration/paticipant/TestParticipantFreeze.java | 15 +++++++++------
.../org/apache/helix/manager/zk/TestZkHelixAdmin.java | 4 ++--
.../helix/model/TestControllerHistoryModel.java | 2 +-
.../java/org/apache/helix/model/TestLiveInstance.java | 4 +++-
.../rest/server/resources/helix/ClusterAccessor.java | 4 ++--
.../apache/helix/rest/server/TestClusterAccessor.java | 19 ++++++++++---------
15 files changed, 49 insertions(+), 44 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
index 86817d1..a342a8c 100644
--- a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
+++ b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
@@ -33,10 +33,10 @@ public class ClusterManagementMode {
NORMAL,
/**
- * Puts a cluster into pause mode, which will pause controller and participants.
+ * Puts a cluster into freeze mode, which will freeze controller and participants.
* This can be used to retain the cluster state.
*/
- CLUSTER_PAUSE,
+ CLUSTER_FREEZE,
/** Pause controller only, but not participants. */
CONTROLLER_PAUSE,
@@ -88,7 +88,7 @@ public class ClusterManagementMode {
*/
public LiveInstance.LiveInstanceStatus getDesiredParticipantStatus() {
switch (mode) {
- case CLUSTER_PAUSE:
+ case CLUSTER_FREEZE:
return LiveInstance.LiveInstanceStatus.PAUSED;
case NORMAL:
return LiveInstance.LiveInstanceStatus.NORMAL;
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
index c795deb..4479d50 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
@@ -96,9 +96,7 @@ public class ManagementMessageGenerationPhase extends MessageGenerationPhase {
LiveInstance liveInstance = liveInstanceMap.get(instanceName);
Collection<Message> pendingMessages = allInstanceMessages.get(instanceName);
String sessionId = liveInstance.getEphemeralOwner();
- LiveInstanceStatus liveInstanceStatus = liveInstance.getStatus();
- LiveInstanceStatus currentStatus = (liveInstanceStatus == null
- ? LiveInstanceStatus.NORMAL : liveInstanceStatus);
+ LiveInstanceStatus currentStatus = liveInstance.getStatus();
if (needStatusChangeMessage(pendingMessages, currentStatus, desiredStatus)) {
Message statusChangeMessage = MessageUtil.createStatusChangeMessage(currentStatus,
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
index b639685..55a320a 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
@@ -30,7 +30,6 @@ import org.apache.helix.HelixManager;
import org.apache.helix.PropertyKey;
import org.apache.helix.PropertyType;
import org.apache.helix.api.status.ClusterManagementMode;
-import org.apache.helix.controller.LogUtil;
import org.apache.helix.controller.dataproviders.ManagementControllerDataProvider;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
import org.apache.helix.controller.pipeline.StageException;
@@ -40,7 +39,6 @@ import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.LiveInstance.LiveInstanceStatus;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.MessageType;
-import org.apache.helix.model.Partition;
import org.apache.helix.model.PauseSignal;
import org.apache.helix.model.Resource;
import org.apache.helix.util.HelixUtil;
@@ -104,7 +102,7 @@ public class ManagementModeStage extends AbstractBaseStage {
status = ClusterManagementMode.Status.IN_PROGRESS;
}
} else if (pauseSignal.isClusterPause()) {
- type = ClusterManagementMode.Type.CLUSTER_PAUSE;
+ type = ClusterManagementMode.Type.CLUSTER_FREEZE;
if (!instancesFullyFrozen(enabledLiveInstances, liveInstanceMap, allInstanceMessages)) {
status = ClusterManagementMode.Status.IN_PROGRESS;
}
@@ -144,9 +142,9 @@ public class ManagementModeStage extends AbstractBaseStage {
// If there is any pending message sent by users, status could be computed as in progress.
// Skip recording status change to avoid confusion after cluster is already fully frozen.
- if (ClusterManagementMode.Type.CLUSTER_PAUSE.equals(recordedType)
+ if (ClusterManagementMode.Type.CLUSTER_FREEZE.equals(recordedType)
&& ClusterManagementMode.Status.COMPLETED.equals(recordedStatus)
- && ClusterManagementMode.Type.CLUSTER_PAUSE.equals(mode.getMode())
+ && ClusterManagementMode.Type.CLUSTER_FREEZE.equals(mode.getMode())
&& ClusterManagementMode.Status.IN_PROGRESS.equals(mode.getStatus())) {
LOG.info("Skip recording status mode={}, status={}, because cluster is fully frozen",
mode.getMode(), mode.getStatus());
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
index 73c699e..05bef08 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
@@ -509,7 +509,7 @@ public class ZKHelixAdmin implements HelixAdmin {
// TODO: support other modes
switch (mode) {
- case CLUSTER_PAUSE:
+ case CLUSTER_FREEZE:
enableClusterPauseMode(clusterName, request.isCancelPendingST(), reason);
break;
case NORMAL:
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
index ff970c2..11badac 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
@@ -187,6 +187,7 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
_timer = new Timer("HelixTaskExecutor_Timer", true);
_isShuttingDown = false;
+ _liveInstanceStatus = LiveInstanceStatus.NORMAL;
startMonitorThread();
}
@@ -1370,7 +1371,7 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
manager.getStateMachineEngine(), false);
}
_freezeSessionId = null;
- _liveInstanceStatus = null;
+ _liveInstanceStatus = toStatus;
success = accessor.getBaseDataAccessor().update(path, record -> {
// Remove the status field for backwards compatibility
record.getSimpleFields().remove(LiveInstance.LiveInstanceProperty.STATUS.name());
diff --git a/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java b/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
index fed2940..e711fea 100644
--- a/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
+++ b/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
@@ -142,10 +142,12 @@ public class LiveInstance extends HelixProperty {
}
/**
- * Gets the live instance's status. Returns null if the status field is not set.
+ * Gets the live instance's status. Returns {@link LiveInstanceStatus#NORMAL} if the status
+ * field is not set.
*/
public LiveInstanceStatus getStatus() {
- return _record.getEnumField(LiveInstanceProperty.STATUS.name(), LiveInstanceStatus.class, null);
+ return _record.getEnumField(LiveInstanceProperty.STATUS.name(), LiveInstanceStatus.class,
+ LiveInstanceStatus.NORMAL);
}
/**
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementMessageGeneration.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementMessageGeneration.java
index c16b503..7c20b02 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementMessageGeneration.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementMessageGeneration.java
@@ -128,7 +128,7 @@ public class TestManagementMessageGeneration extends ManagementMessageGeneration
RebalanceUtil.buildBestPossibleState(resourceMap.keySet(), currentStateOutput);
// Process the event
- ClusterManagementMode mode = new ClusterManagementMode(ClusterManagementMode.Type.CLUSTER_PAUSE,
+ ClusterManagementMode mode = new ClusterManagementMode(ClusterManagementMode.Type.CLUSTER_FREEZE,
ClusterManagementMode.Status.IN_PROGRESS);
event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), bestPossibleStateOutput);
event.addAttribute(AttributeName.CLUSTER_STATUS.name(), mode);
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
index 2737bf7..1be5c6b 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
@@ -76,7 +76,7 @@ public class TestManagementModeStage extends ZkTestBase {
// Freeze cluster
ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
.withClusterName(_clusterName)
- .withMode(ClusterManagementMode.Type.CLUSTER_PAUSE)
+ .withMode(ClusterManagementMode.Type.CLUSTER_FREEZE)
.withReason("test")
.build();
_gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
@@ -91,7 +91,7 @@ public class TestManagementModeStage extends ZkTestBase {
// In frozen mode
ClusterStatus clusterStatus = _accessor.getProperty(_accessor.keyBuilder().clusterStatus());
- Assert.assertEquals(clusterStatus.getManagementMode(), ClusterManagementMode.Type.CLUSTER_PAUSE);
+ Assert.assertEquals(clusterStatus.getManagementMode(), ClusterManagementMode.Type.CLUSTER_FREEZE);
// Mark a live instance to be pause state
diff --git a/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java b/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
index cbc8535..afaea91 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
@@ -153,7 +153,7 @@ public class TestClusterFreezeMode extends ZkTestBase {
// Freeze cluster
ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
.withClusterName(_clusterName)
- .withMode(ClusterManagementMode.Type.CLUSTER_PAUSE)
+ .withMode(ClusterManagementMode.Type.CLUSTER_FREEZE)
.withReason(methodName)
.build();
_gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
@@ -164,7 +164,7 @@ public class TestClusterFreezeMode extends ZkTestBase {
// Cluster is in progress to cluster pause because there is a pending state transition message
ClusterStatus expectedClusterStatus = new ClusterStatus();
- expectedClusterStatus.setManagementMode(ClusterManagementMode.Type.CLUSTER_PAUSE);
+ expectedClusterStatus.setManagementMode(ClusterManagementMode.Type.CLUSTER_FREEZE);
expectedClusterStatus.setManagementModeStatus(ClusterManagementMode.Status.IN_PROGRESS);
verifyClusterStatus(expectedClusterStatus);
@@ -175,7 +175,7 @@ public class TestClusterFreezeMode extends ZkTestBase {
verifyLiveInstanceStatus(_participants, LiveInstance.LiveInstanceStatus.PAUSED);
expectedClusterStatus = new ClusterStatus();
- expectedClusterStatus.setManagementMode(ClusterManagementMode.Type.CLUSTER_PAUSE);
+ expectedClusterStatus.setManagementMode(ClusterManagementMode.Type.CLUSTER_FREEZE);
expectedClusterStatus.setManagementModeStatus(ClusterManagementMode.Status.COMPLETED);
verifyClusterStatus(expectedClusterStatus);
@@ -187,7 +187,7 @@ public class TestClusterFreezeMode extends ZkTestBase {
return false;
}
String lastHistory = managementHistory.get(managementHistory.size() - 1);
- return lastHistory.contains("MODE=" + ClusterManagementMode.Type.CLUSTER_PAUSE)
+ return lastHistory.contains("MODE=" + ClusterManagementMode.Type.CLUSTER_FREEZE)
&& lastHistory.contains("STATUS=" + ClusterManagementMode.Status.COMPLETED)
&& lastHistory.contains("REASON=" + methodName);
}, TestHelper.WAIT_DURATION));
@@ -303,7 +303,7 @@ public class TestClusterFreezeMode extends ZkTestBase {
.build();
_gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
- verifyLiveInstanceStatus(_participants, null);
+ verifyLiveInstanceStatus(_participants, LiveInstance.LiveInstanceStatus.NORMAL);
ClusterStatus expectedClusterStatus = new ClusterStatus();
expectedClusterStatus.setManagementMode(ClusterManagementMode.Type.NORMAL);
diff --git a/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java b/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java
index fedee37..1eb760e 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java
@@ -117,7 +117,9 @@ public class TestParticipantFreeze extends ZkTestBase {
@Test
public void testNormalLiveInstanceStatus() {
LiveInstance liveInstance = _accessor.getProperty(_keyBuilder.liveInstance(_instanceName));
- Assert.assertNull(liveInstance.getStatus());
+ Assert.assertEquals(liveInstance.getStatus(), LiveInstance.LiveInstanceStatus.NORMAL);
+ Assert.assertNull(
+ liveInstance.getRecord().getSimpleField(LiveInstance.LiveInstanceProperty.STATUS.name()));
}
@Test(dependsOnMethods = "testNormalLiveInstanceStatus")
@@ -151,7 +153,7 @@ public class TestParticipantFreeze extends ZkTestBase {
// New live instance ephemeral node
Assert.assertEquals(liveInstance.getEphemeralOwner(), _participants[1].getSessionId());
// Status is not frozen because controller is not running, no freeze message sent.
- verifyLiveInstanceStatus(_participants[1], null);
+ verifyLiveInstanceStatus(_participants[1], LiveInstance.LiveInstanceStatus.NORMAL);
// Old session current state is deleted because of current state carry-over
Assert.assertTrue(TestHelper.verify(
@@ -221,7 +223,7 @@ public class TestParticipantFreeze extends ZkTestBase {
// Live instance status is NORMAL, but set to null value in both memory and zk.
// After live instance status is updated, the process is completed.
- verifyLiveInstanceStatus(_participants[0], null);
+ verifyLiveInstanceStatus(_participants[0], LiveInstance.LiveInstanceStatus.NORMAL);
// Unfreeze message is correctly deleted
Assert.assertNull(
_accessor.getProperty(_keyBuilder.message(_instanceName, unfreezeMessage.getId())));
@@ -242,7 +244,7 @@ public class TestParticipantFreeze extends ZkTestBase {
private void verifyLiveInstanceStatus(MockParticipantManager participant,
LiveInstance.LiveInstanceStatus status) throws Exception {
- // Live instance status is frozen in both memory and zk
+ // Verify live instance status in both memory and zk
Assert.assertTrue(TestHelper.verify(() -> {
LiveInstance.LiveInstanceStatus inMemoryLiveInstanceStatus =
((DefaultMessagingService) participant.getMessagingService()).getExecutor()
@@ -285,7 +287,8 @@ public class TestParticipantFreeze extends ZkTestBase {
// Live instance status is frozen in both memory and zk
verifyLiveInstanceStatus(participant, LiveInstance.LiveInstanceStatus.PAUSED);
// Freeze message is correctly deleted
- Assert.assertNull(_accessor
- .getProperty(_keyBuilder.message(participant.getInstanceName(), freezeMessage.getId())));
+ Assert.assertTrue(TestHelper.verify(() -> !_gZkClient.exists(
+ _keyBuilder.message(participant.getInstanceName(), freezeMessage.getId()).getPath()),
+ TestHelper.WAIT_DURATION));
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java
index 7ddf677..968a5d4 100644
--- a/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java
+++ b/helix-core/src/test/java/org/apache/helix/manager/zk/TestZkHelixAdmin.java
@@ -1087,7 +1087,7 @@ public class TestZkHelixAdmin extends ZkUnitTestBase {
try {
ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
.withClusterName(clusterName)
- .withMode(ClusterManagementMode.Type.CLUSTER_PAUSE)
+ .withMode(ClusterManagementMode.Type.CLUSTER_FREEZE)
.withCancelPendingST(true)
.withReason(methodName)
.build();
@@ -1100,7 +1100,7 @@ public class TestZkHelixAdmin extends ZkUnitTestBase {
ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
.withClusterName(clusterName)
- .withMode(ClusterManagementMode.Type.CLUSTER_PAUSE)
+ .withMode(ClusterManagementMode.Type.CLUSTER_FREEZE)
.withReason(methodName)
.build();
_gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
diff --git a/helix-core/src/test/java/org/apache/helix/model/TestControllerHistoryModel.java b/helix-core/src/test/java/org/apache/helix/model/TestControllerHistoryModel.java
index 2e0bab3..d77287a 100644
--- a/helix-core/src/test/java/org/apache/helix/model/TestControllerHistoryModel.java
+++ b/helix-core/src/test/java/org/apache/helix/model/TestControllerHistoryModel.java
@@ -37,7 +37,7 @@ public class TestControllerHistoryModel {
public void testManagementModeHistory() {
ControllerHistory controllerHistory = new ControllerHistory("HISTORY");
String controller = "controller-0";
- ClusterManagementMode mode = new ClusterManagementMode(ClusterManagementMode.Type.CLUSTER_PAUSE,
+ ClusterManagementMode mode = new ClusterManagementMode(ClusterManagementMode.Type.CLUSTER_FREEZE,
ClusterManagementMode.Status.COMPLETED);
long time = System.currentTimeMillis();
String fromHost = NetworkUtil.getLocalhostName();
diff --git a/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java b/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java
index 18285f5..b8024bb 100644
--- a/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java
+++ b/helix-core/src/test/java/org/apache/helix/model/TestLiveInstance.java
@@ -150,7 +150,9 @@ public class TestLiveInstance extends ZkUnitTestBase {
@Test
public void testLiveInstanceStatus() {
LiveInstance testLiveInstance = new LiveInstance("testLiveInstanceStatus");
- Assert.assertNull(testLiveInstance.getStatus());
+ Assert.assertNull(testLiveInstance.getRecord()
+ .getSimpleField(LiveInstance.LiveInstanceProperty.STATUS.name()));
+ Assert.assertEquals(testLiveInstance.getStatus(), LiveInstance.LiveInstanceStatus.NORMAL);
testLiveInstance.setStatus(LiveInstance.LiveInstanceStatus.PAUSED);
Assert.assertEquals(testLiveInstance.getStatus(), LiveInstance.LiveInstanceStatus.PAUSED);
}
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 9efcdc7..552a530 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
@@ -344,7 +344,7 @@ public class ClusterAccessor extends AbstractHelixResource {
List<LiveInstance> liveInstances = accessor.getChildValues(keyBuilder.liveInstances());
BaseDataAccessor<ZNRecord> baseAccessor = accessor.getBaseDataAccessor();
- if (ClusterManagementMode.Type.CLUSTER_PAUSE.equals(mode.getMode())) {
+ if (ClusterManagementMode.Type.CLUSTER_FREEZE.equals(mode.getMode())) {
// Entering cluster freeze mode, check live instance freeze status and pending ST
for (LiveInstance liveInstance : liveInstances) {
String instanceName = liveInstance.getInstanceName();
@@ -370,7 +370,7 @@ public class ClusterAccessor extends AbstractHelixResource {
participantDetails.put("status", status.name());
participantDetails.put("liveInstancesInProgress", liveInstancesInProgress);
- if (ClusterManagementMode.Type.CLUSTER_PAUSE.equals(mode.getMode())) {
+ if (ClusterManagementMode.Type.CLUSTER_FREEZE.equals(mode.getMode())) {
// Add pending ST result for cluster freeze mode
participantDetails.put("hasPendingStateTransition", hasPendingST);
}
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 8d2de70..8b8f38a 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
@@ -1302,7 +1302,7 @@ public class TestClusterAccessor extends AbstractTestClass {
// Set cluster pause mode
ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
- .withMode(ClusterManagementMode.Type.CLUSTER_PAUSE)
+ .withMode(ClusterManagementMode.Type.CLUSTER_FREEZE)
.withClusterName(cluster)
.build();
String payload = OBJECT_MAPPER.writeValueAsString(request);
@@ -1321,22 +1321,23 @@ public class TestClusterAccessor extends AbstractTestClass {
// Verify get cluster status
String body = get(endpoint, null, Response.Status.OK.getStatusCode(), true);
- ClusterManagementMode mode =
- OBJECT_MAPPER.readerFor(ClusterManagementMode.class).readValue(body);
- Assert.assertEquals(mode.getMode(), ClusterManagementMode.Type.CLUSTER_PAUSE);
+ Map<String, Object> responseMap = OBJECT_MAPPER.readerFor(Map.class).readValue(body);
+ Assert.assertEquals(responseMap.get("mode"), ClusterManagementMode.Type.CLUSTER_FREEZE.name());
// Depending on timing, it could IN_PROGRESS or COMPLETED.
// It's just to verify the rest response format is correct
- Assert.assertTrue(ClusterManagementMode.Status.IN_PROGRESS.equals(mode.getStatus())
- || ClusterManagementMode.Status.COMPLETED.equals(mode.getStatus()));
+ String status = (String) responseMap.get("status");
+ Assert.assertTrue(ClusterManagementMode.Status.IN_PROGRESS.name().equals(status)
+ || ClusterManagementMode.Status.COMPLETED.name().equals(status));
body = get(endpoint, ImmutableMap.of("showDetails", "true"), Response.Status.OK.getStatusCode(),
true);
- Map<String, Object> responseMap = OBJECT_MAPPER.readerFor(Map.class).readValue(body);
+ responseMap = OBJECT_MAPPER.readerFor(Map.class).readValue(body);
Map<String, Object> detailsMap = (Map<String, Object>) responseMap.get("details");
+ status = (String) responseMap.get("status");
Assert.assertEquals(responseMap.get("cluster"), cluster);
- Assert.assertEquals(responseMap.get("mode"), mode.getMode().name());
- Assert.assertEquals(responseMap.get("status"), mode.getStatus().name());
+ Assert.assertEquals(responseMap.get("mode"), ClusterManagementMode.Type.CLUSTER_FREEZE.name());
+ Assert.assertEquals(responseMap.get("status"), status);
Assert.assertTrue(responseMap.containsKey("details"));
Assert.assertTrue(detailsMap.containsKey("cluster"));
Assert.assertTrue(detailsMap.containsKey("liveInstances"));
[helix] 08/13: Add message generation logic for management pipeline
(#1803)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 02c7ac0cdd3e02697fcc126b059f2c7d94e5803b
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Mon Jun 28 22:53:46 2021 -0700
Add message generation logic for management pipeline (#1803)
In cluster freeze mode, controller sends freeze/unfreeze and pending ST cancellation messages to participants.
In this commit, the existing message generation stage is leveraged to create the ST cancellation messages.
Best possible out is built by copying the state map from current state for generating cancellation messages.
For freeze/unfreeze messages, the logic is added in ManagementMessageGenerationPhase.
The existing MessageDispatchStage is also used for dispatching all the messages.
---
.../helix/api/status/ClusterManagementMode.java | 24 ++++
.../helix/controller/GenericHelixController.java | 11 +-
.../helix/controller/stages/AttributeName.java | 3 +
.../stages/ManagementMessageDispatchStage.java | 62 +++++++++
.../stages/ManagementMessageGenerationPhase.java | 119 ++++++++++++++++++
.../controller/stages/ManagementModeStage.java | 22 ++--
.../helix/controller/stages/MessageOutput.java | 11 +-
.../java/org/apache/helix/model/LiveInstance.java | 3 +-
.../java/org/apache/helix/util/RebalanceUtil.java | 27 ++++
.../stages/TestManagementMessageGeneration.java | 140 +++++++++++++++++++++
.../controller/stages/TestManagementModeStage.java | 2 +
11 files changed, 410 insertions(+), 14 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
index cbd2019..86817d1 100644
--- a/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
+++ b/helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
@@ -19,6 +19,8 @@ package org.apache.helix.api.status;
* under the License.
*/
+import org.apache.helix.model.LiveInstance;
+
/**
* Represents the management mode of the cluster:
* 1. what type of mode it targets to be;
@@ -73,4 +75,26 @@ public class ClusterManagementMode {
public Type getMode() {
return mode;
}
+
+ public boolean isFullyInNormalMode() {
+ return Type.NORMAL.equals(mode) && Status.COMPLETED.equals(status);
+ }
+
+ /**
+ * Gets the desired live instance status for this management mode.
+ *
+ * @return The desired {@link org.apache.helix.model.LiveInstance.LiveInstanceStatus}.
+ * If participants status is not expected to change for the management mode, null is returned.
+ */
+ public LiveInstance.LiveInstanceStatus getDesiredParticipantStatus() {
+ switch (mode) {
+ case CLUSTER_PAUSE:
+ return LiveInstance.LiveInstanceStatus.PAUSED;
+ case NORMAL:
+ return LiveInstance.LiveInstanceStatus.NORMAL;
+ default:
+ // Other modes don't need to change participant status
+ return null;
+ }
+ }
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
index 00bdfcc..95416d4 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
@@ -82,6 +82,8 @@ import org.apache.helix.controller.stages.CustomizedViewAggregationStage;
import org.apache.helix.controller.stages.ExternalViewComputeStage;
import org.apache.helix.controller.stages.IntermediateStateCalcStage;
import org.apache.helix.controller.stages.MaintenanceRecoveryStage;
+import org.apache.helix.controller.stages.ManagementMessageDispatchStage;
+import org.apache.helix.controller.stages.ManagementMessageGenerationPhase;
import org.apache.helix.controller.stages.ManagementModeStage;
import org.apache.helix.controller.stages.MessageGenerationPhase;
import org.apache.helix.controller.stages.MessageSelectionStage;
@@ -648,9 +650,16 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
Pipeline dataRefresh = new Pipeline(pipelineName);
dataRefresh.addStage(new ReadClusterDataStage());
+ // data pre-process pipeline
+ Pipeline dataPreprocess = new Pipeline(pipelineName);
+ dataPreprocess.addStage(new ResourceComputationStage());
+ dataPreprocess.addStage(new CurrentStateComputationStage());
+
// cluster management mode process
Pipeline managementMode = new Pipeline(pipelineName);
managementMode.addStage(new ManagementModeStage());
+ managementMode.addStage(new ManagementMessageGenerationPhase());
+ managementMode.addStage(new ManagementMessageDispatchStage());
PipelineRegistry registry = new PipelineRegistry();
Arrays.asList(
@@ -659,7 +668,7 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
ClusterEventType.MessageChange,
ClusterEventType.OnDemandRebalance,
ClusterEventType.PeriodicalRebalance
- ).forEach(type -> registry.register(type, dataRefresh, managementMode));
+ ).forEach(type -> registry.register(type, dataRefresh, dataPreprocess, managementMode));
return registry;
}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/AttributeName.java b/helix-core/src/main/java/org/apache/helix/controller/stages/AttributeName.java
index d9ca40b..8771bff 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/AttributeName.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/AttributeName.java
@@ -45,6 +45,9 @@ public enum AttributeName {
/** This is the cluster manager's session id when event is received. */
EVENT_SESSION,
+ /** Represents cluster's status, used in management mode pipeline. */
+ CLUSTER_STATUS,
+
// This attribute should only be used in TaskGarbageCollectionStage, misuse could cause race conditions.
TO_BE_PURGED_WORKFLOWS,
// This attribute should only be used in TaskGarbageCollectionStage, misuse could cause race conditions.
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageDispatchStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageDispatchStage.java
new file mode 100644
index 0000000..ca7aa31
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageDispatchStage.java
@@ -0,0 +1,62 @@
+package org.apache.helix.controller.stages;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.List;
+
+import org.apache.helix.HelixManager;
+import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.controller.LogUtil;
+import org.apache.helix.controller.dataproviders.ManagementControllerDataProvider;
+import org.apache.helix.model.Message;
+import org.apache.helix.util.RebalanceUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Dispatches participant status change and pending state transition cancellation messages
+ * for the management pipeline.
+ */
+public class ManagementMessageDispatchStage extends MessageDispatchStage {
+ private static final Logger LOG = LoggerFactory.getLogger(ManagementMessageDispatchStage.class);
+
+ @Override
+ public void process(ClusterEvent event) throws Exception {
+ MessageOutput messageOutput =
+ event.getAttribute(AttributeName.MESSAGES_ALL.name());
+ processEvent(event, messageOutput);
+
+ // Send participant status change messages.
+ HelixManager manager = event.getAttribute(AttributeName.helixmanager.name());
+ List<Message> messagesSent =
+ super.sendMessages(manager.getHelixDataAccessor(), messageOutput.getStatusChangeMessages());
+ ManagementControllerDataProvider cache =
+ event.getAttribute(AttributeName.ControllerDataProvider.name());
+ cache.cacheMessages(messagesSent);
+
+ // Can exit management mode pipeline after fully in normal mode
+ ClusterManagementMode managementMode = event.getAttribute(AttributeName.CLUSTER_STATUS.name());
+ if (managementMode.isFullyInNormalMode()) {
+ LogUtil.logInfo(LOG, _eventId,
+ "Exiting management mode pipeline for cluster " + event.getClusterName());
+ RebalanceUtil.enableManagementMode(event.getClusterName(), false);
+ }
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
new file mode 100644
index 0000000..4479d50
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
@@ -0,0 +1,119 @@
+package org.apache.helix.controller.stages;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.helix.HelixManager;
+import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.controller.LogUtil;
+import org.apache.helix.controller.dataproviders.ManagementControllerDataProvider;
+import org.apache.helix.controller.pipeline.StageException;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.LiveInstance.LiveInstanceStatus;
+import org.apache.helix.model.Message;
+import org.apache.helix.model.PauseSignal;
+import org.apache.helix.util.MessageUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Generates participant status change (freeze/unfreeze) and pending state transition cancellation
+ * messages for management mode pipeline.
+ */
+public class ManagementMessageGenerationPhase extends MessageGenerationPhase {
+ private static final Logger LOG = LoggerFactory.getLogger(ManagementMessageGenerationPhase.class);
+
+ @Override
+ public void process(ClusterEvent event) throws Exception {
+ _eventId = event.getEventId();
+ String clusterName = event.getClusterName();
+ HelixManager manager = event.getAttribute(AttributeName.helixmanager.name());
+ ClusterManagementMode managementMode = event.getAttribute(AttributeName.CLUSTER_STATUS.name());
+ ManagementControllerDataProvider cache =
+ event.getAttribute(AttributeName.ControllerDataProvider.name());
+ if (manager == null || managementMode == null || cache == null) {
+ throw new StageException("Missing attributes in event: " + event
+ + ". Requires HelixManager|ClusterStatus|DataCache");
+ }
+
+ PauseSignal pauseSignal = cache.getPauseSignal();
+ if (cache.getClusterConfig().isStateTransitionCancelEnabled()
+ && pauseSignal != null && pauseSignal.getCancelPendingST()) {
+ // Generate ST cancellation messages.
+ LogUtil.logInfo(LOG, _eventId,
+ "Generating ST cancellation messages for cluster " + clusterName);
+ super.process(event);
+ }
+
+ MessageOutput messageOutput =
+ event.getAttributeWithDefault(AttributeName.MESSAGES_ALL.name(), new MessageOutput());
+ // Is participant status change still in progress? Create messages
+ if (!ClusterManagementMode.Status.COMPLETED.equals(managementMode.getStatus())) {
+ LogUtil.logInfo(LOG, _eventId, "Generating messages as cluster " + clusterName
+ + " is still in progress to change participant status");
+ List<Message> messages = generateStatusChangeMessages(managementMode,
+ cache.getEnabledLiveInstances(), cache.getLiveInstances(),
+ cache.getAllInstancesMessages(), manager.getInstanceName(), manager.getSessionId());
+ messageOutput.addStatusChangeMessages(messages);
+ }
+
+ event.addAttribute(AttributeName.MESSAGES_ALL.name(), messageOutput);
+ }
+
+ private List<Message> generateStatusChangeMessages(ClusterManagementMode managementMode,
+ Set<String> enabledLiveInstances, Map<String, LiveInstance> liveInstanceMap,
+ Map<String, Collection<Message>> allInstanceMessages, String managerInstance,
+ String managerSessionId) {
+ List<Message> messagesGenerated = new ArrayList<>();
+
+ LiveInstanceStatus desiredStatus = managementMode.getDesiredParticipantStatus();
+
+ // Check status and pending status change messages for all enabled live instances.
+ // Send freeze/unfreeze messages if necessary
+ for (String instanceName : enabledLiveInstances) {
+ LiveInstance liveInstance = liveInstanceMap.get(instanceName);
+ Collection<Message> pendingMessages = allInstanceMessages.get(instanceName);
+ String sessionId = liveInstance.getEphemeralOwner();
+ LiveInstanceStatus currentStatus = liveInstance.getStatus();
+
+ if (needStatusChangeMessage(pendingMessages, currentStatus, desiredStatus)) {
+ Message statusChangeMessage = MessageUtil.createStatusChangeMessage(currentStatus,
+ desiredStatus, managerInstance, managerSessionId, instanceName, sessionId);
+ messagesGenerated.add(statusChangeMessage);
+ }
+ }
+
+ return messagesGenerated;
+ }
+
+ private boolean needStatusChangeMessage(Collection<Message> messages,
+ LiveInstanceStatus currentStatus, LiveInstanceStatus desiredStatus) {
+ // 1. current status is not equal to desired status
+ // 2. participant change status message is not sent
+ return currentStatus != desiredStatus && messages.stream().noneMatch(
+ message -> message.isParticipantStatusChangeType() && desiredStatus.name()
+ .equals(message.getToState()));
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
index 94ff1d3..b639685 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
@@ -40,7 +40,9 @@ import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.LiveInstance.LiveInstanceStatus;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.MessageType;
+import org.apache.helix.model.Partition;
import org.apache.helix.model.PauseSignal;
+import org.apache.helix.model.Resource;
import org.apache.helix.util.HelixUtil;
import org.apache.helix.util.RebalanceUtil;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
@@ -66,6 +68,14 @@ public class ManagementModeStage extends AbstractBaseStage {
HelixDataAccessor accessor = manager.getHelixDataAccessor();
ManagementControllerDataProvider cache =
event.getAttribute(AttributeName.ControllerDataProvider.name());
+ CurrentStateOutput currentStateOutput =
+ event.getAttribute(AttributeName.CURRENT_STATE.name());
+ final Map<String, Resource> resourceMap =
+ event.getAttribute(AttributeName.RESOURCES_TO_REBALANCE.name());
+
+ final BestPossibleStateOutput bestPossibleStateOutput =
+ RebalanceUtil.buildBestPossibleState(resourceMap.keySet(), currentStateOutput);
+ event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), bestPossibleStateOutput);
ClusterManagementMode managementMode =
checkClusterFreezeStatus(cache.getEnabledLiveInstances(), cache.getLiveInstances(),
@@ -75,17 +85,7 @@ public class ManagementModeStage extends AbstractBaseStage {
recordManagementModeHistory(managementMode, cache.getPauseSignal(), manager.getInstanceName(),
accessor);
- // TODO: move to the last stage of management pipeline
- checkInManagementMode(clusterName, cache);
- }
-
- private void checkInManagementMode(String clusterName, ManagementControllerDataProvider cache) {
- // Should exit management mode
- if (!HelixUtil.inManagementMode(cache.getPauseSignal(), cache.getLiveInstances(),
- cache.getEnabledLiveInstances(), cache.getAllInstancesMessages())) {
- LogUtil.logInfo(LOG, _eventId, "Exiting management mode pipeline for cluster " + clusterName);
- RebalanceUtil.enableManagementMode(clusterName, false);
- }
+ event.addAttribute(AttributeName.CLUSTER_STATUS.name(), managementMode);
}
// Checks cluster freeze, controller pause mode and status.
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageOutput.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageOutput.java
index ad7e6c8..1f47ee4 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageOutput.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageOutput.java
@@ -30,10 +30,11 @@ import org.apache.helix.model.Partition;
public class MessageOutput {
private final Map<String, Map<Partition, List<Message>>> _messagesMap;
+ private final List<Message> _statusChangeMessages;
public MessageOutput() {
_messagesMap = new HashMap<>();
-
+ _statusChangeMessages = new ArrayList<>();
}
public void addMessage(String resourceName, Partition partition, Message message) {
@@ -56,6 +57,10 @@ public class MessageOutput {
_messagesMap.get(resourceName).put(partition, messages);
}
+ public void addStatusChangeMessages(List<Message> messages) {
+ _statusChangeMessages.addAll(messages);
+ }
+
public List<Message> getMessages(String resourceName, Partition resource) {
Map<Partition, List<Message>> map = _messagesMap.get(resourceName);
if (map != null && map.get(resource) != null) {
@@ -68,6 +73,10 @@ public class MessageOutput {
return _messagesMap.getOrDefault(resourceName, Collections.emptyMap());
}
+ public List<Message> getStatusChangeMessages() {
+ return _statusChangeMessages;
+ }
+
@Override
public String toString() {
return _messagesMap.toString();
diff --git a/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java b/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
index 7a9671d..fed2940 100644
--- a/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
+++ b/helix-core/src/main/java/org/apache/helix/model/LiveInstance.java
@@ -50,7 +50,8 @@ public class LiveInstance extends HelixProperty {
* Saved values for the {@link LiveInstanceProperty#STATUS} field
*/
public enum LiveInstanceStatus {
- PAUSED
+ PAUSED,
+ NORMAL
}
/**
diff --git a/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java b/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java
index f74b98f..bc4054a 100644
--- a/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/util/RebalanceUtil.java
@@ -20,6 +20,7 @@ package org.apache.helix.util;
*/
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -30,8 +31,11 @@ import org.apache.helix.HelixException;
import org.apache.helix.controller.GenericHelixController;
import org.apache.helix.controller.pipeline.Stage;
import org.apache.helix.controller.pipeline.StageContext;
+import org.apache.helix.controller.stages.BestPossibleStateOutput;
import org.apache.helix.controller.stages.ClusterEvent;
+import org.apache.helix.controller.stages.CurrentStateOutput;
import org.apache.helix.model.IdealState;
+import org.apache.helix.model.Partition;
import org.apache.helix.model.StateModelDefinition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -192,6 +196,29 @@ public class RebalanceUtil {
}
/**
+ * Build best possible state out by copying the state map from current state output.
+ * It'll be used for generating pending ST cancellation messages.
+ *
+ * @param resourceNames collection of resource names
+ * @param currentStateOutput Current state output {@link CurrentStateOutput}
+ * @return {@link BestPossibleStateOutput}
+ */
+ public static BestPossibleStateOutput buildBestPossibleState(Collection<String> resourceNames,
+ CurrentStateOutput currentStateOutput) {
+ BestPossibleStateOutput output = new BestPossibleStateOutput();
+
+ for (String resource : resourceNames) {
+ Map<Partition, Map<String, String>> currentStateMap =
+ currentStateOutput.getCurrentStateMap(resource);
+ if (currentStateMap != null) {
+ output.setState(resource, currentStateMap);
+ }
+ }
+
+ return output;
+ }
+
+ /**
* runStage allows the run of individual stages. It can be used to mock a part of the Controller
* pipeline run.
*
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementMessageGeneration.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementMessageGeneration.java
new file mode 100644
index 0000000..c16b503
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementMessageGeneration.java
@@ -0,0 +1,140 @@
+package org.apache.helix.controller.stages;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.helix.HelixManager;
+import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.controller.dataproviders.ManagementControllerDataProvider;
+import org.apache.helix.model.ClusterConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.Message;
+import org.apache.helix.model.Partition;
+import org.apache.helix.model.PauseSignal;
+import org.apache.helix.model.Resource;
+import org.apache.helix.model.StateModelDefinition;
+import org.apache.helix.task.TaskConstants;
+import org.apache.helix.util.RebalanceUtil;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestManagementMessageGeneration extends ManagementMessageGenerationPhase {
+ private static final String TEST_CLUSTER = "testCluster";
+ private static final String TEST_RESOURCE = "resource0";
+ private static final String TEST_INSTANCE = "instance0";
+ private static final String TEST_PARTITION = "partition0";
+
+ @Test
+ public void testCancelPendingSTMessage() throws Exception {
+ List<Message> messages = generateMessages("ONLINE", "ONLINE", "OFFLINE", true);
+
+ Assert.assertEquals(messages.size(), 1, "Should create cancellation message");
+
+ Message msg = messages.get(0);
+ Assert.assertEquals(msg.getMsgType(), Message.MessageType.STATE_TRANSITION_CANCELLATION.name());
+ Assert.assertEquals(msg.getFromState(), "ONLINE");
+ Assert.assertEquals(msg.getToState(), "OFFLINE");
+
+ messages = generateMessages("ONLINE", "ONLINE", "OFFLINE", false);
+ Assert.assertEquals(messages.size(), 0);
+ }
+
+ private List<Message> generateMessages(String currentState, String fromState, String toState,
+ boolean cancelPendingST) throws Exception {
+ ClusterEvent event = new ClusterEvent(TEST_CLUSTER, ClusterEventType.Unknown);
+
+ // Set current state to event
+ CurrentStateOutput currentStateOutput = mock(CurrentStateOutput.class);
+ Partition partition = mock(Partition.class);
+ when(partition.getPartitionName()).thenReturn(TEST_PARTITION);
+ when(currentStateOutput.getCurrentState(TEST_RESOURCE, partition, TEST_INSTANCE))
+ .thenReturn(currentState);
+ when(currentStateOutput.getCurrentStateMap(TEST_RESOURCE))
+ .thenReturn(ImmutableMap.of(partition, ImmutableMap.of(TEST_INSTANCE, currentState)));
+
+ // Pending message for error partition reset
+ Message pendingMessage = mock(Message.class);
+ when(pendingMessage.getFromState()).thenReturn(fromState);
+ when(pendingMessage.getToState()).thenReturn(toState);
+ when(currentStateOutput.getPendingMessage(TEST_RESOURCE, partition, TEST_INSTANCE))
+ .thenReturn(pendingMessage);
+ event.addAttribute(AttributeName.CURRENT_STATE.name(), currentStateOutput);
+
+ // Set helix manager to event
+ event.addAttribute(AttributeName.helixmanager.name(), mock(HelixManager.class));
+
+ StateModelDefinition stateModelDefinition = new StateModelDefinition.Builder("TestStateModel")
+ .addState("ONLINE", 1).addState("OFFLINE")
+ .addState("DROPPED").addState("ERROR")
+ .initialState("OFFLINE")
+ .addTransition("ERROR", "OFFLINE", 1).addTransition("ONLINE", "OFFLINE", 2)
+ .addTransition("OFFLINE", "DROPPED", 3).addTransition("OFFLINE", "ONLINE", 4)
+ .build();
+
+ // Set controller data provider to event
+ ManagementControllerDataProvider cache = mock(ManagementControllerDataProvider.class);
+ when(cache.getStateModelDef(TaskConstants.STATE_MODEL_NAME)).thenReturn(stateModelDefinition);
+ Map<String, LiveInstance> liveInstances = mock(Map.class);
+ LiveInstance mockLiveInstance = mock(LiveInstance.class);
+ when(mockLiveInstance.getInstanceName()).thenReturn(TEST_INSTANCE);
+ when(mockLiveInstance.getEphemeralOwner()).thenReturn("TEST");
+ when(liveInstances.values()).thenReturn(Collections.singletonList(mockLiveInstance));
+ when(cache.getLiveInstances()).thenReturn(liveInstances);
+ ClusterConfig clusterConfig = mock(ClusterConfig.class);
+ when(cache.getClusterConfig()).thenReturn(clusterConfig);
+ when(clusterConfig.isStateTransitionCancelEnabled()).thenReturn(cancelPendingST);
+ PauseSignal pauseSignal = mock(PauseSignal.class);
+ when(pauseSignal.getCancelPendingST()).thenReturn(true);
+ when(cache.getPauseSignal()).thenReturn(pauseSignal);
+ event.addAttribute(AttributeName.ControllerDataProvider.name(), cache);
+
+ // Set event attribute: resources to rebalance
+ Map<String, Resource> resourceMap = new HashMap<>();
+ Resource resource = mock(Resource.class);
+ when(resource.getResourceName()).thenReturn(TEST_RESOURCE);
+ List<Partition> partitions = Collections.singletonList(partition);
+ when(resource.getPartitions()).thenReturn(partitions);
+ when(resource.getStateModelDefRef()).thenReturn(TaskConstants.STATE_MODEL_NAME);
+ resourceMap.put(TEST_RESOURCE, resource);
+ event.addAttribute(AttributeName.RESOURCES_TO_REBALANCE.name(), resourceMap);
+
+ // set up resource state map
+ BestPossibleStateOutput bestPossibleStateOutput =
+ RebalanceUtil.buildBestPossibleState(resourceMap.keySet(), currentStateOutput);
+
+ // Process the event
+ ClusterManagementMode mode = new ClusterManagementMode(ClusterManagementMode.Type.CLUSTER_PAUSE,
+ ClusterManagementMode.Status.IN_PROGRESS);
+ event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), bestPossibleStateOutput);
+ event.addAttribute(AttributeName.CLUSTER_STATUS.name(), mode);
+ process(event);
+ MessageOutput output = event.getAttribute(AttributeName.MESSAGES_ALL.name());
+
+ return output.getMessages(TEST_RESOURCE, partition);
+ }
+}
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
index 28ca524..2737bf7 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestManagementModeStage.java
@@ -83,6 +83,8 @@ public class TestManagementModeStage extends ZkTestBase {
Pipeline dataRefresh = new Pipeline();
dataRefresh.addStage(new ReadClusterDataStage());
+ dataRefresh.addStage(new ResourceComputationStage());
+ dataRefresh.addStage(new CurrentStateComputationStage());
runPipeline(event, dataRefresh, false);
ManagementModeStage managementModeStage = new ManagementModeStage();
managementModeStage.process(event);
[helix] 13/13: Rename property CLUSTER_PAUSE to CLUSTER_FREEZE in
PauseSignal (#1820)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit ac877c5c631c12ad570969e897528824625a0461
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Thu Jul 15 20:26:11 2021 -0700
Rename property CLUSTER_PAUSE to CLUSTER_FREEZE in PauseSignal (#1820)
Rename property CLUSTER_PAUSE to CLUSTER_FREEZE in PauseSignal.
---
.../helix/controller/stages/ManagementModeStage.java | 16 +++++++++++-----
.../main/java/org/apache/helix/model/PauseSignal.java | 6 +++---
.../integration/controller/TestClusterFreezeMode.java | 6 +++++-
3 files changed, 19 insertions(+), 9 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
index a281e08..3329d8c 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
@@ -24,6 +24,7 @@ import java.util.Collection;
import java.util.Map;
import java.util.Set;
+import com.google.common.collect.ImmutableSet;
import org.apache.helix.AccessOption;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
@@ -53,11 +54,17 @@ import org.slf4j.LoggerFactory;
public class ManagementModeStage extends AbstractBaseStage {
private static final Logger LOG = LoggerFactory.getLogger(ManagementModeStage.class);
+ // Check message types to determine instance is in progress to be frozen
+ private static final Set<String> PENDING_MESSAGE_TYPES = ImmutableSet.of(
+ MessageType.PARTICIPANT_STATUS_CHANGE.name(),
+ MessageType.STATE_TRANSITION.name(),
+ MessageType.STATE_TRANSITION_CANCELLATION.name()
+ );
+
@Override
public void process(ClusterEvent event) throws Exception {
// TODO: implement the stage
_eventId = event.getEventId();
- String clusterName = event.getClusterName();
HelixManager manager = event.getAttribute(AttributeName.helixmanager.name());
if (manager == null) {
throw new StageException("HelixManager attribute value is null");
@@ -120,13 +127,12 @@ public class ManagementModeStage extends AbstractBaseStage {
// 2. No pending participant status change message.
return enabledLiveInstances.stream().noneMatch(
instance -> !LiveInstanceStatus.FROZEN.equals(liveInstanceMap.get(instance).getStatus())
- || hasPendingMessage(allInstanceMessages.get(instance),
- MessageType.PARTICIPANT_STATUS_CHANGE));
+ || hasPendingMessage(allInstanceMessages.get(instance)));
}
- private boolean hasPendingMessage(Collection<Message> messages, MessageType type) {
+ private boolean hasPendingMessage(Collection<Message> messages) {
return messages != null && messages.stream()
- .anyMatch(message -> type.name().equals(message.getMsgType()));
+ .anyMatch(message -> PENDING_MESSAGE_TYPES.contains(message.getMsgType()));
}
private void recordClusterStatus(ClusterManagementMode mode, HelixDataAccessor accessor) {
diff --git a/helix-core/src/main/java/org/apache/helix/model/PauseSignal.java b/helix-core/src/main/java/org/apache/helix/model/PauseSignal.java
index bcfd17d..65674ad 100644
--- a/helix-core/src/main/java/org/apache/helix/model/PauseSignal.java
+++ b/helix-core/src/main/java/org/apache/helix/model/PauseSignal.java
@@ -32,7 +32,7 @@ public class PauseSignal extends HelixProperty {
public enum PauseSignalProperty {
REASON,
- CLUSTER_PAUSE,
+ CLUSTER_FREEZE,
FROM_HOST,
CANCEL_PENDING_ST,
TRIGGER_TIME
@@ -76,11 +76,11 @@ public class PauseSignal extends HelixProperty {
}
public void setClusterPause(boolean pause) {
- _record.setBooleanField(PauseSignalProperty.CLUSTER_PAUSE.name(), pause);
+ _record.setBooleanField(PauseSignalProperty.CLUSTER_FREEZE.name(), pause);
}
public boolean isClusterPause() {
- return _record.getBooleanField(PauseSignalProperty.CLUSTER_PAUSE.name(), false);
+ return _record.getBooleanField(PauseSignalProperty.CLUSTER_FREEZE.name(), false);
}
public void setFromHost(String host) {
diff --git a/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java b/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
index 09de4dc..e068f38 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
@@ -158,11 +158,15 @@ public class TestClusterFreezeMode extends ZkTestBase {
.build();
_gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
+ // Wait for all live instances are marked as frozen
+ verifyLiveInstanceStatus(_participants, LiveInstance.LiveInstanceStatus.FROZEN);
+
// Pending ST message exists
Assert.assertTrue(
_gZkClient.exists(keyBuilder.message(message.getTgtName(), message.getMsgId()).getPath()));
- // Cluster is in progress to cluster pause because there is a pending state transition message
+ // Even live instance status is marked as frozen, Cluster is in progress to cluster freeze
+ // because there is a pending state transition message
ClusterStatus expectedClusterStatus = new ClusterStatus();
expectedClusterStatus.setManagementMode(ClusterManagementMode.Type.CLUSTER_FREEZE);
expectedClusterStatus.setManagementModeStatus(ClusterManagementMode.Status.IN_PROGRESS);
[helix] 10/13: Add integration tests for cluster freeze mode (#1816)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 849ab1109c2e5f21ebb16b6b254a4d30a1a365e4
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Wed Jul 14 15:10:33 2021 -0700
Add integration tests for cluster freeze mode (#1816)
Integration tests are needed to cover the scenarios for the cluster freeze mode.
This commit adds integration tests for cluster freeze mode:
a. freeze cluster when there are pending state transition messages
b. handle new session when frozen - it should be reset and then frozen
c. restart participants when frozen - it should be reset and then frozen
d. reset partition when frozen
f. unfreeze cluster
---
.../stages/ManagementMessageGenerationPhase.java | 4 +-
.../controller/TestClusterFreezeMode.java | 372 +++++++++++++++++++++
2 files changed, 375 insertions(+), 1 deletion(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
index 4479d50..c795deb 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
@@ -96,7 +96,9 @@ public class ManagementMessageGenerationPhase extends MessageGenerationPhase {
LiveInstance liveInstance = liveInstanceMap.get(instanceName);
Collection<Message> pendingMessages = allInstanceMessages.get(instanceName);
String sessionId = liveInstance.getEphemeralOwner();
- LiveInstanceStatus currentStatus = liveInstance.getStatus();
+ LiveInstanceStatus liveInstanceStatus = liveInstance.getStatus();
+ LiveInstanceStatus currentStatus = (liveInstanceStatus == null
+ ? LiveInstanceStatus.NORMAL : liveInstanceStatus);
if (needStatusChangeMessage(pendingMessages, currentStatus, desiredStatus)) {
Message statusChangeMessage = MessageUtil.createStatusChangeMessage(currentStatus,
diff --git a/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java b/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
new file mode 100644
index 0000000..cbc8535
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterFreezeMode.java
@@ -0,0 +1,372 @@
+package org.apache.helix.integration.controller;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+
+import org.apache.helix.HelixConstants;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.TestHelper;
+import org.apache.helix.api.status.ClusterManagementMode;
+import org.apache.helix.api.status.ClusterManagementModeRequest;
+import org.apache.helix.common.ZkTestBase;
+import org.apache.helix.integration.manager.ClusterControllerManager;
+import org.apache.helix.integration.manager.ClusterManager;
+import org.apache.helix.integration.manager.MockParticipantManager;
+import org.apache.helix.mock.participant.ErrTransition;
+import org.apache.helix.mock.participant.MockTransition;
+import org.apache.helix.model.ClusterStatus;
+import org.apache.helix.model.ControllerHistory;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.Message;
+import org.apache.helix.model.PauseSignal;
+import org.apache.helix.model.Resource;
+import org.apache.helix.model.StateModelDefinition;
+import org.apache.helix.tools.ClusterStateVerifier;
+import org.apache.helix.util.MessageUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+public class TestClusterFreezeMode extends ZkTestBase {
+ private HelixManager _manager;
+ private HelixDataAccessor _accessor;
+ private String _clusterName;
+ private int _numNodes;
+ private MockParticipantManager[] _participants;
+ private ClusterControllerManager _controller;
+
+ @BeforeClass
+ public void beforeClass() throws Exception {
+ _numNodes = 3;
+ _clusterName = "CLUSTER_" + TestHelper.getTestClassName();
+ _participants = new MockParticipantManager[_numNodes];
+ TestHelper.setupCluster(_clusterName, ZK_ADDR, 12918, // participant port
+ "localhost", // participant name prefix
+ "TestDB", // resource name prefix
+ 1, // resources
+ 2, // partitions per resource
+ _numNodes, // number of nodes
+ 3, // replicas
+ "MasterSlave", true);
+
+ _manager = HelixManagerFactory
+ .getZKHelixManager(_clusterName, "Admin", InstanceType.ADMINISTRATOR, ZK_ADDR);
+ _manager.connect();
+ _accessor = _manager.getHelixDataAccessor();
+
+ // start controller
+ _controller = new ClusterControllerManager(ZK_ADDR, _clusterName, "controller_0");
+ _controller.syncStart();
+
+ Map<String, Set<String>> errPartitions = new HashMap<String, Set<String>>() {
+ {
+ put("OFFLINE-SLAVE", TestHelper.setOf("TestDB0_0"));
+ }
+ };
+
+ // start participants
+ for (int i = 0; i < _numNodes; i++) {
+ String instanceName = "localhost_" + (12918 + i);
+ _participants[i] = new MockParticipantManager(ZK_ADDR, _clusterName, instanceName);
+ if (i == 0) {
+ // Make TestDB0_0 be error state on participant_0
+ _participants[i].setTransition(new ErrTransition(errPartitions));
+ }
+ _participants[i].syncStart();
+ }
+
+ boolean result = ClusterStateVerifier.verifyByZkCallback(
+ new ClusterStateVerifier.BestPossAndExtViewZkVerifier(ZK_ADDR, _clusterName));
+ Assert.assertTrue(result);
+ }
+
+ @AfterClass
+ public void afterClass() {
+ _manager.disconnect();
+ _controller.syncStop();
+ Arrays.stream(_participants).forEach(ClusterManager::syncStop);
+ deleteCluster(_clusterName);
+ }
+
+ /*
+ * Tests below scenarios:
+ * 1. cluster is in progress to freeze mode if there is a pending state transition message;
+ * 2. after state transition is completed, cluster freeze mode is completed
+ *
+ * Also tests cluster status and management mode history recording.
+ */
+ @Test
+ public void testEnableFreezeMode() throws Exception {
+ String methodName = TestHelper.getTestMethodName();
+ // Not in freeze mode
+ PropertyKey.Builder keyBuilder = _accessor.keyBuilder();
+ PauseSignal pauseSignal = _accessor.getProperty(keyBuilder.pause());
+ Assert.assertNull(pauseSignal);
+
+ // Block state transition for participants[1]
+ CountDownLatch latch = new CountDownLatch(1);
+ _participants[1].setTransition(new BlockingTransition(latch));
+
+ // Send a state transition message to participants[1]
+ Resource resource = new Resource("TestDB0");
+ resource.setStateModelFactoryName(HelixConstants.DEFAULT_STATE_MODEL_FACTORY);
+ Message message = MessageUtil
+ .createStateTransitionMessage(_manager.getInstanceName(), _manager.getSessionId(), resource,
+ "TestDB0_1", _participants[1].getInstanceName(), "SLAVE", "OFFLINE",
+ _participants[1].getSessionId(), "MasterSlave");
+ Assert.assertTrue(_accessor
+ .updateProperty(keyBuilder.message(message.getTgtName(), message.getMsgId()), message));
+
+ // Freeze cluster
+ ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
+ .withClusterName(_clusterName)
+ .withMode(ClusterManagementMode.Type.CLUSTER_PAUSE)
+ .withReason(methodName)
+ .build();
+ _gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
+
+ // Pending ST message exists
+ Assert.assertTrue(
+ _gZkClient.exists(keyBuilder.message(message.getTgtName(), message.getMsgId()).getPath()));
+
+ // Cluster is in progress to cluster pause because there is a pending state transition message
+ ClusterStatus expectedClusterStatus = new ClusterStatus();
+ expectedClusterStatus.setManagementMode(ClusterManagementMode.Type.CLUSTER_PAUSE);
+ expectedClusterStatus.setManagementModeStatus(ClusterManagementMode.Status.IN_PROGRESS);
+ verifyClusterStatus(expectedClusterStatus);
+
+ // Unblock to finish state transition and delete the ST message
+ latch.countDown();
+
+ // Verify live instance status and cluster status
+ verifyLiveInstanceStatus(_participants, LiveInstance.LiveInstanceStatus.PAUSED);
+
+ expectedClusterStatus = new ClusterStatus();
+ expectedClusterStatus.setManagementMode(ClusterManagementMode.Type.CLUSTER_PAUSE);
+ expectedClusterStatus.setManagementModeStatus(ClusterManagementMode.Status.COMPLETED);
+ verifyClusterStatus(expectedClusterStatus);
+
+ // Verify management mode history
+ Assert.assertTrue(TestHelper.verify(() -> {
+ ControllerHistory history = _accessor.getProperty(keyBuilder.controllerLeaderHistory());
+ List<String> managementHistory = history.getManagementModeHistory();
+ if (managementHistory == null || managementHistory.isEmpty()) {
+ return false;
+ }
+ String lastHistory = managementHistory.get(managementHistory.size() - 1);
+ return lastHistory.contains("MODE=" + ClusterManagementMode.Type.CLUSTER_PAUSE)
+ && lastHistory.contains("STATUS=" + ClusterManagementMode.Status.COMPLETED)
+ && lastHistory.contains("REASON=" + methodName);
+ }, TestHelper.WAIT_DURATION));
+ }
+
+ @Test(dependsOnMethods = "testEnableFreezeMode")
+ public void testNewLiveInstanceAddedWhenFrozen() throws Exception {
+ // Add a new live instance. Simulate an instance is rebooted and back to online
+ String newInstanceName = "localhost_" + (12918 + _numNodes + 1);
+ _gSetupTool.addInstancesToCluster(_clusterName, new String[]{newInstanceName});
+ MockParticipantManager newParticipant =
+ new MockParticipantManager(ZK_ADDR, _clusterName, newInstanceName);
+ newParticipant.syncStart();
+
+ // The new participant/live instance should be frozen by controller
+ verifyLiveInstanceStatus(new MockParticipantManager[]{newParticipant},
+ LiveInstance.LiveInstanceStatus.PAUSED);
+
+ newParticipant.syncStop();
+ }
+
+ // Simulates instance is restarted and the in-memory status is gone.
+ // When instance comes back alive, it'll reset state model, carry over
+ // and set current state to init state.
+ @Test(dependsOnMethods = "testNewLiveInstanceAddedWhenFrozen")
+ public void testRestartParticipantWhenFrozen() throws Exception {
+ String instanceName = _participants[1].getInstanceName();
+ PropertyKey.Builder keyBuilder = _accessor.keyBuilder();
+ List<CurrentState> originCurStates = _accessor
+ .getChildValues(keyBuilder.currentStates(instanceName, _participants[1].getSessionId()),
+ false);
+ String oldSession = _participants[1].getSessionId();
+
+ // Restart participants[1]
+ _participants[1].syncStop();
+ _participants[1] = new MockParticipantManager(ZK_ADDR, _participants[1].getClusterName(),
+ instanceName);
+ _participants[1].syncStart();
+
+ Assert.assertTrue(TestHelper.verify(() ->
+ _gZkClient.exists(keyBuilder.liveInstance(instanceName).getPath()),
+ TestHelper.WAIT_DURATION));
+ LiveInstance liveInstance = _accessor.getProperty(keyBuilder.liveInstance(instanceName));
+
+ // New live instance ephemeral node
+ Assert.assertEquals(liveInstance.getEphemeralOwner(), _participants[1].getSessionId());
+ // Status is frozen because controller sends a freeze message.
+ verifyLiveInstanceStatus(new MockParticipantManager[]{_participants[1]},
+ LiveInstance.LiveInstanceStatus.PAUSED);
+
+ // Old session current state is deleted because of current state carry-over
+ Assert.assertTrue(TestHelper.verify(
+ () -> !_gZkClient.exists(keyBuilder.currentStates(instanceName, oldSession).getPath()),
+ TestHelper.WAIT_DURATION));
+
+ // Current states are set to init states (OFFLINE)
+ List<CurrentState> curStates = _accessor
+ .getChildValues(keyBuilder.currentStates(instanceName, _participants[1].getSessionId()),
+ false);
+ Assert.assertEquals(curStates.size(), 1);
+ Assert.assertTrue(TestHelper.verify(() -> {
+ for (CurrentState cs : originCurStates) {
+ String stateModelDefRef = cs.getStateModelDefRef();
+ for (String partition : cs.getPartitionStateMap().keySet()) {
+ StateModelDefinition stateModelDef =
+ _accessor.getProperty(keyBuilder.stateModelDef(stateModelDefRef));
+ String initState = stateModelDef.getInitialState();
+ if (!initState.equals(curStates.get(0).getPartitionStateMap().get(partition))) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }, TestHelper.WAIT_DURATION));
+ }
+
+ // Partition reset is allowed when cluster is frozen
+ @Test(dependsOnMethods = "testRestartParticipantWhenFrozen")
+ public void testResetPartitionWhenFrozen() throws Exception {
+ String instanceName = _participants[0].getInstanceName();
+ // Remove errTransition
+ _participants[0].setTransition(null);
+ _gSetupTool.getClusterManagementTool().resetPartition(_clusterName, instanceName, "TestDB0",
+ Collections.singletonList("TestDB0_0"));
+
+ // Error partition is reset: ERROR -> OFFLINE
+ Assert.assertTrue(TestHelper.verify(() -> {
+ CurrentState currentState = _accessor.getProperty(_accessor.keyBuilder()
+ .currentState(instanceName, _participants[0].getSessionId(), "TestDB0"));
+ return "OFFLINE".equals(currentState.getPartitionStateMap().get("TestDB0_0"));
+ }, TestHelper.WAIT_DURATION));
+ }
+
+ @Test(dependsOnMethods = "testResetPartitionWhenFrozen")
+ public void testCreateResourceWhenFrozen() {
+ // Add a new resource
+ _gSetupTool.addResourceToCluster(_clusterName, "TestDB1", 2, "MasterSlave");
+ _gSetupTool.rebalanceStorageCluster(_clusterName, "TestDB1", 3);
+
+ // TestDB1 external view is empty
+ TestHelper.verifyWithTimeout("verifyEmptyCurStateAndExtView", 1000, _clusterName, "TestDB1",
+ TestHelper.setOf("localhost_12918", "localhost_12919", "localhost_12920"), ZK_ADDR);
+ }
+
+ @Test(dependsOnMethods = "testCreateResourceWhenFrozen")
+ public void testUnfreezeCluster() throws Exception {
+ String methodName = TestHelper.getTestMethodName();
+ // Unfreeze cluster
+ ClusterManagementModeRequest request = ClusterManagementModeRequest.newBuilder()
+ .withClusterName(_clusterName)
+ .withMode(ClusterManagementMode.Type.NORMAL)
+ .withReason(methodName)
+ .build();
+ _gSetupTool.getClusterManagementTool().setClusterManagementMode(request);
+
+ verifyLiveInstanceStatus(_participants, null);
+
+ ClusterStatus expectedClusterStatus = new ClusterStatus();
+ expectedClusterStatus.setManagementMode(ClusterManagementMode.Type.NORMAL);
+ expectedClusterStatus.setManagementModeStatus(ClusterManagementMode.Status.COMPLETED);
+ verifyClusterStatus(expectedClusterStatus);
+
+ // Verify management mode history: NORMAL + COMPLETED
+ Assert.assertTrue(TestHelper.verify(() -> {
+ ControllerHistory history =
+ _accessor.getProperty(_accessor.keyBuilder().controllerLeaderHistory());
+ List<String> managementHistory = history.getManagementModeHistory();
+ if (managementHistory == null || managementHistory.isEmpty()) {
+ return false;
+ }
+ String lastHistory = managementHistory.get(managementHistory.size() - 1);
+ return lastHistory.contains("MODE=" + ClusterManagementMode.Type.NORMAL)
+ && lastHistory.contains("STATUS=" + ClusterManagementMode.Status.COMPLETED);
+ }, TestHelper.WAIT_DURATION));
+
+ // Verify cluster's normal rebalance ability after unfrozen.
+ Assert.assertTrue(ClusterStateVerifier.verifyByZkCallback(
+ new ClusterStateVerifier.BestPossAndExtViewZkVerifier(ZK_ADDR, _clusterName)));
+ }
+
+ private void verifyLiveInstanceStatus(MockParticipantManager[] participants,
+ LiveInstance.LiveInstanceStatus status) throws Exception {
+ final PropertyKey.Builder keyBuilder = _accessor.keyBuilder();
+ Assert.assertTrue(TestHelper.verify(() -> {
+ for (MockParticipantManager participant : participants) {
+ String instanceName = participant.getInstanceName();
+ LiveInstance liveInstance = _accessor.getProperty(keyBuilder.liveInstance(instanceName));
+ if (status != liveInstance.getStatus()) {
+ return false;
+ }
+ }
+ return true;
+ }, TestHelper.WAIT_DURATION));
+ }
+
+ private void verifyClusterStatus(ClusterStatus expectedMode) throws Exception {
+ final PropertyKey statusPropertyKey = _accessor.keyBuilder().clusterStatus();
+ TestHelper.verify(() -> {
+ ClusterStatus clusterStatus = _accessor.getProperty(statusPropertyKey);
+ return clusterStatus != null
+ && expectedMode.getManagementMode().equals(clusterStatus.getManagementMode())
+ && expectedMode.getManagementModeStatus().equals(clusterStatus.getManagementModeStatus());
+ }, TestHelper.WAIT_DURATION);
+ }
+
+ private static class BlockingTransition extends MockTransition {
+ private static final Logger LOG = LoggerFactory.getLogger(BlockingTransition.class);
+ private final CountDownLatch _countDownLatch;
+
+ private BlockingTransition(CountDownLatch countDownLatch) {
+ _countDownLatch = countDownLatch;
+ }
+
+ @Override
+ public void doTransition(Message message, NotificationContext context)
+ throws InterruptedException {
+ LOG.info("Transition is blocked");
+ _countDownLatch.await();
+ LOG.info("Transition is completed");
+ }
+ }
+}
[helix] 05/13: Add message util to create messages (#1796)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit c95362e7211de5c03660f63f1c97256381d8ad3a
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Mon Jun 14 21:32:25 2021 -0700
Add message util to create messages (#1796)
Message creation methods are private in message generation phase. Management mode stage will also need message generation methods to create ST cancellation and participant status change messages.
Message util will help with the purposes.
This commit moves the common message creation logic to a message util so multiple stages can reuse the code.
---
.../controller/stages/MessageGenerationPhase.java | 97 +++-----------
.../java/org/apache/helix/util/MessageUtil.java | 139 +++++++++++++++++++++
.../TestStateTransitionAppFailureHandling.java | 5 +-
3 files changed, 157 insertions(+), 84 deletions(-)
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 836e5df..38d9908 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,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
@@ -45,13 +44,12 @@ import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.model.Message;
-import org.apache.helix.model.Message.MessageState;
-import org.apache.helix.model.Message.MessageType;
import org.apache.helix.model.Partition;
import org.apache.helix.model.Resource;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.util.HelixUtil;
+import org.apache.helix.util.MessageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,8 +68,6 @@ public class MessageGenerationPhase extends AbstractBaseStage {
.getSystemPropertyAsLong(SystemPropertyKeys.CONTROLLER_MESSAGE_PURGE_DELAY, 60 * 1000);
private final static String PENDING_MESSAGE = "pending message";
private final static String STALE_MESSAGE = "stale message";
- // TODO: Make the message retry count configurable through the Cluster Config or IdealStates.
- public final static int DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT = 3;
private static Logger logger = LoggerFactory.getLogger(MessageGenerationPhase.class);
@@ -215,10 +211,12 @@ public class MessageGenerationPhase extends AbstractBaseStage {
if (desiredState.equals(NO_DESIRED_STATE) || desiredState.equalsIgnoreCase(currentState)) {
if (shouldCreateSTCancellation(pendingMessage, desiredState,
stateModelDef.getInitialState())) {
- message = createStateTransitionCancellationMessage(manager, resource,
- partition.getPartitionName(), instanceName, sessionIdMap.get(instanceName),
- stateModelDef.getId(), pendingMessage.getFromState(), pendingMessage.getToState(),
- null, cancellationMessage, isCancellationEnabled, currentState);
+ message = MessageUtil
+ .createStateTransitionCancellationMessage(manager.getInstanceName(),
+ manager.getSessionId(), resource, partition.getPartitionName(), instanceName,
+ sessionIdMap.get(instanceName), stateModelDef.getId(),
+ pendingMessage.getFromState(), pendingMessage.getToState(), null,
+ cancellationMessage, isCancellationEnabled, currentState);
}
} else {
if (nextState == null) {
@@ -236,9 +234,10 @@ public class MessageGenerationPhase extends AbstractBaseStage {
cancellationMessage, isCancellationEnabled);
} else {
// Create new state transition message
- message = createStateTransitionMessage(manager, resource, partition.getPartitionName(),
- instanceName, currentState, nextState, sessionIdMap.get(instanceName),
- stateModelDef.getId());
+ message = MessageUtil
+ .createStateTransitionMessage(manager.getInstanceName(), manager.getSessionId(),
+ resource, partition.getPartitionName(), instanceName, currentState, nextState,
+ sessionIdMap.get(instanceName), stateModelDef.getId());
if (logger.isDebugEnabled()) {
LogUtil.logDebug(logger, _eventId, String.format(
@@ -331,10 +330,10 @@ public class MessageGenerationPhase extends AbstractBaseStage {
+ instanceName + ", pendingState: " + pendingState + ", currentState: "
+ currentState + ", nextState: " + nextState + ", isRelay: " + pendingMessage.isRelayMessage());
- message = createStateTransitionCancellationMessage(manager, resource,
- partition.getPartitionName(), instanceName, sessionIdMap.get(instanceName),
- stateModelDef.getId(), pendingMessage.getFromState(), pendingState, nextState,
- cancellationMessage, isCancellationEnabled, currentState);
+ message = MessageUtil.createStateTransitionCancellationMessage(manager.getInstanceName(),
+ manager.getSessionId(), resource, partition.getPartitionName(), instanceName,
+ sessionIdMap.get(instanceName), stateModelDef.getId(), pendingMessage.getFromState(),
+ pendingState, nextState, cancellationMessage, isCancellationEnabled, currentState);
}
}
return message;
@@ -417,72 +416,6 @@ public class MessageGenerationPhase extends AbstractBaseStage {
}
}
- private Message createStateTransitionMessage(HelixManager manager, Resource resource,
- String partitionName, String instanceName, String currentState, String nextState,
- String sessionId, String stateModelDefName) {
- String uuid = UUID.randomUUID().toString();
- String managerSessionId = manager.getSessionId();
- Message message = new Message(MessageType.STATE_TRANSITION, uuid);
- message.setSrcName(manager.getInstanceName());
- message.setTgtName(instanceName);
- message.setMsgState(MessageState.NEW);
- message.setPartitionName(partitionName);
- message.setResourceName(resource.getResourceName());
- message.setFromState(currentState);
- message.setToState(nextState);
- message.setTgtSessionId(sessionId);
- message.setSrcSessionId(managerSessionId);
- message.setExpectedSessionId(managerSessionId);
- message.setStateModelDef(stateModelDefName);
- message.setStateModelFactoryName(resource.getStateModelFactoryname());
- message.setBucketSize(resource.getBucketSize());
- // Set the retry count for state transition messages.
- // TODO: make the retry count configurable in ClusterConfig or IdealState
- message.setRetryCount(DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT);
-
- if (resource.getResourceGroupName() != null) {
- message.setResourceGroupName(resource.getResourceGroupName());
- }
- if (resource.getResourceTag() != null) {
- message.setResourceTag(resource.getResourceTag());
- }
-
- return message;
- }
-
- private Message createStateTransitionCancellationMessage(HelixManager manager, Resource resource,
- String partitionName, String instanceName, String sessionId, String stateModelDefName,
- String fromState, String toState, String nextState, Message cancellationMessage,
- boolean isCancellationEnabled, String currentState) {
-
- if (isCancellationEnabled && cancellationMessage == null) {
- logger.info("Event {} : Send cancellation message of the state transition for {}.{} on {}, "
- + "currentState: {}, nextState: {}, toState: {}",
- _eventId, resource.getResourceName(), partitionName, instanceName,
- currentState, nextState == null ? "N/A" : nextState, toState);
-
- String uuid = UUID.randomUUID().toString();
- String managerSessionId = manager.getSessionId();
- Message message = new Message(MessageType.STATE_TRANSITION_CANCELLATION, uuid);
- message.setSrcName(manager.getInstanceName());
- message.setTgtName(instanceName);
- message.setMsgState(MessageState.NEW);
- message.setPartitionName(partitionName);
- message.setResourceName(resource.getResourceName());
- message.setFromState(fromState);
- message.setToState(toState);
- message.setTgtSessionId(sessionId);
- message.setSrcSessionId(managerSessionId);
- message.setExpectedSessionId(managerSessionId);
- message.setStateModelDef(stateModelDefName);
- message.setStateModelFactoryName(resource.getStateModelFactoryname());
- message.setBucketSize(resource.getBucketSize());
- return message;
- }
-
- return null;
- }
-
private int getTimeOut(ClusterConfig clusterConfig, ResourceConfig resourceConfig,
String currentState, String nextState, IdealState idealState, Partition partition) {
StateTransitionTimeoutConfig stateTransitionTimeoutConfig =
diff --git a/helix-core/src/main/java/org/apache/helix/util/MessageUtil.java b/helix-core/src/main/java/org/apache/helix/util/MessageUtil.java
new file mode 100644
index 0000000..94de833
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/util/MessageUtil.java
@@ -0,0 +1,139 @@
+package org.apache.helix.util;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.UUID;
+
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.Message;
+import org.apache.helix.model.Resource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Message utils to operate on message such creating messages.
+ */
+public class MessageUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(MessageUtil.class);
+
+ // TODO: Make the message retry count configurable through the Cluster Config or IdealStates.
+ public final static int DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT = 3;
+
+ public static Message createStateTransitionCancellationMessage(String srcInstanceName,
+ String srcSessionId, Resource resource, String partitionName, String instanceName,
+ String sessionId, String stateModelDefName, String fromState, String toState,
+ String nextState, Message cancellationMessage, boolean isCancellationEnabled,
+ String currentState) {
+ if (isCancellationEnabled && cancellationMessage == null) {
+ LOG.info("Create cancellation message of the state transition for {}.{} on {}, "
+ + "currentState: {}, nextState: {}, toState: {}", resource.getResourceName(),
+ partitionName, instanceName, currentState, nextState == null ? "N/A" : nextState,
+ toState);
+
+ Message message =
+ createStateTransitionMessage(Message.MessageType.STATE_TRANSITION_CANCELLATION,
+ srcInstanceName, srcSessionId, resource, partitionName, instanceName, currentState,
+ nextState, sessionId, stateModelDefName);
+
+ message.setFromState(fromState);
+ message.setToState(toState);
+ return message;
+ }
+
+ return null;
+ }
+
+ public static Message createStateTransitionMessage(String srcInstanceName, String srcSessionId,
+ Resource resource, String partitionName, String instanceName, String currentState,
+ String nextState, String tgtSessionId, String stateModelDefName) {
+ Message message =
+ createStateTransitionMessage(Message.MessageType.STATE_TRANSITION, srcInstanceName,
+ srcSessionId, resource, partitionName, instanceName, currentState, nextState, tgtSessionId,
+ stateModelDefName);
+
+ // Set the retry count for state transition messages.
+ // TODO: make the retry count configurable in ClusterConfig or IdealState
+ message.setRetryCount(DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT);
+
+ if (resource.getResourceGroupName() != null) {
+ message.setResourceGroupName(resource.getResourceGroupName());
+ }
+ if (resource.getResourceTag() != null) {
+ message.setResourceTag(resource.getResourceTag());
+ }
+
+ return message;
+ }
+
+ /**
+ * Creates a message to change participant status
+ * {@link org.apache.helix.model.LiveInstance.LiveInstanceStatus}
+ *
+ * @param currentState current status of the live instance
+ * @param nextState next status that will be changed to
+ * @param srcInstanceName source instance name
+ * @param srcSessionId session id for the source instance
+ * @param tgtInstanceName target instance name
+ * @param tgtSessionId target instance session id
+ * @return participant status change message
+ */
+ public static Message createStatusChangeMessage(LiveInstance.LiveInstanceStatus currentState,
+ LiveInstance.LiveInstanceStatus nextState, String srcInstanceName, String srcSessionId,
+ String tgtInstanceName, String tgtSessionId) {
+ return createBasicMessage(Message.MessageType.PARTICIPANT_STATUS_CHANGE, srcInstanceName,
+ srcSessionId, tgtInstanceName, tgtSessionId, currentState.name(), nextState.name());
+ }
+
+ /* Creates a message that that has the least required fields. */
+ private static Message createBasicMessage(Message.MessageType messageType, String srcInstanceName,
+ String srcSessionId, String tgtInstanceName, String tgtSessionId, String currentState,
+ String nextState) {
+ String uuid = UUID.randomUUID().toString();
+
+ Message message = new Message(messageType, uuid);
+ message.setSrcName(srcInstanceName);
+ message.setTgtName(tgtInstanceName);
+ message.setMsgState(Message.MessageState.NEW);
+ message.setFromState(currentState);
+ message.setToState(nextState);
+ message.setTgtSessionId(tgtSessionId);
+ message.setSrcSessionId(srcSessionId);
+ message.setExpectedSessionId(srcSessionId);
+
+ return message;
+ }
+
+ /* Creates state transition or state transition cancellation message */
+ private static Message createStateTransitionMessage(Message.MessageType messageType,
+ String srcInstanceName, String srcSessionId, Resource resource, String partitionName,
+ String instanceName, String currentState, String nextState, String tgtSessionId,
+ String stateModelDefName) {
+ Message message =
+ createBasicMessage(messageType, srcInstanceName, srcSessionId, instanceName, tgtSessionId,
+ currentState, nextState);
+ message.setPartitionName(partitionName);
+ message.setStateModelDef(stateModelDefName);
+ message.setResourceName(resource.getResourceName());
+ message.setStateModelFactoryName(resource.getStateModelFactoryname());
+ message.setBucketSize(resource.getBucketSize());
+
+ return message;
+ }
+}
diff --git a/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestStateTransitionAppFailureHandling.java b/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestStateTransitionAppFailureHandling.java
index 12ede56..e97424c 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestStateTransitionAppFailureHandling.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestStateTransitionAppFailureHandling.java
@@ -37,6 +37,7 @@ import org.apache.helix.mock.participant.MockTransition;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.Message;
import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.apache.helix.util.MessageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
@@ -122,7 +123,7 @@ public class TestStateTransitionAppFailureHandling extends ZkStandAloneCMTestBas
// Check if the factory has tried enough times before fail the message.
Assert.assertEquals(retryCountUntilSucceed - retryFactoryMap.get(instanceName)
.getRemainingRetryCountUntilSucceed(), instanceMessages.size()
- * MessageGenerationPhase.DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT);
+ * MessageUtil.DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT);
}
// Verify that the partition is not initialized.
@@ -146,7 +147,7 @@ public class TestStateTransitionAppFailureHandling extends ZkStandAloneCMTestBas
// Make the mock StateModelFactory return handler before last retry. So it will successfully
// finish handler initialization.
int retryCountUntilSucceed =
- MessageGenerationPhase.DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT - 1;
+ MessageUtil.DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT - 1;
Map<String, RetryStateModelFactory> retryFactoryMap = resetParticipants(retryCountUntilSucceed);
_gSetupTool.addResourceToCluster(CLUSTER_NAME, TEST_DB, _PARTITIONS, STATE_MODEL);
[helix] 09/13: Implement Participant Freeze Process (#1812)
Posted by hz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hzlu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 0aca85a6b867594900b801a173c2159f437d2f7a
Author: Huizhi Lu <51...@users.noreply.github.com>
AuthorDate: Wed Jul 14 12:22:19 2021 -0700
Implement Participant Freeze Process (#1812)
After controller sends a freeze message to participant, participant needs to process the freeze message and execute the freeze request. It will mark the live instance as "frozen" internally in memory and in live instance znode.
For session change during freeze, handle new session won't do anything, except creating the live instance znode.
To unfreeze, if there is session change during the freeze period, state model state will be synced and current states will be carried over. Then live instance status field is removed.
---
.../helix/manager/zk/CurStateCarryOverUpdater.java | 19 +-
.../helix/manager/zk/ParticipantManager.java | 79 ++++--
.../apache/helix/manager/zk/ZKHelixManager.java | 47 +++-
.../messaging/handling/HelixTaskExecutor.java | 115 ++++++--
.../messaging/handling/MessageHandlerFactory.java | 10 +
.../helix/participant/HelixStateMachineEngine.java | 27 +-
.../helix/participant/statemachine/StateModel.java | 7 +
.../paticipant/TestParticipantFreeze.java | 291 +++++++++++++++++++++
8 files changed, 541 insertions(+), 54 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java b/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java
index a00c337..f62aa65 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/CurStateCarryOverUpdater.java
@@ -19,10 +19,10 @@ package org.apache.helix.manager.zk;
* under the License.
*/
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import java.util.Map;
+
import org.apache.helix.model.CurrentState;
-import org.apache.helix.task.TaskConstants;
-import org.apache.helix.task.TaskPartitionState;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
@@ -35,16 +35,17 @@ import org.apache.helix.zookeeper.zkclient.DataUpdater;
*/
class CurStateCarryOverUpdater implements DataUpdater<ZNRecord> {
final String _curSessionId;
- final String _initState;
+ final Map<String, String> _expectedStateMap;
final CurrentState _lastCurState;
- public CurStateCarryOverUpdater(String curSessionId, String initState, CurrentState lastCurState) {
- if (curSessionId == null || initState == null || lastCurState == null) {
+ public CurStateCarryOverUpdater(String curSessionId, Map<String, String> expectedStateMap,
+ CurrentState lastCurState) {
+ if (curSessionId == null || lastCurState == null) {
throw new IllegalArgumentException(
- "missing curSessionId|initState|lastCurState for carry-over");
+ "missing curSessionId|lastCurState for carry-over");
}
_curSessionId = curSessionId;
- _initState = initState;
+ _expectedStateMap = expectedStateMap;
_lastCurState = lastCurState;
}
@@ -63,7 +64,7 @@ class CurStateCarryOverUpdater implements DataUpdater<ZNRecord> {
for (String partitionName : _lastCurState.getPartitionStateMap().keySet()) {
// carry-over only when current-state does not exist for regular Helix resource partitions
if (curState.getState(partitionName) == null) {
- curState.setState(partitionName, _initState);
+ curState.setState(partitionName, _expectedStateMap.get(partitionName));
}
}
return curState.getRecord();
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
index 6e071f6..b7a0014 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
@@ -23,6 +23,7 @@ import java.lang.management.ManagementFactory;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@@ -32,6 +33,7 @@ import org.apache.helix.BaseDataAccessor;
import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixCloudProperty;
+import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.HelixManager;
import org.apache.helix.HelixManagerProperty;
@@ -52,12 +54,15 @@ import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
import org.apache.helix.participant.StateMachineEngine;
import org.apache.helix.participant.statemachine.ScheduledTaskStateModelFactory;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelFactory;
import org.apache.helix.task.TaskConstants;
import org.apache.helix.task.TaskUtil;
import org.apache.helix.util.HelixUtil;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.ZNRecordBucketizer;
+import org.apache.helix.zookeeper.exception.ZkClientException;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.exception.ZkNodeExistsException;
import org.apache.helix.zookeeper.zkclient.exception.ZkSessionMismatchedException;
@@ -153,7 +158,10 @@ public class ParticipantManager {
// Live instance creation also checks if the expected session is valid or not. Live instance
// should not be created by an expired zk session.
createLiveInstance();
- carryOverPreviousCurrentState();
+ if (shouldCarryOver()) {
+ carryOverPreviousCurrentState(_dataAccessor, _instanceName, _sessionId,
+ _manager.getStateMachineEngine(), true);
+ }
removePreviousTaskCurrentStates();
/**
@@ -162,6 +170,17 @@ public class ParticipantManager {
setupMsgHandler();
}
+ private boolean shouldCarryOver() {
+ if (_liveInstanceInfoProvider == null
+ || _liveInstanceInfoProvider.getAdditionalLiveInstanceInfo() == null) {
+ return true;
+ }
+ String status = _liveInstanceInfoProvider.getAdditionalLiveInstanceInfo()
+ .getSimpleField(LiveInstance.LiveInstanceProperty.STATUS.name());
+ // If frozen, no carry-over
+ return !LiveInstance.LiveInstanceStatus.PAUSED.name().equals(status);
+ }
+
private void joinCluster() {
// Read cluster config and see if an instance can auto join or auto register to the cluster
boolean autoJoin = false;
@@ -339,21 +358,24 @@ public class ParticipantManager {
* carry over current-states from last sessions
* set to initial state for current session only when state doesn't exist in current session
*/
- private void carryOverPreviousCurrentState() {
- List<String> sessions = _dataAccessor.getChildNames(_keyBuilder.sessions(_instanceName));
+ public static synchronized void carryOverPreviousCurrentState(HelixDataAccessor dataAccessor,
+ String instanceName, String sessionId, StateMachineEngine stateMachineEngine,
+ boolean setToInitState) {
+ PropertyKey.Builder keyBuilder = dataAccessor.keyBuilder();
+ List<String> sessions = dataAccessor.getChildNames(keyBuilder.sessions(instanceName));
for (String session : sessions) {
- if (session.equals(_sessionId)) {
+ if (session.equals(sessionId)) {
continue;
}
// Ignore if any current states in the previous folder cannot be read.
List<CurrentState> lastCurStates =
- _dataAccessor.getChildValues(_keyBuilder.currentStates(_instanceName, session), false);
+ dataAccessor.getChildValues(keyBuilder.currentStates(instanceName, session), false);
for (CurrentState lastCurState : lastCurStates) {
LOG.info("Carrying over old session: " + session + ", resource: " + lastCurState.getId()
- + " to current session: " + _sessionId);
+ + " to current session: " + sessionId + ", setToInitState: " + setToInitState);
String stateModelDefRef = lastCurState.getStateModelDefRef();
if (stateModelDefRef == null) {
LOG.error(
@@ -368,21 +390,37 @@ public class ParticipantManager {
continue;
}
- StateModelDefinition stateModel =
- _dataAccessor.getProperty(_keyBuilder.stateModelDef(stateModelDefRef));
+ StateModelDefinition stateModelDef =
+ dataAccessor.getProperty(keyBuilder.stateModelDef(stateModelDefRef));
+ String initState = stateModelDef.getInitialState();
+ Map<String, String> partitionExpectedStateMap = new HashMap<>();
+ if (setToInitState) {
+ lastCurState.getPartitionStateMap().keySet()
+ .forEach(partition -> partitionExpectedStateMap.put(partition, initState));
+ } else {
+ String factoryName = lastCurState.getStateModelFactoryName();
+ StateModelFactory<? extends StateModel> stateModelFactory =
+ stateMachineEngine.getStateModelFactory(stateModelDefRef, factoryName);
+ lastCurState.getPartitionStateMap().keySet().forEach(partition -> {
+ StateModel stateModel =
+ stateModelFactory.getStateModel(lastCurState.getResourceName(), partition);
+ if (stateModel != null) {
+ partitionExpectedStateMap.put(partition, stateModel.getCurrentState());
+ }
+ });
+ }
- BaseDataAccessor<ZNRecord> baseAccessor = _dataAccessor.getBaseDataAccessor();
+ BaseDataAccessor<ZNRecord> baseAccessor = dataAccessor.getBaseDataAccessor();
String curStatePath =
- _keyBuilder.currentState(_instanceName, _sessionId, lastCurState.getResourceName())
+ keyBuilder.currentState(instanceName, sessionId, lastCurState.getResourceName())
.getPath();
- String initState = stateModel.getInitialState();
if (lastCurState.getBucketSize() > 0) {
// update parent node
ZNRecord metaRecord = new ZNRecord(lastCurState.getId());
metaRecord.setSimpleFields(lastCurState.getRecord().getSimpleFields());
DataUpdater<ZNRecord> metaRecordUpdater =
- new CurStateCarryOverUpdater(_sessionId, initState, new CurrentState(metaRecord));
+ new CurStateCarryOverUpdater(sessionId, partitionExpectedStateMap, new CurrentState(metaRecord));
boolean success =
baseAccessor.update(curStatePath, metaRecordUpdater, AccessOption.PERSISTENT);
if (success) {
@@ -394,7 +432,7 @@ public class ParticipantManager {
List<DataUpdater<ZNRecord>> updaters = new ArrayList<DataUpdater<ZNRecord>>();
for (String bucketName : map.keySet()) {
paths.add(curStatePath + "/" + bucketName);
- updaters.add(new CurStateCarryOverUpdater(_sessionId, initState, new CurrentState(map
+ updaters.add(new CurStateCarryOverUpdater(sessionId, partitionExpectedStateMap, new CurrentState(map
.get(bucketName))));
}
@@ -402,8 +440,8 @@ public class ParticipantManager {
}
} else {
- _dataAccessor.getBaseDataAccessor().update(curStatePath,
- new CurStateCarryOverUpdater(_sessionId, initState, lastCurState),
+ dataAccessor.getBaseDataAccessor().update(curStatePath,
+ new CurStateCarryOverUpdater(sessionId, partitionExpectedStateMap, lastCurState),
AccessOption.PERSISTENT);
}
}
@@ -413,13 +451,16 @@ public class ParticipantManager {
* remove previous current state parent nodes
*/
for (String session : sessions) {
- if (session.equals(_sessionId)) {
+ if (session.equals(sessionId)) {
continue;
}
- String path = _keyBuilder.currentStates(_instanceName, session).getPath();
- LOG.info("Removing current states from previous sessions. path: " + path);
- _zkclient.deleteRecursively(path);
+ PropertyKey currentStatesProperty = keyBuilder.currentStates(instanceName, session);
+ String path = currentStatesProperty.getPath();
+ LOG.info("Removing current states from previous sessions. path: {}", path);
+ if (!dataAccessor.removeProperty(currentStatesProperty)) {
+ throw new ZkClientException("Failed to delete " + path);
+ }
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
index ef8e733..4b88f74 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
@@ -1317,6 +1317,15 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
LOG.info("Handle new session, instance: {}, type: {}, session id: {}.", _instanceName,
_instanceType, sessionId);
+ // Will only create live instance
+ LiveInstance.LiveInstanceStatus liveInstanceStatus =
+ _messagingService.getExecutor().getLiveInstanceStatus();
+ if (LiveInstance.LiveInstanceStatus.PAUSED
+ .equals(liveInstanceStatus)) {
+ handleNewSessionInManagementMode(sessionId, liveInstanceStatus);
+ return;
+ }
+
/**
* stop all timer tasks, reset all handlers, make sure cleanup completed for previous session
* disconnect if fail to cleanup
@@ -1343,13 +1352,13 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
switch (_instanceType) {
case PARTICIPANT:
- handleNewSessionAsParticipant(sessionId);
+ handleNewSessionAsParticipant(sessionId, _liveInstanceInfoProvider);
break;
case CONTROLLER:
handleNewSessionAsController();
break;
case CONTROLLER_PARTICIPANT:
- handleNewSessionAsParticipant(sessionId);
+ handleNewSessionAsParticipant(sessionId, _liveInstanceInfoProvider);
handleNewSessionAsController();
break;
case ADMINISTRATOR:
@@ -1376,12 +1385,25 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
}
}
- void handleNewSessionAsParticipant(final String sessionId) throws Exception {
+ private void handleNewSessionInManagementMode(String sessionId,
+ LiveInstance.LiveInstanceStatus liveInstanceStatus) throws Exception {
+ LOG.info("Skip reset because instance is in {} status", LiveInstance.LiveInstanceStatus.PAUSED);
+ if (!InstanceType.PARTICIPANT.equals(_instanceType)
+ && !InstanceType.CONTROLLER_PARTICIPANT.equals(_instanceType)) {
+ return;
+ }
+ // Add STATUS to info provider so the new live instance will have STATUS field
+ handleNewSessionAsParticipant(sessionId,
+ new LiveInstanceStatusInfoProvider(liveInstanceStatus));
+ }
+
+ void handleNewSessionAsParticipant(final String sessionId, LiveInstanceInfoProvider provider)
+ throws Exception {
if (_participantManager != null) {
_participantManager.reset();
}
_participantManager =
- new ParticipantManager(this, _zkclient, _sessionTimeout, _liveInstanceInfoProvider,
+ new ParticipantManager(this, _zkclient, _sessionTimeout, provider,
_preConnectCallbacks, sessionId, _helixManagerProperty);
_participantManager.handleNewSession();
}
@@ -1535,4 +1557,21 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
}
return zkConnectionInfo;
}
+
+ /*
+ * Provides live instance status as additional live instance info in the info provider.
+ */
+ private static class LiveInstanceStatusInfoProvider implements LiveInstanceInfoProvider {
+ private final ZNRecord _record;
+
+ public LiveInstanceStatusInfoProvider(LiveInstance.LiveInstanceStatus status) {
+ _record = new ZNRecord("STATUS_PROVIDER");
+ _record.setEnumField(LiveInstance.LiveInstanceProperty.STATUS.name(), status);
+ }
+
+ @Override
+ public ZNRecord getAdditionalLiveInstanceInfo() {
+ return _record;
+ }
+ }
}
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
index 0e8aace..ff970c2 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
@@ -56,10 +56,12 @@ import org.apache.helix.SystemPropertyKeys;
import org.apache.helix.api.listeners.MessageListener;
import org.apache.helix.api.listeners.PreFetch;
import org.apache.helix.controller.GenericHelixController;
+import org.apache.helix.manager.zk.ParticipantManager;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.HelixConfigScope;
import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.LiveInstance.LiveInstanceStatus;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.MessageState;
import org.apache.helix.model.Message.MessageType;
@@ -127,6 +129,8 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
private MessageQueueMonitor _messageQueueMonitor;
private GenericHelixController _controller;
private Long _lastSessionSyncTime;
+ private String _freezeSessionId;
+ private LiveInstanceStatus _liveInstanceStatus;
private static final int SESSION_SYNC_INTERVAL = 2000; // 2 seconds
private static final String SESSION_SYNC = "SESSION-SYNC";
/**
@@ -630,6 +634,24 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
+ ", pool: " + pool);
}
+ private void syncFactoryState() {
+ LOG.info("Start to sync factory state");
+ // Lock on the registry to avoid race condition when concurrently calling sync() and reset()
+ synchronized (_hdlrFtyRegistry) {
+ for (Map.Entry<String, MsgHandlerFactoryRegistryItem> entry : _hdlrFtyRegistry.entrySet()) {
+ MsgHandlerFactoryRegistryItem item = entry.getValue();
+ if (item.factory() != null) {
+ try {
+ item.factory().sync();
+ } catch (Exception ex) {
+ LOG.error("Failed to syncState the factory {} of message type {}.", item.factory(),
+ entry.getKey(), ex);
+ }
+ }
+ }
+ }
+ }
+
void reset() {
LOG.info("Reset HelixTaskExecutor");
@@ -637,22 +659,24 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
_messageQueueMonitor.reset();
}
- for (String msgType : _hdlrFtyRegistry.keySet()) {
- // don't un-register factories, just shutdown all executors
- ExecutorService pool = _executorMap.remove(msgType);
- _monitor.removeExecutorMonitor(msgType);
- if (pool != null) {
- LOG.info("Reset exectuor for msgType: " + msgType + ", pool: " + pool);
- shutdownAndAwaitTermination(pool);
- }
+ synchronized (_hdlrFtyRegistry) {
+ for (String msgType : _hdlrFtyRegistry.keySet()) {
+ // don't un-register factories, just shutdown all executors
+ ExecutorService pool = _executorMap.remove(msgType);
+ _monitor.removeExecutorMonitor(msgType);
+ if (pool != null) {
+ LOG.info("Reset exectuor for msgType: " + msgType + ", pool: " + pool);
+ shutdownAndAwaitTermination(pool);
+ }
- MsgHandlerFactoryRegistryItem item = _hdlrFtyRegistry.get(msgType);
- if (item.factory() != null) {
- try {
- item.factory().reset();
- } catch (Exception ex) {
- LOG.error("Failed to reset the factory {} of message type {}.", item.factory().toString(),
- msgType, ex);
+ MsgHandlerFactoryRegistryItem item = _hdlrFtyRegistry.get(msgType);
+ if (item.factory() != null) {
+ try {
+ item.factory().reset();
+ } catch (Exception ex) {
+ LOG.error("Failed to reset the factory {} of message type {}.", item.factory().toString(),
+ msgType, ex);
+ }
}
}
}
@@ -1068,6 +1092,13 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
}
}
+ if (MessageType.PARTICIPANT_STATUS_CHANGE.name().equals(message.getMsgType())) {
+ LiveInstanceStatus toStatus = LiveInstanceStatus.valueOf(message.getToState());
+ changeParticipantStatus(instanceName, toStatus, manager);
+ reportAndRemoveMessage(message, accessor, instanceName, ProcessedMessageState.COMPLETED);
+ return true;
+ }
+
_monitor.reportReceivedMessage(message);
} catch (Exception e) {
LOG.error("Failed to process the message {}. Deleting the message from ZK. Exception: {}",
@@ -1305,6 +1336,56 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
return String.format("%s_%s", resourceName, partitionName);
}
+ private void changeParticipantStatus(String instanceName,
+ LiveInstance.LiveInstanceStatus toStatus, HelixManager manager) {
+ if (toStatus == null) {
+ LOG.warn("To status is null! Skip participant status change.");
+ return;
+ }
+
+ LOG.info("Changing participant {} status to {} from {}", instanceName, toStatus,
+ _liveInstanceStatus);
+ HelixDataAccessor accessor = manager.getHelixDataAccessor();
+ String sessionId = manager.getSessionId();
+ String path = accessor.keyBuilder().liveInstance(instanceName).getPath();
+ boolean success = false;
+
+ switch (toStatus) {
+ case PAUSED:
+ _freezeSessionId = sessionId;
+ _liveInstanceStatus = toStatus;
+ // Entering freeze mode, update live instance status.
+ // If the update fails, another new freeze message will be sent by controller.
+ success = accessor.getBaseDataAccessor().update(path, record -> {
+ record.setEnumField(LiveInstance.LiveInstanceProperty.STATUS.name(), toStatus);
+ return record;
+ }, AccessOption.EPHEMERAL);
+ break;
+ case NORMAL:
+ // Exiting freeze mode
+ // session changed, should call state model sync
+ if (_freezeSessionId != null && !_freezeSessionId.equals(sessionId)) {
+ syncFactoryState();
+ ParticipantManager.carryOverPreviousCurrentState(accessor, instanceName, sessionId,
+ manager.getStateMachineEngine(), false);
+ }
+ _freezeSessionId = null;
+ _liveInstanceStatus = null;
+ success = accessor.getBaseDataAccessor().update(path, record -> {
+ // Remove the status field for backwards compatibility
+ record.getSimpleFields().remove(LiveInstance.LiveInstanceProperty.STATUS.name());
+ return record;
+ }, AccessOption.EPHEMERAL);
+ break;
+ default:
+ LOG.warn("To status {} is not supported", toStatus);
+ break;
+ }
+
+ LOG.info("Changed participant {} status to {}. FreezeSessionId={}, update success={}",
+ instanceName, _liveInstanceStatus, _freezeSessionId, success);
+ }
+
private String getStateTransitionType(String prefix, String fromState, String toState) {
if (prefix == null || fromState == null || toState == null) {
return null;
@@ -1316,6 +1397,10 @@ public class HelixTaskExecutor implements MessageListener, TaskExecutor {
return MessageType.STATE_TRANSITION.name() + "." + resourceName;
}
+ public LiveInstanceStatus getLiveInstanceStatus() {
+ return _liveInstanceStatus;
+ }
+
private void removeMessageFromZK(HelixDataAccessor accessor, Message message,
String instanceName) {
if (HelixUtil.removeMessageFromZK(accessor, message, instanceName)) {
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/MessageHandlerFactory.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/MessageHandlerFactory.java
index acbeadc..b402c29 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/MessageHandlerFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/MessageHandlerFactory.java
@@ -21,6 +21,8 @@ package org.apache.helix.messaging.handling;
import org.apache.helix.NotificationContext;
import org.apache.helix.model.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
@Deprecated
public interface MessageHandlerFactory {
@@ -37,4 +39,12 @@ public interface MessageHandlerFactory {
String getMessageType();
void reset();
+
+ default void sync() {
+ LogHolder.LOG.warn("Invoked default sync() without any operation");
+ }
+}
+
+final class LogHolder {
+ static final Logger LOG = LoggerFactory.getLogger(MessageHandlerFactory.class);
}
diff --git a/helix-core/src/main/java/org/apache/helix/participant/HelixStateMachineEngine.java b/helix-core/src/main/java/org/apache/helix/participant/HelixStateMachineEngine.java
index 4757c44..e421638 100644
--- a/helix-core/src/main/java/org/apache/helix/participant/HelixStateMachineEngine.java
+++ b/helix-core/src/main/java/org/apache/helix/participant/HelixStateMachineEngine.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
import com.google.common.collect.ImmutableList;
import org.apache.helix.HelixConstants;
@@ -147,17 +148,31 @@ public class HelixStateMachineEngine implements StateMachineEngine {
@Override
public void reset() {
logger.info("Resetting HelixStateMachineEngine");
+ loopStateModelFactories(stateModel -> {
+ stateModel.reset();
+ String initialState = _stateModelParser.getInitialState(stateModel.getClass());
+ stateModel.updateState(initialState);
+ });
+ logger.info("Successfully reset HelixStateMachineEngine");
+ }
+
+ @Override
+ public void sync() {
+ logger.info("Syncing HelixStateMachineEngine");
+ loopStateModelFactories(StateModel::syncState);
+ logger.info("Successfully synced HelixStateMachineEngine");
+ }
+
+ private void loopStateModelFactories(Consumer<StateModel> consumer) {
for (Map<String, StateModelFactory<? extends StateModel>> ftyMap : _stateModelFactoryMap
.values()) {
for (StateModelFactory<? extends StateModel> stateModelFactory : ftyMap.values()) {
for (String resourceName : stateModelFactory.getResourceSet()) {
for (String partitionKey : stateModelFactory.getPartitionSet(resourceName)) {
- logger.info("Resetting {}::{}", resourceName, partitionKey);
+ logger.info("Operating on {}::{}", resourceName, partitionKey);
StateModel stateModel = stateModelFactory.getStateModel(resourceName, partitionKey);
if (stateModel != null) {
- stateModel.reset();
- String initialState = _stateModelParser.getInitialState(stateModel.getClass());
- stateModel.updateState(initialState);
+ consumer.accept(stateModel);
// TODO probably should update the state on ZK. Shi confirm what needs
// to be done here.
} else {
@@ -170,15 +185,13 @@ public class HelixStateMachineEngine implements StateMachineEngine {
// We may need to add more processing here to make sure things are being set to
// initialState. Otherwise, there might be inconsistencies that might cause partitions
// to be stuck in some state (because reset() would be a NOP here)
- logger.warn(
- "Failed to reset due to StateModel being null! Resource: {}, Partition: {}",
+ logger.warn("Failed operation due to StateModel being null! Resource: {}, Partition: {}",
resourceName, partitionKey);
}
}
}
}
}
- logger.info("Successfully reset HelixStateMachineEngine");
}
@Override
diff --git a/helix-core/src/main/java/org/apache/helix/participant/statemachine/StateModel.java b/helix-core/src/main/java/org/apache/helix/participant/statemachine/StateModel.java
index 2c88a68..143c14a 100644
--- a/helix-core/src/main/java/org/apache/helix/participant/statemachine/StateModel.java
+++ b/helix-core/src/main/java/org/apache/helix/participant/statemachine/StateModel.java
@@ -63,6 +63,13 @@ public abstract class StateModel {
}
/**
+ * Called when cluster is recovering from freeze mode if session changed during freeze mode.
+ */
+ public void syncState() {
+ logger.warn("Default syncState method invoked without any operation.");
+ }
+
+ /**
* Called when the state model is reset
*/
public void reset() {
diff --git a/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java b/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java
new file mode 100644
index 0000000..fedee37
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestParticipantFreeze.java
@@ -0,0 +1,291 @@
+package org.apache.helix.integration.paticipant;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.TestHelper;
+import org.apache.helix.ZkTestHelper;
+import org.apache.helix.common.ZkTestBase;
+import org.apache.helix.integration.manager.ClusterControllerManager;
+import org.apache.helix.integration.manager.ClusterManager;
+import org.apache.helix.integration.manager.MockParticipantManager;
+import org.apache.helix.messaging.DefaultMessagingService;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.Message;
+import org.apache.helix.model.StateModelDefinition;
+import org.apache.helix.tools.ClusterStateVerifier;
+import org.apache.helix.util.MessageUtil;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+public class TestParticipantFreeze extends ZkTestBase {
+ private HelixManager _manager;
+ private HelixDataAccessor _accessor;
+ private PropertyKey.Builder _keyBuilder;
+ private String _clusterName;
+ private int _numNodes;
+ private String _resourceName;
+ private String _instanceName;
+ private MockParticipantManager[] _participants;
+ // current states in participant[0]
+ private List<CurrentState> _originCurStates;
+ private String _originSession;
+
+ @BeforeClass
+ public void beforeClass() throws Exception {
+ _clusterName = "CLUSTER_" + TestHelper.getTestClassName();
+ _numNodes = 3;
+ _resourceName = "TestDB";
+ _participants = new MockParticipantManager[_numNodes];
+ TestHelper.setupCluster(_clusterName, ZK_ADDR, 12918, // participant port
+ "localhost", // participant name prefix
+ _resourceName, // resource name prefix
+ 1, // resources
+ 1, // partitions per resource
+ _numNodes, // number of nodes
+ 3, // replicas
+ "MasterSlave", true);
+
+ _manager = HelixManagerFactory
+ .getZKHelixManager(_clusterName, "Admin", InstanceType.ADMINISTRATOR, ZK_ADDR);
+ _manager.connect();
+ _accessor = _manager.getHelixDataAccessor();
+ _keyBuilder = _accessor.keyBuilder();
+
+ // start controller
+ ClusterControllerManager controller =
+ new ClusterControllerManager(ZK_ADDR, _clusterName, "controller_0");
+ controller.syncStart();
+
+ // start participants
+ for (int i = 0; i < _numNodes; i++) {
+ String instanceName = "localhost_" + (12918 + i);
+ _participants[i] = new MockParticipantManager(ZK_ADDR, _clusterName, instanceName);
+ _participants[i].syncStart();
+ }
+ _instanceName = _participants[0].getInstanceName();
+
+ Assert.assertTrue(ClusterStateVerifier.verifyByZkCallback(
+ new ClusterStateVerifier.BestPossAndExtViewZkVerifier(ZK_ADDR, _clusterName)));
+
+ // We just need controller to rebalance the cluster once to get current states.
+ controller.syncStop();
+
+ _originSession = _participants[0].getSessionId();
+ _originCurStates =
+ _accessor.getChildValues(_keyBuilder.currentStates(_instanceName, _originSession), false);
+ }
+
+ @AfterClass
+ public void afterClass() {
+ _manager.disconnect();
+ Arrays.stream(_participants).forEach(ClusterManager::syncStop);
+ deleteCluster(_clusterName);
+ }
+
+ /*
+ * Live instance is not frozen and does not have a frozen status field
+ */
+ @Test
+ public void testNormalLiveInstanceStatus() {
+ LiveInstance liveInstance = _accessor.getProperty(_keyBuilder.liveInstance(_instanceName));
+ Assert.assertNull(liveInstance.getStatus());
+ }
+
+ @Test(dependsOnMethods = "testNormalLiveInstanceStatus")
+ public void testFreezeParticipant() throws Exception {
+ freezeParticipant(_participants[0]);
+ }
+
+ // Simulates instance is restarted and the in-memory status is gone.
+ // When instance comes back alive, it'll reset state model, carry over
+ // and set current state to init state.
+ @Test(dependsOnMethods = "testFreezeParticipant")
+ public void testRestartParticipantWhenFrozen() throws Exception {
+ String instanceName = _participants[1].getInstanceName();
+ List<CurrentState> originCurStates = _accessor
+ .getChildValues(_keyBuilder.currentStates(instanceName, _participants[1].getSessionId()),
+ false);
+ String oldSession = _participants[1].getSessionId();
+ freezeParticipant(_participants[1]);
+
+ // Restart participants[1]
+ _participants[1].syncStop();
+ _participants[1] = new MockParticipantManager(ZK_ADDR, _participants[1].getClusterName(),
+ instanceName);
+ _participants[1].syncStart();
+
+ Assert.assertTrue(TestHelper.verify(() ->
+ _gZkClient.exists(_keyBuilder.liveInstance(instanceName).getPath()),
+ TestHelper.WAIT_DURATION));
+ LiveInstance liveInstance = _accessor.getProperty(_keyBuilder.liveInstance(instanceName));
+
+ // New live instance ephemeral node
+ Assert.assertEquals(liveInstance.getEphemeralOwner(), _participants[1].getSessionId());
+ // Status is not frozen because controller is not running, no freeze message sent.
+ verifyLiveInstanceStatus(_participants[1], null);
+
+ // Old session current state is deleted because of current state carry-over
+ Assert.assertTrue(TestHelper.verify(
+ () -> !_gZkClient.exists(_keyBuilder.currentStates(instanceName, oldSession).getPath()),
+ TestHelper.WAIT_DURATION));
+
+ // Current states are set to init states (OFFLINE)
+ List<CurrentState> curStates = _accessor
+ .getChildValues(_keyBuilder.currentStates(instanceName, _participants[1].getSessionId()),
+ false);
+ Assert.assertEquals(curStates.size(), 1);
+ Assert.assertTrue(TestHelper.verify(() -> {
+ for (CurrentState cs : originCurStates) {
+ String stateModelDefRef = cs.getStateModelDefRef();
+ for (String partition : cs.getPartitionStateMap().keySet()) {
+ StateModelDefinition stateModelDef =
+ _accessor.getProperty(_keyBuilder.stateModelDef(stateModelDefRef));
+ String initState = stateModelDef.getInitialState();
+ if (!initState.equals(curStates.get(0).getPartitionStateMap().get(partition))) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }, TestHelper.WAIT_DURATION));
+ }
+
+ // Simulates session expires but in-memory status is still kept.
+ // No state model reset or current state carry-over
+ @Test(dependsOnMethods = "testRestartParticipantWhenFrozen")
+ public void testHandleNewSessionWhenFrozen() throws Exception {
+ // there are current states for the resource
+ Assert.assertFalse(_originCurStates.isEmpty());
+
+ ZkTestHelper.expireSession(_participants[0].getZkClient());
+ String currentSession = _participants[0].getSessionId();
+ Assert.assertFalse(_originSession.equals(currentSession));
+
+ Assert.assertTrue(TestHelper.verify(() ->
+ _gZkClient.exists(_keyBuilder.liveInstance(_instanceName).getPath()),
+ TestHelper.WAIT_DURATION));
+ LiveInstance liveInstance = _accessor.getProperty(_keyBuilder.liveInstance(_instanceName));
+
+ // New live instance ephemeral node with FROZEN status
+ Assert.assertFalse(_originSession.equals(liveInstance.getEphemeralOwner()));
+ Assert.assertEquals(liveInstance.getStatus(), LiveInstance.LiveInstanceStatus.PAUSED);
+
+ // New session path does not exist since no current state carry over for the current session.
+ Assert.assertFalse(
+ _gZkClient.exists(_keyBuilder.currentStates(_instanceName, currentSession).getPath()));
+ // Old session CS still exist.
+ Assert.assertTrue(
+ _gZkClient.exists(_keyBuilder.currentStates(_instanceName, _originSession).getPath()));
+ }
+
+ @Test(dependsOnMethods = "testHandleNewSessionWhenFrozen")
+ public void testUnfreezeParticipant() throws Exception {
+ Message unfreezeMessage = MessageUtil
+ .createStatusChangeMessage(LiveInstance.LiveInstanceStatus.PAUSED,
+ LiveInstance.LiveInstanceStatus.NORMAL, _manager.getInstanceName(),
+ _manager.getSessionId(), _instanceName, _participants[0].getSessionId());
+ List<PropertyKey> keys = Collections
+ .singletonList(_keyBuilder.message(unfreezeMessage.getTgtName(), unfreezeMessage.getId()));
+
+ boolean[] success = _accessor.createChildren(keys, Collections.singletonList(unfreezeMessage));
+ Assert.assertTrue(success[0]);
+
+ // Live instance status is NORMAL, but set to null value in both memory and zk.
+ // After live instance status is updated, the process is completed.
+ verifyLiveInstanceStatus(_participants[0], null);
+ // Unfreeze message is correctly deleted
+ Assert.assertNull(
+ _accessor.getProperty(_keyBuilder.message(_instanceName, unfreezeMessage.getId())));
+
+ // current state is carried over
+ List<CurrentState> curStates = _accessor
+ .getChildValues(_keyBuilder.currentStates(_instanceName, _participants[0].getSessionId()),
+ false);
+ Assert.assertFalse(curStates.isEmpty());
+ // The original current states are deleted.
+ Assert.assertFalse(
+ _gZkClient.exists(_keyBuilder.currentStates(_instanceName, _originSession).getPath()));
+
+ // current states should be the same as the original current states
+ // with CS carry-over when unfreezing
+ Assert.assertTrue(verifyCurrentStates(_originCurStates, curStates));
+ }
+
+ private void verifyLiveInstanceStatus(MockParticipantManager participant,
+ LiveInstance.LiveInstanceStatus status) throws Exception {
+ // Live instance status is frozen in both memory and zk
+ Assert.assertTrue(TestHelper.verify(() -> {
+ LiveInstance.LiveInstanceStatus inMemoryLiveInstanceStatus =
+ ((DefaultMessagingService) participant.getMessagingService()).getExecutor()
+ .getLiveInstanceStatus();
+ return inMemoryLiveInstanceStatus == status;
+ }, TestHelper.WAIT_DURATION));
+
+ Assert.assertTrue(TestHelper.verify(() -> {
+ LiveInstance liveInstance =
+ _accessor.getProperty(_keyBuilder.liveInstance(participant.getInstanceName()));
+ return liveInstance.getStatus() == status;
+ }, TestHelper.WAIT_DURATION));
+ }
+
+ private boolean verifyCurrentStates(List<CurrentState> originCurStates,
+ List<CurrentState> curStates) {
+ for (CurrentState ocs : originCurStates) {
+ for (CurrentState cs : curStates) {
+ if (cs.getId().equals(ocs.getId())
+ && !cs.getPartitionStateMap().equals(ocs.getPartitionStateMap())) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private void freezeParticipant(MockParticipantManager participant) throws Exception {
+ Message freezeMessage = MessageUtil
+ .createStatusChangeMessage(LiveInstance.LiveInstanceStatus.NORMAL,
+ LiveInstance.LiveInstanceStatus.PAUSED, _manager.getInstanceName(),
+ _manager.getSessionId(), participant.getInstanceName(), participant.getSessionId());
+
+ List<PropertyKey> keys = Collections
+ .singletonList(_keyBuilder.message(freezeMessage.getTgtName(), freezeMessage.getId()));
+
+ boolean[] success = _accessor.createChildren(keys, Collections.singletonList(freezeMessage));
+ Assert.assertTrue(success[0]);
+
+ // Live instance status is frozen in both memory and zk
+ verifyLiveInstanceStatus(participant, LiveInstance.LiveInstanceStatus.PAUSED);
+ // Freeze message is correctly deleted
+ Assert.assertNull(_accessor
+ .getProperty(_keyBuilder.message(participant.getInstanceName(), freezeMessage.getId())));
+ }
+}