You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@helix.apache.org by "zpinto (via GitHub)" <gi...@apache.org> on 2024/07/23 19:22:12 UTC
[PR] [Draft] Implement Helix ST handling logic in Helix Gateway [helix]
zpinto opened a new pull request, #2845:
URL: https://github.com/apache/helix/pull/2845
### Issues
- [ ] My PR addresses the following Helix issues and references them in the PR description:
(#200 - Link your issue number here: You can write "Fixes #XXX". Please use the proper keyword so that the issue gets closed automatically. See https://docs.github.com/en/github/managing-your-work-on-github/linking-a-pull-request-to-an-issue
Any of the following keywords can be used: close, closes, closed, fix, fixes, fixed, resolve, resolves, resolved)
### Description
- [ ] Here are some details about my PR, including screenshots of any UI changes:
(Write a concise description including what, why, how)
### 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.
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "zpinto (via GitHub)" <gi...@apache.org>.
zpinto commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697769176
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,246 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
+
+ boolean success = future.get();
+ if (!success) {
+ throw new Exception("Failed to transition to state " + toState);
+ }
+
+ updateState(resourceId, shardId, toState);
+ } finally {
+ _stateTransitionResultMap.remove(transitionId);
+ }
+ }
+
+ @Override
+ public void handleStateTransitionError(Message message, StateTransitionError error) {
+ // Remove the stateTransitionResultMap future for the message
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+
+ // Remove the future from the stateTransitionResultMap since we are no longer able
+ // to process the state transition due to participant manager either timing out
+ // or failing to process the state transition
+ _stateTransitionResultMap.remove(transitionId);
+
+ // Set the replica state to ERROR
+ updateState(resourceId, shardId, HelixDefinedState.ERROR.name());
+
+ // Notify the HelixGatewayParticipantClient that it is in ERROR state
Review Comment:
I think we should hold off on this for now. We may want to have a different type of message because it is not an actual ST to ERROR. Let's discuss it more and add implementation in a following PR.
--
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
Re: [PR] [Draft] Implement Helix ST handling logic in Helix Gateway [helix]
Posted by "junkaixue (via GitHub)" <gi...@apache.org>.
junkaixue commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1688749341
##########
helix-gateway/src/main/java/org/apache/helix/gateway/constant/TransitionType.java:
##########
@@ -1,5 +1,5 @@
package org.apache.helix.gateway.constant;
-public enum MessageType {
+public enum TransitionType {
Review Comment:
I think Xiaoyuan refactored this out?
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1693481471
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,206 @@
+package org.apache.helix.gateway.participant;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixGatewayMultiTopStateStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.constant.MessageType;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+public class HelixGatewayParticipant implements HelixGatewayMultiTopStateStateTransitionProcessor {
Review Comment:
nit: doc
##########
helix-gateway/src/main/java/org/apache/helix/gateway/grpcservice/HelixGatewayServiceGrpcService.java:
##########
@@ -73,17 +75,20 @@ public void onCompleted() {
/**
* Send state transition message to the instance.
* The instance must already have established a connection to the gateway service.
- * @param instanceName
- * @return
+ * @param instanceName the instance name to send the message to
+ * @param messageType the type of the message
+ * @param message the message to convert to the transition message
*/
@Override
- public boolean sendStateTransitionMessage(String instanceName) {
+ public void sendStateTransitionMessage(String instanceName, MessageType messageType,
Review Comment:
Maybe we do not need to pass the MessageType here. we could read the from and f=to state In translation util. Message type also confuses with org.apache.helix.model.Message.MessageType.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,206 @@
+package org.apache.helix.gateway.participant;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixGatewayMultiTopStateStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.constant.MessageType;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+public class HelixGatewayParticipant implements HelixGatewayMultiTopStateStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processMultiTopStateModelStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(
+ _participantManager.getInstanceName(),
+ determineTransitionType(resourceId, shardId, toState), message);
+
+ boolean success = future.get();
+ if (!success) {
+ throw new Exception("Failed to transition to state " + toState);
+ }
+
+ updateState(resourceId, shardId, toState);
+ } finally {
+ _stateTransitionResultMap.remove(transitionId);
+ }
+ }
+
+ @Override
+ public void handleStateTransitionError(Message message, StateTransitionError error) {
+ // Remove the stateTransitionResultMap future for the message
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+
+ // Remove the future from the stateTransitionResultMap since we are no longer able
+ // to process the state transition due to participant manager either timing out
+ // or failing to process the state transition
+ _stateTransitionResultMap.remove(transitionId);
+
+ // Set the replica state to ERROR
+ updateState(resourceId, shardId, HelixDefinedState.ERROR.name());
+
+ // Notify the HelixGatewayParticipantClient that it is in ERROR state
+ // TODO: We need a better way than sending the state transition with a toState of ERROR
+ }
+
+ /**
+ * Completes the state transition with the given transitionId.
+ *
+ * @param transitionId the transitionId to complete
+ * @param isSuccess whether the state transition was successful
+ */
+ public void completeStateTransition(String transitionId, boolean isSuccess) {
+ CompletableFuture<Boolean> future = _stateTransitionResultMap.get(transitionId);
+ if (future != null) {
+ future.complete(isSuccess);
+ }
+ }
+
+ private boolean isCurrentStateAlreadyTarget(String resourceId, String shardId,
+ String targetState) {
+ return (_shardStateMap.containsKey(resourceId) && _shardStateMap.get(resourceId)
+ .containsKey(shardId) && _shardStateMap.get(resourceId).get(shardId).equals(targetState))
+ || targetState.equals(HelixDefinedState.DROPPED.name());
+ }
+
+ private MessageType determineTransitionType(String resourceId, String shardId,
Review Comment:
maybe move to util? Then it can be shared with the translator util as well.
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697859886
##########
helix-gateway/src/main/java/org/apache/helix/gateway/grpcservice/HelixGatewayServiceGrpcService.java:
##########
@@ -92,17 +94,20 @@ public void onCompleted() {
/**
* Send state transition message to the instance.
* The instance must already have established a connection to the gateway service.
- * @param instanceName
- * @return
+ *
+ * @param instanceName the instance name to send the message to
+ * @param currentState the current state of shard
+ * @param message the message to convert to the transition message
*/
@Override
- public boolean sendStateTransitionMessage(String instanceName) {
+ public void sendStateTransitionMessage(String instanceName, String currentState,
Review Comment:
Why we need currentState here?
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,245 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
+
+ if (!future.get()) {
+ throw new Exception("Failed to transition to state " + toState);
+ }
+
+ updateState(resourceId, shardId, toState);
+ } finally {
+ _stateTransitionResultMap.remove(transitionId);
+ }
+ }
+
+ @Override
+ public void handleStateTransitionError(Message message, StateTransitionError error) {
+ // Remove the stateTransitionResultMap future for the message
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+
+ // Remove the future from the stateTransitionResultMap since we are no longer able
+ // to process the state transition due to participant manager either timing out
+ // or failing to process the state transition
+ _stateTransitionResultMap.remove(transitionId);
+
+ // Set the replica state to ERROR
+ updateState(resourceId, shardId, HelixDefinedState.ERROR.name());
+
+ // Notify the HelixGatewayParticipantClient that it is in ERROR state
+ // TODO: We need a better way than sending the state transition with a toState of ERROR
+ }
+
+ /**
+ * Get the instance name of the participant.
+ *
+ * @return participant instance name
+ */
+ public String getInstanceName() {
+ return _participantManager.getInstanceName();
+ }
+
+ /**
+ * Completes the state transition with the given transitionId.
+ *
+ * @param transitionId the transitionId to complete
+ * @param isSuccess whether the state transition was successful
+ */
+ public void completeStateTransition(String transitionId, boolean isSuccess) {
+ CompletableFuture<Boolean> future = _stateTransitionResultMap.get(transitionId);
+ if (future != null) {
+ future.complete(isSuccess);
+ }
+ }
+
+ private boolean isCurrentStateAlreadyTarget(String resourceId, String shardId,
+ String targetState) {
+ return getCurrentState(resourceId, shardId).equals(targetState);
+ }
+
+ @VisibleForTesting
+ public Map<String, Map<String, String>> getShardStateMap() {
+ return _shardStateMap;
+ }
+
+ /**
+ * Get the current state of the shard.
+ *
+ * @param resourceId the resource id
+ * @param shardId the shard id
+ * @return the current state of the shard or DROPPED if it does not exist
+ */
+ public String getCurrentState(String resourceId, String shardId) {
+ return getShardStateMap().getOrDefault(resourceId, Collections.emptyMap())
+ .getOrDefault(shardId, HelixDefinedState.DROPPED.name());
Review Comment:
Returning DROPPED as default state is a bit wired. Maybe we can have a special case when target state is DROPPED state?
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,246 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
Review Comment:
Then probably we dont need current state in `sendStateTransitionMessage `?
##########
helix-gateway/src/main/java/org/apache/helix/gateway/api/participant/HelixStateTransitionProcessor.java:
##########
@@ -19,6 +19,26 @@
* under the License.
*/
-public enum MessageStatus {
- SUCCESS, FAILURE
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * Process state transition message.
+ */
+public interface HelixStateTransitionProcessor {
Review Comment:
Just wondering, why we want to have the interface? We have "HelixGatewayServiceProcessor" so that user can have other endpoint than grpc service. Are we expecting other implementations?
##########
helix-gateway/src/main/java/org/apache/helix/gateway/util/StateTransitionMessageTranslateUtil.java:
##########
@@ -33,27 +36,63 @@
public final class StateTransitionMessageTranslateUtil {
+ /**
+ * Determine the transition type based on the current state and the target state.
+ *
+ * @param currentState current state
+ * @param toState target state
+ * @return TransitionType
+ */
+ public static HelixGatewayServiceOuterClass.SingleTransitionMessage.TransitionType translateStatesToTransitionType(
+ String currentState, String toState) {
+ boolean isCurrentDropped = HelixDefinedState.DROPPED.name().equals(currentState);
+ boolean isToStateDropped = HelixDefinedState.DROPPED.name().equals(toState);
+
+ if (isToStateDropped && !isCurrentDropped) {
+ return HelixGatewayServiceOuterClass.SingleTransitionMessage.TransitionType.DELETE_SHARD;
+ }
+ if (!isToStateDropped && isCurrentDropped) {
+ return HelixGatewayServiceOuterClass.SingleTransitionMessage.TransitionType.ADD_SHARD;
+ }
+ return HelixGatewayServiceOuterClass.SingleTransitionMessage.TransitionType.CHANGE_ROLE;
+ }
- public static TransitionMessage translateSTMsgToTransitionMessage() {
- return null;
+ /**
+ * Translate from user sent Message to Helix Gateway Service event.
Review Comment:
I think the comment is off.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,245 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
+
+ if (!future.get()) {
+ throw new Exception("Failed to transition to state " + toState);
+ }
+
+ updateState(resourceId, shardId, toState);
+ } finally {
+ _stateTransitionResultMap.remove(transitionId);
+ }
+ }
+
+ @Override
+ public void handleStateTransitionError(Message message, StateTransitionError error) {
+ // Remove the stateTransitionResultMap future for the message
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+
+ // Remove the future from the stateTransitionResultMap since we are no longer able
+ // to process the state transition due to participant manager either timing out
+ // or failing to process the state transition
+ _stateTransitionResultMap.remove(transitionId);
+
+ // Set the replica state to ERROR
+ updateState(resourceId, shardId, HelixDefinedState.ERROR.name());
+
+ // Notify the HelixGatewayParticipantClient that it is in ERROR state
+ // TODO: We need a better way than sending the state transition with a toState of ERROR
+ }
+
+ /**
+ * Get the instance name of the participant.
+ *
+ * @return participant instance name
+ */
+ public String getInstanceName() {
+ return _participantManager.getInstanceName();
+ }
+
+ /**
+ * Completes the state transition with the given transitionId.
+ *
+ * @param transitionId the transitionId to complete
+ * @param isSuccess whether the state transition was successful
+ */
+ public void completeStateTransition(String transitionId, boolean isSuccess) {
+ CompletableFuture<Boolean> future = _stateTransitionResultMap.get(transitionId);
+ if (future != null) {
+ future.complete(isSuccess);
+ }
+ }
+
+ private boolean isCurrentStateAlreadyTarget(String resourceId, String shardId,
+ String targetState) {
+ return getCurrentState(resourceId, shardId).equals(targetState);
+ }
+
+ @VisibleForTesting
+ public Map<String, Map<String, String>> getShardStateMap() {
+ return _shardStateMap;
+ }
+
+ /**
+ * Get the current state of the shard.
+ *
+ * @param resourceId the resource id
+ * @param shardId the shard id
+ * @return the current state of the shard or DROPPED if it does not exist
+ */
+ public String getCurrentState(String resourceId, String shardId) {
+ return getShardStateMap().getOrDefault(resourceId, Collections.emptyMap())
+ .getOrDefault(shardId, HelixDefinedState.DROPPED.name());
+ }
+
+ private void updateState(String resourceId, String shardId, String state) {
+ if (state.equals(HelixDefinedState.DROPPED.name())) {
+ getShardStateMap().computeIfPresent(resourceId, (k, v) -> {
+ v.remove(shardId);
+ if (v.isEmpty()) {
+ return null;
+ }
+ return v;
+ });
+ } else {
+ getShardStateMap().computeIfAbsent(resourceId, k -> new ConcurrentHashMap<>())
+ .put(shardId, state);
+ }
+ }
+
+ public void disconnect() {
+ _participantManager.disconnect();
+ }
+
+ public static class Builder {
+ private final HelixGatewayServiceProcessor _helixGatewayServiceProcessor;
+ private final String _instanceName;
+ private final String _clusterName;
+ private final String _zkAddress;
+ private final List<String> _multiTopStateModelDefinitions;
+ private final Map<String, Map<String, String>> _initialShardStateMap;
+
+ public Builder(HelixGatewayServiceProcessor helixGatewayServiceProcessor, String instanceName,
+ String clusterName, String zkAddress) {
+ _helixGatewayServiceProcessor = helixGatewayServiceProcessor;
+ _instanceName = instanceName;
+ _clusterName = clusterName;
+ _zkAddress = zkAddress;
+ _multiTopStateModelDefinitions = new ArrayList<>();
+ _initialShardStateMap = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a multi-top state model definition to the participant to be registered in the
+ * participant's state machine engine.
+ *
+ * @param stateModelDefinitionName the state model definition name to add (should be multi-top
+ * state model)
+ * @return the builder
+ */
+ public Builder addMultiTopStateStateModelDefinition(String stateModelDefinitionName) {
+ // TODO: Add validation that the state model definition is a multi-top state model
+ _multiTopStateModelDefinitions.add(stateModelDefinitionName);
+ return this;
+ }
+
+ /**
+ * Add initial shard state to the participant. This is used to initialize the participant with
+ * the initial state of the shards in order to reduce unnecessary state transitions from being
+ * forwarded to the participant.
+ *
+ * @param initialShardStateMap the initial shard state map to add
+ * @return the Builder
+ */
+ public Builder setInitialShardState(Map<String, Map<String, String>> initialShardStateMap) {
+ // TODO: Add handling for shard states that where never assigned to the participant since
+ // the participant was last online.
+ // deep copy into the initialShardStateMap into concurrent hash map
+ initialShardStateMap.forEach((resourceId, shardStateMap) -> {
+ _initialShardStateMap.put(resourceId, new ConcurrentHashMap<>(shardStateMap));
+ });
+
+ return this;
+ }
+
+ /**
+ * Build the HelixGatewayParticipant. This will create a HelixManager for the participant and
+ * connect to the Helix cluster. The participant will be registered with the multi-top state
+ * model definitions and initialized with the initial shard state map.
+ *
+ * @return the HelixGatewayParticipant
+ */
+ public HelixGatewayParticipant build() {
+ HelixManager participantManager =
+ new ZKHelixManager(_clusterName, _instanceName, InstanceType.PARTICIPANT, _zkAddress);
+ HelixGatewayParticipant participant =
+ new HelixGatewayParticipant(_helixGatewayServiceProcessor, participantManager,
+ _initialShardStateMap);
+ _multiTopStateModelDefinitions.forEach(
+ stateModelDefinition -> participantManager.getStateMachineEngine()
+ .registerStateModelFactory(stateModelDefinition,
+ new HelixGatewayMultiTopStateStateModelFactory(participant)));
+ try {
+ participantManager.connect();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
Review Comment:
Maybe not runtime exception? If zk cannot be connected, maybe we dont want to disconnect gracefully instead of shutdown the service with exception?
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1693698665
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceManager.java:
##########
@@ -89,33 +91,47 @@ public void run() {
* It includes waiting for ZK connection, and also wait for previous LiveInstance to expire.
*/
class participantConnectionProcessor implements Runnable {
- GatewayServiceEvent _event;
+ private final GatewayServiceEvent _event;
- public participantConnectionProcessor(GatewayServiceEvent event) {
+ private participantConnectionProcessor(GatewayServiceEvent event) {
_event = event;
}
@Override
public void run() {
- HelixGatewayService helixGatewayService;
- _helixGatewayServiceMap.computeIfAbsent(_event.getClusterName(),
- k -> new HelixGatewayService(GatewayServiceManager.this, _event.getClusterName()));
- helixGatewayService = _helixGatewayServiceMap.get(_event.getClusterName());
if (_event.getEventType().equals(GatewayServiceEventType.CONNECT)) {
- helixGatewayService.registerParticipant();
+ createHelixGatewayParticipant(_event.getClusterName(), _event.getInstanceName(),
+ _event.getShardStateMap());
} else {
- helixGatewayService.deregisterParticipant(_event.getClusterName(), _event.getInstanceName());
+ removeHelixGatewayParticipant(_event.getClusterName(), _event.getInstanceName());
}
}
}
- @VisibleForTesting
- HelixGatewayServiceGrpcService getGrpcService() {
- return _grpcService;
+ private void createHelixGatewayParticipant(String clusterName, String instanceName,
+ Map<String, Map<String, String>> initialShardStateMap) {
+ // Create and add the participant to the participant map
+ HelixGatewayParticipant.Builder participantBuilder =
+ new HelixGatewayParticipant.Builder(_gatewayServiceProcessor, instanceName, clusterName,
+ _zkAddress).addInitialShardState(initialShardStateMap);
+ SUPPORTED_MULTI_STATE_MODEL_TYPES.forEach(
+ participantBuilder::addMultiTopStateStateModelDefinition);
+ _helixGatewayParticipantMap.computeIfAbsent(clusterName, k -> new ConcurrentHashMap<>())
+ .put(instanceName, participantBuilder.build());
+ }
+
+ private void removeHelixGatewayParticipant(String clusterName, String instanceName) {
+ // Disconnect and remove the participant from the participant map
+ HelixGatewayParticipant participant = getHelixGatewayParticipant(clusterName, instanceName);
+ if (participant != null) {
Review Comment:
can we have null check in getter?
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697706402
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,246 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
+
+ boolean success = future.get();
+ if (!success) {
+ throw new Exception("Failed to transition to state " + toState);
+ }
+
+ updateState(resourceId, shardId, toState);
+ } finally {
+ _stateTransitionResultMap.remove(transitionId);
+ }
+ }
+
+ @Override
+ public void handleStateTransitionError(Message message, StateTransitionError error) {
+ // Remove the stateTransitionResultMap future for the message
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+
+ // Remove the future from the stateTransitionResultMap since we are no longer able
+ // to process the state transition due to participant manager either timing out
+ // or failing to process the state transition
+ _stateTransitionResultMap.remove(transitionId);
+
+ // Set the replica state to ERROR
+ updateState(resourceId, shardId, HelixDefinedState.ERROR.name());
+
+ // Notify the HelixGatewayParticipantClient that it is in ERROR state
Review Comment:
should't we call " _gatewayServiceProcessor.sendStateTransitionMessage" here?
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697698123
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,246 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
Review Comment:
question. If we had a partiton being assigned to the instance, do we translate the Helix ST message "INIT->ONLINE" to "DROPPED->ONLINE"? That would be wired for user to process.
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu merged PR #2845:
URL: https://github.com/apache/helix/pull/2845
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on PR #2845:
URL: https://github.com/apache/helix/pull/2845#issuecomment-2261448579
Had an offline discussion with @zpinto.
LGTM, Please address comments based on our discussion. :D
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "zpinto (via GitHub)" <gi...@apache.org>.
zpinto commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697790847
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceManager.java:
##########
@@ -89,33 +91,47 @@ public void run() {
* It includes waiting for ZK connection, and also wait for previous LiveInstance to expire.
*/
class participantConnectionProcessor implements Runnable {
- GatewayServiceEvent _event;
+ private final GatewayServiceEvent _event;
- public participantConnectionProcessor(GatewayServiceEvent event) {
+ private participantConnectionProcessor(GatewayServiceEvent event) {
_event = event;
}
@Override
public void run() {
- HelixGatewayService helixGatewayService;
- _helixGatewayServiceMap.computeIfAbsent(_event.getClusterName(),
- k -> new HelixGatewayService(GatewayServiceManager.this, _event.getClusterName()));
- helixGatewayService = _helixGatewayServiceMap.get(_event.getClusterName());
if (_event.getEventType().equals(GatewayServiceEventType.CONNECT)) {
- helixGatewayService.registerParticipant();
+ createHelixGatewayParticipant(_event.getClusterName(), _event.getInstanceName(),
+ _event.getShardStateMap());
} else {
- helixGatewayService.deregisterParticipant(_event.getClusterName(), _event.getInstanceName());
+ removeHelixGatewayParticipant(_event.getClusterName(), _event.getInstanceName());
}
}
}
- @VisibleForTesting
- HelixGatewayServiceGrpcService getGrpcService() {
- return _grpcService;
+ private void createHelixGatewayParticipant(String clusterName, String instanceName,
+ Map<String, Map<String, String>> initialShardStateMap) {
+ // Create and add the participant to the participant map
+ HelixGatewayParticipant.Builder participantBuilder =
+ new HelixGatewayParticipant.Builder(_gatewayServiceProcessor, instanceName, clusterName,
+ _zkAddress).addInitialShardState(initialShardStateMap);
+ SUPPORTED_MULTI_STATE_MODEL_TYPES.forEach(
+ participantBuilder::addMultiTopStateStateModelDefinition);
+ _helixGatewayParticipantMap.computeIfAbsent(clusterName, k -> new ConcurrentHashMap<>())
+ .put(instanceName, participantBuilder.build());
+ }
+
+ private void removeHelixGatewayParticipant(String clusterName, String instanceName) {
+ // Disconnect and remove the participant from the participant map
+ HelixGatewayParticipant participant = getHelixGatewayParticipant(clusterName, instanceName);
+ if (participant != null) {
Review Comment:
I think we need the getter to return null, otherwise what will it return if null check is there?
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697699138
##########
helix-gateway/src/main/java/org/apache/helix/gateway/statemodel/HelixGatewayMultiTopStateStateModel.java:
##########
@@ -0,0 +1,60 @@
+package org.apache.helix.gateway.statemodel;
+
+/*
+ * 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.NotificationContext;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+import org.apache.helix.participant.statemachine.Transition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@StateModelInfo(initialState = "OFFLINE", states = {})
+public class HelixGatewayMultiTopStateStateModel extends StateModel {
+ private static final Logger _logger =
+ LoggerFactory.getLogger(HelixGatewayMultiTopStateStateModel.class);
+
+ private final HelixStateTransitionProcessor _stateTransitionProcessor;
+
+ public HelixGatewayMultiTopStateStateModel(
+ HelixStateTransitionProcessor stateTransitionProcessor) {
+ _stateTransitionProcessor = stateTransitionProcessor;
+ }
+
+ @Transition(to = "*", from = "*")
+ public void genericStateTransitionHandler(Message message, NotificationContext context)
+ throws Exception {
+ _stateTransitionProcessor.processStateTransitionMessage(message);
+ }
+
+ @Override
+ public void reset() {
+ // no-op we don't want to start from init state again.
+ }
+
+ @Override
+ public void rollbackOnError(Message message, NotificationContext context,
Review Comment:
Please correct me if I am wrong. We are not using this method now in out first version right?
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "zpinto (via GitHub)" <gi...@apache.org>.
zpinto commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697982618
##########
helix-gateway/src/main/java/org/apache/helix/gateway/grpcservice/HelixGatewayServiceGrpcService.java:
##########
@@ -92,17 +94,20 @@ public void onCompleted() {
/**
* Send state transition message to the instance.
* The instance must already have established a connection to the gateway service.
- * @param instanceName
- * @return
+ *
+ * @param instanceName the instance name to send the message to
+ * @param currentState the current state of shard
+ * @param message the message to convert to the transition message
*/
@Override
- public boolean sendStateTransitionMessage(String instanceName) {
+ public void sendStateTransitionMessage(String instanceName, String currentState,
Review Comment:
It is necessary to figure out the TransitionType. It is used to see if it is ADD, DELETE, or CHANGE_ROLE
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697699138
##########
helix-gateway/src/main/java/org/apache/helix/gateway/statemodel/HelixGatewayMultiTopStateStateModel.java:
##########
@@ -0,0 +1,60 @@
+package org.apache.helix.gateway.statemodel;
+
+/*
+ * 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.NotificationContext;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+import org.apache.helix.participant.statemachine.Transition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@StateModelInfo(initialState = "OFFLINE", states = {})
+public class HelixGatewayMultiTopStateStateModel extends StateModel {
+ private static final Logger _logger =
+ LoggerFactory.getLogger(HelixGatewayMultiTopStateStateModel.class);
+
+ private final HelixStateTransitionProcessor _stateTransitionProcessor;
+
+ public HelixGatewayMultiTopStateStateModel(
+ HelixStateTransitionProcessor stateTransitionProcessor) {
+ _stateTransitionProcessor = stateTransitionProcessor;
+ }
+
+ @Transition(to = "*", from = "*")
+ public void genericStateTransitionHandler(Message message, NotificationContext context)
+ throws Exception {
+ _stateTransitionProcessor.processStateTransitionMessage(message);
+ }
+
+ @Override
+ public void reset() {
+ // no-op we don't want to start from init state again.
+ }
+
+ @Override
+ public void rollbackOnError(Message message, NotificationContext context,
Review Comment:
Please correct me if I am wrong. We are not using this method now in out first version right?
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "zpinto (via GitHub)" <gi...@apache.org>.
zpinto commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697768612
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,246 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
Review Comment:
We don't send the from state, we are only sending the target state and the TransitionType so they have no exposure to the from state.
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "zpinto (via GitHub)" <gi...@apache.org>.
zpinto commented on PR #2845:
URL: https://github.com/apache/helix/pull/2845#issuecomment-2261544116
This PR is ready to be merged!
Comment:
- Add ST handling logic to support multi top state state model definitions without intermediary states
- Encapsulate the participant manager create, connect, disconnect logic in HelixGatewayParticipant
- Add replica state tracking in HelixGatewayParticipant
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "zpinto (via GitHub)" <gi...@apache.org>.
zpinto commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1699157425
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,245 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
+
+ if (!future.get()) {
+ throw new Exception("Failed to transition to state " + toState);
+ }
+
+ updateState(resourceId, shardId, toState);
+ } finally {
+ _stateTransitionResultMap.remove(transitionId);
+ }
+ }
+
+ @Override
+ public void handleStateTransitionError(Message message, StateTransitionError error) {
+ // Remove the stateTransitionResultMap future for the message
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+
+ // Remove the future from the stateTransitionResultMap since we are no longer able
+ // to process the state transition due to participant manager either timing out
+ // or failing to process the state transition
+ _stateTransitionResultMap.remove(transitionId);
+
+ // Set the replica state to ERROR
+ updateState(resourceId, shardId, HelixDefinedState.ERROR.name());
+
+ // Notify the HelixGatewayParticipantClient that it is in ERROR state
+ // TODO: We need a better way than sending the state transition with a toState of ERROR
+ }
+
+ /**
+ * Get the instance name of the participant.
+ *
+ * @return participant instance name
+ */
+ public String getInstanceName() {
+ return _participantManager.getInstanceName();
+ }
+
+ /**
+ * Completes the state transition with the given transitionId.
+ *
+ * @param transitionId the transitionId to complete
+ * @param isSuccess whether the state transition was successful
+ */
+ public void completeStateTransition(String transitionId, boolean isSuccess) {
+ CompletableFuture<Boolean> future = _stateTransitionResultMap.get(transitionId);
+ if (future != null) {
+ future.complete(isSuccess);
+ }
+ }
+
+ private boolean isCurrentStateAlreadyTarget(String resourceId, String shardId,
+ String targetState) {
+ return getCurrentState(resourceId, shardId).equals(targetState);
+ }
+
+ @VisibleForTesting
+ public Map<String, Map<String, String>> getShardStateMap() {
+ return _shardStateMap;
+ }
+
+ /**
+ * Get the current state of the shard.
+ *
+ * @param resourceId the resource id
+ * @param shardId the shard id
+ * @return the current state of the shard or DROPPED if it does not exist
+ */
+ public String getCurrentState(String resourceId, String shardId) {
+ return getShardStateMap().getOrDefault(resourceId, Collections.emptyMap())
+ .getOrDefault(shardId, HelixDefinedState.DROPPED.name());
Review Comment:
Changed to new gateway default of UNASSIGNED
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697690752
##########
helix-gateway/src/main/java/org/apache/helix/gateway/api/service/HelixGatewayServiceProcessor.java:
##########
@@ -19,11 +19,22 @@
* under the License.
*/
+import org.apache.helix.model.Message;
+
/**
- * Translate from/to GRPC function call to Helix Gateway Service event.
+ * Helix Gateway Service Processor interface allows sending state transition messages to
+ * participants through service implementing this interface.
*/
public interface HelixGatewayServiceProcessor {
- public boolean sendStateTransitionMessage( String instanceName);
+ /**
+ * Send a state transition message to a remote participant.
+ *
+ * @param instanceName the name of the participant
+ * @param currentState the current state of the participant
Review Comment:
the current state of the shard?
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "zpinto (via GitHub)" <gi...@apache.org>.
zpinto commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697981290
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,246 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
Review Comment:
It is necessary to figure out the TransitionType. It is used to see if it is ADD, DELETE, or CHANGE_ROLE
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697699138
##########
helix-gateway/src/main/java/org/apache/helix/gateway/statemodel/HelixGatewayMultiTopStateStateModel.java:
##########
@@ -0,0 +1,60 @@
+package org.apache.helix.gateway.statemodel;
+
+/*
+ * 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.NotificationContext;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+import org.apache.helix.participant.statemachine.Transition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@StateModelInfo(initialState = "OFFLINE", states = {})
+public class HelixGatewayMultiTopStateStateModel extends StateModel {
+ private static final Logger _logger =
+ LoggerFactory.getLogger(HelixGatewayMultiTopStateStateModel.class);
+
+ private final HelixStateTransitionProcessor _stateTransitionProcessor;
+
+ public HelixGatewayMultiTopStateStateModel(
+ HelixStateTransitionProcessor stateTransitionProcessor) {
+ _stateTransitionProcessor = stateTransitionProcessor;
+ }
+
+ @Transition(to = "*", from = "*")
+ public void genericStateTransitionHandler(Message message, NotificationContext context)
+ throws Exception {
+ _stateTransitionProcessor.processStateTransitionMessage(message);
+ }
+
+ @Override
+ public void reset() {
+ // no-op we don't want to start from init state again.
+ }
+
+ @Override
+ public void rollbackOnError(Message message, NotificationContext context,
Review Comment:
Please correct me if I am wrong. We are not invoke this method now in out first version right?
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1697682957
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,246 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
+
+ boolean success = future.get();
+ if (!success) {
+ throw new Exception("Failed to transition to state " + toState);
Review Comment:
nit: if(future.get())
--
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
Re: [PR] Implement Helix ST handling logic and HelixGatewayParticipant [helix]
Posted by "zpinto (via GitHub)" <gi...@apache.org>.
zpinto commented on code in PR #2845:
URL: https://github.com/apache/helix/pull/2845#discussion_r1699157869
##########
helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java:
##########
@@ -0,0 +1,245 @@
+package org.apache.helix.gateway.participant;
+
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixManager;
+import org.apache.helix.InstanceType;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
+import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.statemodel.HelixGatewayMultiTopStateStateModelFactory;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
+/**
+ * HelixGatewayParticipant encapsulates the Helix Participant Manager and handles tracking the state
+ * of a remote participant connected to the Helix Gateway Service. It processes state transitions
+ * for the participant and updates the state of the participant's shards upon successful state
+ * transitions signaled by remote participant.
+ */
+public class HelixGatewayParticipant implements HelixStateTransitionProcessor {
+ private final HelixGatewayServiceProcessor _gatewayServiceProcessor;
+ private final HelixManager _participantManager;
+ private final Map<String, Map<String, String>> _shardStateMap;
+ private final Map<String, CompletableFuture<Boolean>> _stateTransitionResultMap;
+
+ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProcessor,
+ HelixManager participantManager, Map<String, Map<String, String>> initialShardStateMap) {
+ _gatewayServiceProcessor = gatewayServiceProcessor;
+ _participantManager = participantManager;
+ _shardStateMap = initialShardStateMap;
+ _stateTransitionResultMap = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public void processStateTransitionMessage(Message message) throws Exception {
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+ String toState = message.getToState();
+
+ try {
+ if (isCurrentStateAlreadyTarget(resourceId, shardId, toState)) {
+ return;
+ }
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ _stateTransitionResultMap.put(transitionId, future);
+ _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+ getCurrentState(resourceId, shardId), message);
+
+ if (!future.get()) {
+ throw new Exception("Failed to transition to state " + toState);
+ }
+
+ updateState(resourceId, shardId, toState);
+ } finally {
+ _stateTransitionResultMap.remove(transitionId);
+ }
+ }
+
+ @Override
+ public void handleStateTransitionError(Message message, StateTransitionError error) {
+ // Remove the stateTransitionResultMap future for the message
+ String transitionId = message.getMsgId();
+ String resourceId = message.getResourceName();
+ String shardId = message.getPartitionName();
+
+ // Remove the future from the stateTransitionResultMap since we are no longer able
+ // to process the state transition due to participant manager either timing out
+ // or failing to process the state transition
+ _stateTransitionResultMap.remove(transitionId);
+
+ // Set the replica state to ERROR
+ updateState(resourceId, shardId, HelixDefinedState.ERROR.name());
+
+ // Notify the HelixGatewayParticipantClient that it is in ERROR state
+ // TODO: We need a better way than sending the state transition with a toState of ERROR
+ }
+
+ /**
+ * Get the instance name of the participant.
+ *
+ * @return participant instance name
+ */
+ public String getInstanceName() {
+ return _participantManager.getInstanceName();
+ }
+
+ /**
+ * Completes the state transition with the given transitionId.
+ *
+ * @param transitionId the transitionId to complete
+ * @param isSuccess whether the state transition was successful
+ */
+ public void completeStateTransition(String transitionId, boolean isSuccess) {
+ CompletableFuture<Boolean> future = _stateTransitionResultMap.get(transitionId);
+ if (future != null) {
+ future.complete(isSuccess);
+ }
+ }
+
+ private boolean isCurrentStateAlreadyTarget(String resourceId, String shardId,
+ String targetState) {
+ return getCurrentState(resourceId, shardId).equals(targetState);
+ }
+
+ @VisibleForTesting
+ public Map<String, Map<String, String>> getShardStateMap() {
+ return _shardStateMap;
+ }
+
+ /**
+ * Get the current state of the shard.
+ *
+ * @param resourceId the resource id
+ * @param shardId the shard id
+ * @return the current state of the shard or DROPPED if it does not exist
+ */
+ public String getCurrentState(String resourceId, String shardId) {
+ return getShardStateMap().getOrDefault(resourceId, Collections.emptyMap())
+ .getOrDefault(shardId, HelixDefinedState.DROPPED.name());
+ }
+
+ private void updateState(String resourceId, String shardId, String state) {
+ if (state.equals(HelixDefinedState.DROPPED.name())) {
+ getShardStateMap().computeIfPresent(resourceId, (k, v) -> {
+ v.remove(shardId);
+ if (v.isEmpty()) {
+ return null;
+ }
+ return v;
+ });
+ } else {
+ getShardStateMap().computeIfAbsent(resourceId, k -> new ConcurrentHashMap<>())
+ .put(shardId, state);
+ }
+ }
+
+ public void disconnect() {
+ _participantManager.disconnect();
+ }
+
+ public static class Builder {
+ private final HelixGatewayServiceProcessor _helixGatewayServiceProcessor;
+ private final String _instanceName;
+ private final String _clusterName;
+ private final String _zkAddress;
+ private final List<String> _multiTopStateModelDefinitions;
+ private final Map<String, Map<String, String>> _initialShardStateMap;
+
+ public Builder(HelixGatewayServiceProcessor helixGatewayServiceProcessor, String instanceName,
+ String clusterName, String zkAddress) {
+ _helixGatewayServiceProcessor = helixGatewayServiceProcessor;
+ _instanceName = instanceName;
+ _clusterName = clusterName;
+ _zkAddress = zkAddress;
+ _multiTopStateModelDefinitions = new ArrayList<>();
+ _initialShardStateMap = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a multi-top state model definition to the participant to be registered in the
+ * participant's state machine engine.
+ *
+ * @param stateModelDefinitionName the state model definition name to add (should be multi-top
+ * state model)
+ * @return the builder
+ */
+ public Builder addMultiTopStateStateModelDefinition(String stateModelDefinitionName) {
+ // TODO: Add validation that the state model definition is a multi-top state model
+ _multiTopStateModelDefinitions.add(stateModelDefinitionName);
+ return this;
+ }
+
+ /**
+ * Add initial shard state to the participant. This is used to initialize the participant with
+ * the initial state of the shards in order to reduce unnecessary state transitions from being
+ * forwarded to the participant.
+ *
+ * @param initialShardStateMap the initial shard state map to add
+ * @return the Builder
+ */
+ public Builder setInitialShardState(Map<String, Map<String, String>> initialShardStateMap) {
+ // TODO: Add handling for shard states that where never assigned to the participant since
+ // the participant was last online.
+ // deep copy into the initialShardStateMap into concurrent hash map
+ initialShardStateMap.forEach((resourceId, shardStateMap) -> {
+ _initialShardStateMap.put(resourceId, new ConcurrentHashMap<>(shardStateMap));
+ });
+
+ return this;
+ }
+
+ /**
+ * Build the HelixGatewayParticipant. This will create a HelixManager for the participant and
+ * connect to the Helix cluster. The participant will be registered with the multi-top state
+ * model definitions and initialized with the initial shard state map.
+ *
+ * @return the HelixGatewayParticipant
+ */
+ public HelixGatewayParticipant build() {
+ HelixManager participantManager =
+ new ZKHelixManager(_clusterName, _instanceName, InstanceType.PARTICIPANT, _zkAddress);
+ HelixGatewayParticipant participant =
+ new HelixGatewayParticipant(_helixGatewayServiceProcessor, participantManager,
+ _initialShardStateMap);
+ _multiTopStateModelDefinitions.forEach(
+ stateModelDefinition -> participantManager.getStateMachineEngine()
+ .registerStateModelFactory(stateModelDefinition,
+ new HelixGatewayMultiTopStateStateModelFactory(participant)));
+ try {
+ participantManager.connect();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
Review Comment:
Left comment to address this in next PR.
--
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