You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@helix.apache.org by GitBox <gi...@apache.org> on 2021/06/24 01:12:21 UTC
[GitHub] [helix] huizhilu opened a new pull request #1803: Add message generation logic for management pipeline
huizhilu opened a new pull request #1803:
URL: https://github.com/apache/helix/pull/1803
### Issues
- [x] My PR addresses the following Helix issues and references them in the PR description:
Resolves #1774
### Description
- [x] Here are some details about my PR, including screenshots of any UI changes:
In cluster freeze mode, controller sends freeze/unfreeze and pending ST cancellation messages to participants.
In this PR, 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.
### Tests
- [ ] The following tests are written for this issue:
(List the names of added unit/integration tests)
- The following is the result of the "mvn test" command on the appropriate module:
(If CI test fails due to known issue, please specify the issue and test PR locally. Then copy & paste the result of "mvn test" to here.)
### Changes that Break Backward Compatibility (Optional)
- My PR contains changes that break backward compatibility or previous assumptions for certain methods or API. They include:
(Consider including all behavior changes for public methods or API. Also include these changes in merge description so that other developers are aware of these changes. This allows them to make relevant code changes in feature branches accounting for the new method/API behavior.)
### Documentation (Optional)
- In case of new functionality, my PR adds documentation in the following wiki page:
(Link the GitHub wiki you added)
### Commits
- My commits all reference appropriate Apache Helix GitHub issues in their subject lines. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
1. Subject is limited to 50 characters (not including Jira issue reference)
1. Subject does not end with a period
1. Subject uses the imperative mood ("add", not "adding")
1. Body wraps at 72 characters
1. Body explains "what" and "why", not "how"
### Code Quality
- My diff has been formatted using helix-style.xml
(helix-style-intellij.xml if IntelliJ IDE is used)
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] huizhilu merged pull request #1803: Add message generation logic for management pipeline
Posted by GitBox <gi...@apache.org>.
huizhilu merged pull request #1803:
URL: https://github.com/apache/helix/pull/1803
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] huizhilu merged pull request #1803: Add message generation logic for management pipeline
Posted by GitBox <gi...@apache.org>.
huizhilu merged pull request #1803:
URL: https://github.com/apache/helix/pull/1803
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1803: Add message generation logic for management pipeline
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1803:
URL: https://github.com/apache/helix/pull/1803#discussion_r658374766
##########
File path: helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
##########
@@ -67,4 +69,26 @@ public Status getStatus() {
public Type getMode() {
return mode;
}
+
+ public boolean isFullyInNormalMode() {
Review comment:
Question, do we have a term for partially normal status, or what does that stand for?
I assume it should be,
1. NORMAL, COMPLETE,
2. CLUSTER_PAUSE, IN_PROGRESS
3. CLUSTER_PAUSE, COMPLETE
4. CLUSTER_PAUSE, IN_PROGRESS ---> it is recovering from pause, but since it is still partially paused, I think to mark it CLUSTER_PAUSE is more appropriate.
5. NORMAL, COMPLETE
##########
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.
+ 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()) {
Review comment:
Why it is determined here instead of ManagementModeStage?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
##########
@@ -0,0 +1,118 @@
+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);
Review comment:
One suggestion, put the current state to best possible state transformation here so it won't be done unless necessary.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
##########
@@ -66,6 +68,14 @@ public void process(ClusterEvent event) throws Exception {
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 =
Review comment:
I assume this is only used to cancel pending ST. Calculate it here might be confusing because it is very far away from where it is used.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] huizhilu commented on a change in pull request #1803: Add message generation logic for management pipeline
Posted by GitBox <gi...@apache.org>.
huizhilu commented on a change in pull request #1803:
URL: https://github.com/apache/helix/pull/1803#discussion_r659118244
##########
File path: helix-core/src/main/java/org/apache/helix/api/status/ClusterManagementMode.java
##########
@@ -67,4 +69,26 @@ public Status getStatus() {
public Type getMode() {
return mode;
}
+
+ public boolean isFullyInNormalMode() {
Review comment:
What I'd like to have is:
1. What is the desired/target mode?
2. What is the status, IN PROGRESS or COMPLETED?
`isFullyInNormalMode` means the cluster is stable in the normal mode. For the situation: recovering from pause, it should be IN_PROGRESS to NORMAL. It's already set to NORMAL mode, so we want to mark it NORMAL, IN PROGRESS.
##########
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.
+ 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()) {
Review comment:
We want to finish running the pipeline. And at the final step, decide whether if it can exit.
The pros:
* we don't need to throw an exception to terminate the pipeline.
* It makes sense to determine if it should switch at the final step after finishing all the previous checks.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/ManagementModeStage.java
##########
@@ -66,6 +68,14 @@ public void process(ClusterEvent event) throws Exception {
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 =
Review comment:
Replied in the above thread. I don't think it's that far. It's used in the next stage. This stage prepares required output for the next stage. It makes sense, also aligned with the default/task pipelines.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/ManagementMessageGenerationPhase.java
##########
@@ -0,0 +1,118 @@
+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);
Review comment:
I'd like to make the stages clearly do what it is supposed to do:
* ManagementModeStage: check status, prepare for message generation.
* ManagementMessageGenerationPhase: only create messages, based on the results of the previous stages.
I don't want to mix the responsibilities. So I think putting the logic: CS -> best possible in management mode stage is appropriate as preparation. It also aligns with the logic as the default pipeline: messages are generated purely based on the previous results.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] huizhilu commented on a change in pull request #1803: Add message generation logic for management pipeline
Posted by GitBox <gi...@apache.org>.
huizhilu commented on a change in pull request #1803:
URL: https://github.com/apache/helix/pull/1803#discussion_r658311558
##########
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:
I'll leave it with the final naming refactoring such as `cluster_pause` -> `cluster_freeze` mode, which won't need any review. Tracking: https://github.com/apache/helix/issues/1805
##########
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:
No. ResourceValidation is to match the ideal state rules, which is not required management pipeline. We can assume in management pipeline, the ideal states are validated since management pipeline is switched from default pipeline.
The pipeline switch logic from default to management is ResourceValidation. To avoid handling it in management pipeline and make the logic simple, I don't want to add ResourceValidation.
##########
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:
It's sent by `super.sendMessages(manager.getHelixDataAccessor(), messageOutput.getStatusChangeMessages());`
Synced offline. Resolved.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] junkaixue commented on a change in pull request #1803: Add message generation logic for management pipeline
Posted by GitBox <gi...@apache.org>.
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] huizhilu commented on pull request #1803: Add message generation logic for management pipeline
Posted by GitBox <gi...@apache.org>.
huizhilu commented on pull request #1803:
URL: https://github.com/apache/helix/pull/1803#issuecomment-870257866
This PR is ready to be merged, approved by @junkaixue
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] junkaixue commented on pull request #1803: Add message generation logic for management pipeline
Posted by GitBox <gi...@apache.org>.
junkaixue commented on pull request #1803:
URL: https://github.com/apache/helix/pull/1803#issuecomment-869378651
Overall, it looks good to me. Please resolve JJ's comments.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] huizhilu commented on pull request #1803: Add message generation logic for management pipeline
Posted by GitBox <gi...@apache.org>.
huizhilu commented on pull request #1803:
URL: https://github.com/apache/helix/pull/1803#issuecomment-870257866
This PR is ready to be merged, approved by @junkaixue
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org