junkaixue commented on a change in pull request #1803:
URL: https://github.com/apache/helix/pull/1803#discussion_r658244096
##########
File path:
helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java
##########
@@ -647,9 +650,16 @@ private static PipelineRegistry
createManagementModeRegistry(String pipelineName
Pipeline dataRefresh = new Pipeline(pipelineName);
dataRefresh.addStage(new ReadClusterDataStage());
+ // data pre-process pipeline
+ Pipeline dataPreprocess = new Pipeline(pipelineName);
+ dataPreprocess.addStage(new ResourceComputationStage());
Review comment:
NIT: Do we need ResourceValidationStage? It seemed to bundle with
ResourceComputationStage.
##########
File path:
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.
Review comment:
Question: was this for participant pause message? I did not see the code
to do that? If you are trying to cache the messages sent out, I think the
original phase already did that.
##########
File path:
helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
##########
@@ -67,4 +67,8 @@ public Status getStatus() {
public Type getMode() {
return mode;
}
+
+ public boolean isFullyInNormalMode() {
+ return Type.NORMAL.equals(mode) && Status.COMPLETED.equals(status);
Review comment:
Just notice this. Naming convention. We should use prefix "_" as the
class member.
##########
File path:
helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
##########
@@ -0,0 +1,129 @@
+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);
+ }
+
+ // 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.getMode(),
cache.getEnabledLiveInstances(),
+ cache.getLiveInstances(), cache.getAllInstancesMessages(),
manager.getInstanceName(),
+ manager.getSessionId());
+ MessageOutput messageOutput =
+ event.getAttributeWithDefault(AttributeName.MESSAGES_ALL.name(), new
MessageOutput());
+ messageOutput.addStatusChangeMessages(messages);
+ event.addAttribute(AttributeName.MESSAGES_ALL.name(), messageOutput);
+ }
+ }
+
+ private List<Message>
generateStatusChangeMessages(ClusterManagementMode.Type managementMode,
+ Set<String> enabledLiveInstances, Map<String, LiveInstance>
liveInstanceMap,
+ Map<String, Collection<Message>> allInstanceMessages, String
managerInstance,
+ String managerSessionId) {
+ List<Message> messagesToSend = new ArrayList<>();
+
+ LiveInstanceStatus fromStatus = null;
Review comment:
I would suggest you do the similar logic as message generation. Use the
current status of instance as "FROMSTATE" and expected state as "TOSTATE". Then
check whether instance already in that state.
This logic could be more generic.
##########
File path:
helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
##########
@@ -0,0 +1,129 @@
+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);
+ }
+
+ // 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.getMode(),
cache.getEnabledLiveInstances(),
+ cache.getLiveInstances(), cache.getAllInstancesMessages(),
manager.getInstanceName(),
+ manager.getSessionId());
+ MessageOutput messageOutput =
+ event.getAttributeWithDefault(AttributeName.MESSAGES_ALL.name(), new
MessageOutput());
+ messageOutput.addStatusChangeMessages(messages);
+ event.addAttribute(AttributeName.MESSAGES_ALL.name(), messageOutput);
+ }
+ }
+
+ private List<Message>
generateStatusChangeMessages(ClusterManagementMode.Type managementMode,
Review comment:
When you generate it, you should add them to cache. Not let dispatch
phase to cache it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
