You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@helix.apache.org by "xyuanlu (via GitHub)" <gi...@apache.org> on 2024/07/23 07:57:26 UTC
[PR] Gatewayservice 2 [helix]
xyuanlu opened a new pull request, #2844:
URL: https://github.com/apache/helix/pull/2844
### 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 GatewayServiceManager [helix]
Posted by "junkaixue (via GitHub)" <gi...@apache.org>.
junkaixue commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1692052245
##########
helix-gateway/src/main/java/org/apache/helix/gateway/util/StateTransitionMessageTranslateUtil.java:
##########
@@ -1,17 +1,70 @@
package org.apache.helix.gateway.util;
-import org.apache.helix.gateway.service.GatewayServiceManager;
-import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.helix.gateway.constant.GatewayServiceEventType;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardState;
import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardStateMessage;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardTransitionStatus;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage;
public final class StateTransitionMessageTranslateUtil {
- public static TransitionMessage translateSTMsgToProto() {
+ public static TransitionMessage translateSTMsgToTransitionMessage() {
return null;
}
- public static GatewayServiceManager.GateWayServiceEvent translateProtoToSTMsg(ShardStateMessage message) {
- return null;
+ /**
+ * Translate from user sent ShardStateMessage message to Helix Gateway Service event.
+ */
+ public static GatewayServiceEvent translateShardStateMessageToEvent(ShardStateMessage request) {
+
+ GatewayServiceEvent.GateWayServiceEventBuilder builder;
+ if (request.hasShardState()) { // init connection to gateway service
+ ShardState shardState = request.getShardState();
+ Map<String, String> shardStateMap = new HashMap<>();
+ for (HelixGatewayServiceOuterClass.SingleResourceState resourceState : shardState.getResourceStateList()) {
+ for (HelixGatewayServiceOuterClass.SingleShardState state : resourceState.getShardStatesList()) {
+ shardStateMap.put(resourceState.getResource() + "_" + state.getShardName(), state.getCurrentState());
+ }
+ }
+ builder = new GatewayServiceEvent.GateWayServiceEventBuilder(GatewayServiceEventType.CONNECT).setClusterName(
+ shardState.getClusterName()).setParticipantName(shardState.getInstanceName());
+ } else {
+ ShardTransitionStatus shardTransitionStatus = request.getShardTransitionStatus();
+ // this is status update for established connection
+ List<HelixGatewayServiceOuterClass.SingleShardTransitionStatus> status =
+ shardTransitionStatus.getShardTransitionStatusList();
+ List<GatewayServiceEvent.StateTransitionResult> stResult = new ArrayList<>();
+ for (HelixGatewayServiceOuterClass.SingleShardTransitionStatus shardTransition : status) {
+ String stateTransitionId = shardTransition.getTransitionID();
+ String stateTransitionStatus = shardTransition.getCurrentState();
+ String shardState = shardTransition.getCurrentState();
Review Comment:
As I said, no creation for these variables when it only used once...
--
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 GatewayServiceManager [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1692188521
##########
helix-gateway/src/main/java/org/apache/helix/gateway/grpcservice/HelixGatewayServiceGrpcService.java:
##########
@@ -0,0 +1,118 @@
+package org.apache.helix.gateway.grpcservice;
+
+import io.grpc.stub.StreamObserver;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import org.apache.helix.gateway.service.GatewayServiceManager;
+import org.apache.helix.gateway.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.util.PerKeyLockRegistry;
Review Comment:
yep. https://github.com/apache/helix/tree/helix-gateway-service/helix-gateway/src/main/java/org/apache/helix/gateway/util
##########
helix-gateway/src/main/java/org/apache/helix/gateway/util/StateTransitionMessageTranslateUtil.java:
##########
@@ -1,17 +1,70 @@
package org.apache.helix.gateway.util;
-import org.apache.helix.gateway.service.GatewayServiceManager;
-import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.helix.gateway.constant.GatewayServiceEventType;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardState;
import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardStateMessage;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardTransitionStatus;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage;
public final class StateTransitionMessageTranslateUtil {
- public static TransitionMessage translateSTMsgToProto() {
+ public static TransitionMessage translateSTMsgToTransitionMessage() {
return null;
}
- public static GatewayServiceManager.GateWayServiceEvent translateProtoToSTMsg(ShardStateMessage message) {
- return null;
+ /**
+ * Translate from user sent ShardStateMessage message to Helix Gateway Service event.
+ */
+ public static GatewayServiceEvent translateShardStateMessageToEvent(ShardStateMessage request) {
+
+ GatewayServiceEvent.GateWayServiceEventBuilder builder;
+ if (request.hasShardState()) { // init connection to gateway service
+ ShardState shardState = request.getShardState();
+ Map<String, String> shardStateMap = new HashMap<>();
+ for (HelixGatewayServiceOuterClass.SingleResourceState resourceState : shardState.getResourceStateList()) {
+ for (HelixGatewayServiceOuterClass.SingleShardState state : resourceState.getShardStatesList()) {
+ shardStateMap.put(resourceState.getResource() + "_" + state.getShardName(), state.getCurrentState());
+ }
+ }
+ builder = new GatewayServiceEvent.GateWayServiceEventBuilder(GatewayServiceEventType.CONNECT).setClusterName(
+ shardState.getClusterName()).setParticipantName(shardState.getInstanceName());
+ } else {
+ ShardTransitionStatus shardTransitionStatus = request.getShardTransitionStatus();
+ // this is status update for established connection
+ List<HelixGatewayServiceOuterClass.SingleShardTransitionStatus> status =
+ shardTransitionStatus.getShardTransitionStatusList();
+ List<GatewayServiceEvent.StateTransitionResult> stResult = new ArrayList<>();
+ for (HelixGatewayServiceOuterClass.SingleShardTransitionStatus shardTransition : status) {
+ String stateTransitionId = shardTransition.getTransitionID();
+ String stateTransitionStatus = shardTransition.getCurrentState();
+ String shardState = shardTransition.getCurrentState();
Review Comment:
Updated.
--
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 GatewayServiceManager [helix]
Posted by "junkaixue (via GitHub)" <gi...@apache.org>.
junkaixue commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1688396795
##########
helix-gateway/src/main/java/org/apache/helix/gateway/util/StateTransitionMessageTranslateUtil.java:
##########
@@ -11,7 +15,24 @@ public static TransitionMessage translateSTMsgToProto() {
return null;
}
- public static GatewayServiceManager.GateWayServiceEvent translateProtoToSTMsg(ShardStateMessage message) {
- return null;
+ public static GatewayServiceEvent translateProtoToEvent(ShardStateMessage request) {
+ String instanceName = request.getInstanceName();
+ String clusterName = request.getClusterName();
+
+ Map<String, String> shardStateMap = new HashMap<>();
+ for (HelixGatewayServiceOuterClass.SingleShardState shardTransitionStatus : request.getShardStateList()) {
+ shardStateMap.put(shardTransitionStatus.getShardName(), shardTransitionStatus.getCurrentState());
+ }
+ GatewayServiceEvent.GateWayServiceEventBuilder builder =
+ new GatewayServiceEvent.GateWayServiceEventBuilder(GatewayServiceEventType.CONNECT).setClusterName(clusterName);
Review Comment:
Why this is connect not update?
##########
helix-gateway/src/main/java/org/apache/helix/gateway/constant/ShardState.java:
##########
@@ -0,0 +1,5 @@
+package org.apache.helix.gateway.constant;
+
+public enum ShardState {
Review Comment:
Why not use state model one? How about error?
##########
helix-gateway/src/main/java/org/apache/helix/gateway/util/StateTransitionMessageTranslateUtil.java:
##########
@@ -11,7 +15,24 @@ public static TransitionMessage translateSTMsgToProto() {
return null;
}
- public static GatewayServiceManager.GateWayServiceEvent translateProtoToSTMsg(ShardStateMessage message) {
- return null;
+ public static GatewayServiceEvent translateProtoToEvent(ShardStateMessage request) {
Review Comment:
I don't get this util at all...
If you would like to differentiate the event type, you can just make a full util and selectively pass in null value for the arguments are not necessary.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceEvent.java:
##########
@@ -0,0 +1,95 @@
+package org.apache.helix.gateway.service;
+
+import java.util.Map;
+import org.apache.helix.gateway.constant.GatewayServiceEventType;
+
+
+public class GatewayServiceEvent {
+ // event type
+ GatewayServiceEventType _eventType;
+ // event data
+ String _clusterName;
+ String _instanceName;
+ Map<String, Map<String, String>> _shardStateMap;
+
+ StateTransitionResult _StateTransitionResult;
+
+ public class StateTransitionResult {
+ String stateTransitionId;
+ String stateTransitionStatus;
+ String shardState;
+
+ public StateTransitionResult(String stateTransitionId, String stateTransitionStatus, String shardState) {
+ this.stateTransitionId = stateTransitionId;
+ this.stateTransitionStatus = stateTransitionStatus;
+ this.shardState = shardState;
+ }
+ }
+
+ private GatewayServiceEvent(GatewayServiceEventType eventType, String clusterName, String instanceName) {
+ this._eventType = eventType;
+ this._clusterName = clusterName;
+ this._instanceName = instanceName;
+ }
+
+ private GatewayServiceEvent(GatewayServiceEventType eventType, String clusterName, String instanceName, Map<String, Map<String, String>> shardStateMap) {
+ this._eventType = eventType;
+ this._clusterName = clusterName;
+ this._instanceName = instanceName;
+ this._shardStateMap = shardStateMap;
+ }
+
+ private GatewayServiceEvent(GatewayServiceEventType eventType, String clusterName, String instanceName, String stateTransitionId,
+ String stateTransitionStatus, String shardState) {
+ this._eventType = eventType;
+ this._clusterName = clusterName;
+ this._instanceName = instanceName;
+ this._StateTransitionResult = new StateTransitionResult(stateTransitionId, stateTransitionStatus, shardState);
+ }
+
Review Comment:
If we have a builder, no need to have so many constructor.
Also even as public constructor, we should call "this(xxxxxxx)" to reduce redundant code.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceManager.java:
##########
@@ -26,40 +29,85 @@ public class GatewayServiceManager {
// TODO: add thread pool for init
// single thread tp for update
- public enum EventType {
- CONNECT, // init connection to gateway service
- UPDATE, // update state transition result
- DISCONNECT // shutdown connection to gateway service.
- }
+ // Map of cluster to HelixGatewayService
- public class GateWayServiceEvent {
- // event type
- EventType eventType;
- // event data
- String clusterName;
- String participantName;
+ // a single thread tp for event processing
+ ExecutorService _participantStateTransitionResultUpdator;
- // todo: add more fields
- }
+ // a thread pool for participant connecting
+ ExecutorService _participantConnectingTp;
+
+ // grpc service
+ HelixGatewayServiceGrpcService _grpcService;
public GatewayServiceManager() {
_helixGatewayServiceMap = new ConcurrentHashMap<>();
+ _participantStateTransitionResultUpdator = Executors.newSingleThreadExecutor();
+ _participantConnectingTp = Executors.newFixedThreadPool(10); // todo: make it configurable
+ _grpcService = new HelixGatewayServiceGrpcService(this);
}
public AtomicBoolean sendTransitionRequestToApplicationInstance() {
return null;
}
- public void updateShardState() {
-
+ public void newGatewayServiceEvent(GatewayServiceEvent event) {
+ if (event.getEventType().equals(GatewayServiceEventType.UPDATE)) {
+ _participantStateTransitionResultUpdator.submit(new shardStateUpdator(event));
+ } else {
+ _participantConnectingTp.submit(new participantConnectionProcessor(event));
+ }
}
- public void newParticipantConnecting() {
-
+ /**
+ * Update shard state
+ */
+ class shardStateUpdator implements Runnable {
+
+ GatewayServiceEvent _event;
+
+ public shardStateUpdator(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService = _helixGatewayServiceMap.get(_event._clusterName);
+ if (helixGatewayService == null) {
+ // TODO: return error code and throw exception.
+ return;
+ }
+ helixGatewayService.receiveSTResponse();
+ }
}
- public void participantDisconnected() {
-
+ /**
+ * Create HelixGatewayService instance and register it to the manager.
+ * It includes waiting for ZK connection, and also wait for previous Liveinstance to expire.
+ */
+ class participantConnectionProcessor implements Runnable {
+ GatewayServiceEvent _event;
+
+ public participantConnectionProcessor(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService;
+ if (_event.getEventType().equals(GatewayServiceEventType.CONNECT)) {
Review Comment:
Where is update???
##########
helix-gateway/src/main/java/org/apache/helix/gateway/constant/GatewayServiceEventType.java:
##########
@@ -0,0 +1,7 @@
+package org.apache.helix.gateway.constant;
+
+public enum GatewayServiceEventType {
Review Comment:
Why we need this new type? I remember we have one in your last pr.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/util/StateTransitionMessageTranslateUtil.java:
##########
@@ -11,7 +15,24 @@ public static TransitionMessage translateSTMsgToProto() {
return null;
}
- public static GatewayServiceManager.GateWayServiceEvent translateProtoToSTMsg(ShardStateMessage message) {
- return null;
+ public static GatewayServiceEvent translateProtoToEvent(ShardStateMessage request) {
Review Comment:
Why it is called photo to event? I don't get it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
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 GatewayServiceManager [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1688381674
##########
helix-gateway/src/main/java/org/apache/helix/gateway/grpcservice/HelixGatewayServiceGrpcService.java:
##########
@@ -0,0 +1,92 @@
+package org.apache.helix.gateway.grpcservice;
+
+import io.grpc.stub.StreamObserver;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.helix.gateway.constant.GatewayServiceEventType;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import org.apache.helix.gateway.service.GatewayServiceManager;
+import org.apache.helix.gateway.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.util.StateTransitionMessageTranslateUtil;
+import proto.org.apache.helix.gateway.*;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.*;
+
+import java.util.Map;
+
+
+/**
+ * Helix Gateway Service GRPC UI implementation.
+ */
+public class HelixGatewayServiceGrpcService extends HelixGatewayServiceGrpc.HelixGatewayServiceImplBase
+ implements HelixGatewayServiceProcessor {
+
+ Map<String, StreamObserver<TransitionMessage>> _observerMap =
+ new ConcurrentHashMap<String, StreamObserver<TransitionMessage>>();
+ Map<StreamObserver<TransitionMessage>, String> _reversedObserverMap =
+ new ConcurrentHashMap<StreamObserver<TransitionMessage>, String>();
+
+ Map<String, String> instanceToClusterMap = new ConcurrentHashMap<String, String>();
+ GatewayServiceManager _manager;
+
+ public HelixGatewayServiceGrpcService(GatewayServiceManager manager) {
+ // register manager
+ _manager = manager;
+ }
+
+ @Override
+ public StreamObserver<proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardStateMessage> report(
+ StreamObserver<proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage> responseObserver) {
+
+ return new StreamObserver<ShardStateMessage>() {
+
+ @Override
+ public void onNext(ShardStateMessage request) {
+ // called when a client sends a message
+ //....
+ String instanceName = request.getInstanceName();
+ String clusterName = request.getClusterName();
+ GatewayServiceEvent event;
+ instanceToClusterMap.put(instanceName, clusterName);
+ if (!_observerMap.containsValue(instanceName)) {
+ // new connection
+ updateObserver(instanceName, responseObserver);
+ }
Review Comment:
Good point. will update.
--
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 GatewayServiceManager [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1688780575
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceManager.java:
##########
@@ -26,40 +29,85 @@ public class GatewayServiceManager {
// TODO: add thread pool for init
// single thread tp for update
- public enum EventType {
- CONNECT, // init connection to gateway service
- UPDATE, // update state transition result
- DISCONNECT // shutdown connection to gateway service.
- }
+ // Map of cluster to HelixGatewayService
- public class GateWayServiceEvent {
- // event type
- EventType eventType;
- // event data
- String clusterName;
- String participantName;
+ // a single thread tp for event processing
+ ExecutorService _participantStateTransitionResultUpdator;
- // todo: add more fields
- }
+ // a thread pool for participant connecting
+ ExecutorService _participantConnectingTp;
+
+ // grpc service
+ HelixGatewayServiceGrpcService _grpcService;
public GatewayServiceManager() {
_helixGatewayServiceMap = new ConcurrentHashMap<>();
+ _participantStateTransitionResultUpdator = Executors.newSingleThreadExecutor();
+ _participantConnectingTp = Executors.newFixedThreadPool(10); // todo: make it configurable
+ _grpcService = new HelixGatewayServiceGrpcService(this);
}
public AtomicBoolean sendTransitionRequestToApplicationInstance() {
return null;
}
- public void updateShardState() {
-
+ public void newGatewayServiceEvent(GatewayServiceEvent event) {
+ if (event.getEventType().equals(GatewayServiceEventType.UPDATE)) {
+ _participantStateTransitionResultUpdator.submit(new shardStateUpdator(event));
+ } else {
+ _participantConnectingTp.submit(new participantConnectionProcessor(event));
+ }
}
- public void newParticipantConnecting() {
-
+ /**
+ * Update shard state
+ */
+ class shardStateUpdator implements Runnable {
+
+ GatewayServiceEvent _event;
+
+ public shardStateUpdator(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService = _helixGatewayServiceMap.get(_event._clusterName);
+ if (helixGatewayService == null) {
+ // TODO: return error code and throw exception.
+ return;
+ }
+ helixGatewayService.receiveSTResponse();
+ }
}
- public void participantDisconnected() {
-
+ /**
+ * Create HelixGatewayService instance and register it to the manager.
+ * It includes waiting for ZK connection, and also wait for previous Liveinstance to expire.
+ */
+ class participantConnectionProcessor implements Runnable {
+ GatewayServiceEvent _event;
+
+ public participantConnectionProcessor(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService;
+ if (_event.getEventType().equals(GatewayServiceEventType.CONNECT)) {
Review Comment:
This is only for init connection and disconnect. update if handled in shardStateUpdator
--
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 GatewayServiceManager [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on PR #2844:
URL: https://github.com/apache/helix/pull/2844#issuecomment-2251468466
This change is ready to be merged. Approved by @junkaixue
commit message: Implement GatewayServiceManager.
--
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 GatewayServiceManager [helix]
Posted by "junkaixue (via GitHub)" <gi...@apache.org>.
junkaixue commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1688972784
##########
helix-gateway/src/main/java/org/apache/helix/gateway/grpcservice/HelixGatewayServiceGrpcService.java:
##########
@@ -0,0 +1,122 @@
+package org.apache.helix.gateway.grpcservice;
+
+import io.grpc.stub.StreamObserver;
+import java.util.HashMap;
+import java.util.concurrent.ConcurrentHashMap;
+
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import org.apache.helix.gateway.service.GatewayServiceManager;
+import org.apache.helix.gateway.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.util.StateTransitionMessageTranslateUtil;
+import proto.org.apache.helix.gateway.*;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.*;
+
+import java.util.Map;
+
+
+/**
+ * Helix Gateway Service GRPC UI implementation.
+ */
+public class HelixGatewayServiceGrpcService extends HelixGatewayServiceGrpc.HelixGatewayServiceImplBase
+ implements HelixGatewayServiceProcessor {
+
+ // Map to store the observer for each instance
+ Map<String, StreamObserver<TransitionMessage>> _observerMap = new HashMap<>();
+ // A reverse map to store the instance name for each observer. It is used to find the instance when connection is closed.
+ Map<StreamObserver<TransitionMessage>, Pair<String, String>> _reversedObserverMap = new HashMap<>();
+
+ ReadWriteLock _observerMapLock;
+ GatewayServiceManager _manager;
+
+ public HelixGatewayServiceGrpcService(GatewayServiceManager manager) {
+ // register manager
+ _manager = manager;
+ _observerMapLock = new ReentrantReadWriteLock();
+ }
+
+ @Override
+ public StreamObserver<proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardStateMessage> report(
+ StreamObserver<proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage> responseObserver) {
+
+ return new StreamObserver<ShardStateMessage>() {
+
+ @Override
+ public void onNext(ShardStateMessage request) {
+ if (request.hasShardState()) {
+ ShardState shardState = request.getShardState();
+ // Process shardState
+ updateObserver(shardState.getInstanceName(), shardState.getClusterName(), responseObserver);
+ }
+ _manager.newGatewayServiceEvent(StateTransitionMessageTranslateUtil.translateShardStateMessageToEvent(request));
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ onClientClose(responseObserver);
+ }
+
+ @Override
+ public void onCompleted() {
+ // called when the client completes
+ onClientClose(responseObserver);
+ }
+ };
+ }
+
+ @Override
+ public boolean sendStateTransitionMessage(String instanceName) {
+ StreamObserver<TransitionMessage> observer;
+ observer = _observerMap.get(instanceName);
+ _observerMapLock.readLock().lock();
Review Comment:
use fine-grained lock for each entry.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceEvent.java:
##########
@@ -0,0 +1,77 @@
+package org.apache.helix.gateway.service;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.helix.gateway.constant.GatewayServiceEventType;
+import org.apache.helix.gateway.constant.MessageStatus;
+
+
+public class GatewayServiceEvent {
Review Comment:
white space.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceManager.java:
##########
@@ -16,50 +19,85 @@
*/
public class GatewayServiceManager {
-
- HelixGatewayServiceService _helixGatewayServiceService;
-
- HelixGatewayServiceProcessor _helixGatewayServiceProcessor;
-
Map<String, HelixGatewayService> _helixGatewayServiceMap;
- // TODO: add thread pool for init
- // single thread tp for update
+ // a single thread tp for event processing
+ ExecutorService _participantStateTransitionResultUpdator;
- public enum EventType {
- CONNECT, // init connection to gateway service
- UPDATE, // update state transition result
- DISCONNECT // shutdown connection to gateway service.
- }
+ // a thread pool for participant connecting
+ ExecutorService _participantConnectingTp;
- public class GateWayServiceEvent {
- // event type
- EventType eventType;
- // event data
- String clusterName;
- String participantName;
-
- // todo: add more fields
- }
+ // grpc service
+ HelixGatewayServiceGrpcService _grpcService;
public GatewayServiceManager() {
_helixGatewayServiceMap = new ConcurrentHashMap<>();
+ _participantStateTransitionResultUpdator = Executors.newSingleThreadExecutor();
+ _participantConnectingTp = Executors.newFixedThreadPool(10); // todo: make it configurable
Review Comment:
Define variable. use DEFAULT one now.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/grpcservice/HelixGatewayServiceGrpcService.java:
##########
@@ -0,0 +1,122 @@
+package org.apache.helix.gateway.grpcservice;
+
+import io.grpc.stub.StreamObserver;
+import java.util.HashMap;
+import java.util.concurrent.ConcurrentHashMap;
+
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import org.apache.helix.gateway.service.GatewayServiceManager;
+import org.apache.helix.gateway.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.util.StateTransitionMessageTranslateUtil;
+import proto.org.apache.helix.gateway.*;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.*;
+
+import java.util.Map;
+
+
+/**
+ * Helix Gateway Service GRPC UI implementation.
+ */
+public class HelixGatewayServiceGrpcService extends HelixGatewayServiceGrpc.HelixGatewayServiceImplBase
+ implements HelixGatewayServiceProcessor {
+
+ // Map to store the observer for each instance
+ Map<String, StreamObserver<TransitionMessage>> _observerMap = new HashMap<>();
+ // A reverse map to store the instance name for each observer. It is used to find the instance when connection is closed.
+ Map<StreamObserver<TransitionMessage>, Pair<String, String>> _reversedObserverMap = new HashMap<>();
+
+ ReadWriteLock _observerMapLock;
+ GatewayServiceManager _manager;
+
+ public HelixGatewayServiceGrpcService(GatewayServiceManager manager) {
+ // register manager
+ _manager = manager;
+ _observerMapLock = new ReentrantReadWriteLock();
+ }
+
+ @Override
+ public StreamObserver<proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardStateMessage> report(
+ StreamObserver<proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage> responseObserver) {
+
+ return new StreamObserver<ShardStateMessage>() {
+
+ @Override
+ public void onNext(ShardStateMessage request) {
+ if (request.hasShardState()) {
+ ShardState shardState = request.getShardState();
+ // Process shardState
+ updateObserver(shardState.getInstanceName(), shardState.getClusterName(), responseObserver);
+ }
+ _manager.newGatewayServiceEvent(StateTransitionMessageTranslateUtil.translateShardStateMessageToEvent(request));
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ onClientClose(responseObserver);
+ }
+
+ @Override
+ public void onCompleted() {
+ // called when the client completes
+ onClientClose(responseObserver);
+ }
+ };
+ }
+
+ @Override
+ public boolean sendStateTransitionMessage(String instanceName) {
+ StreamObserver<TransitionMessage> observer;
+ observer = _observerMap.get(instanceName);
+ _observerMapLock.readLock().lock();
Review Comment:
also no need for read-write lock. read-write lock to guarantee data read after write, not apply for this case.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceEvent.java:
##########
@@ -0,0 +1,77 @@
+package org.apache.helix.gateway.service;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.helix.gateway.constant.GatewayServiceEventType;
+import org.apache.helix.gateway.constant.MessageStatus;
+
+
+public class GatewayServiceEvent {
+ // event type
+ GatewayServiceEventType _eventType;
+ // event data
+ String _clusterName;
+ String _instanceName;
+ Map<String, Map<String, String>> _shardStateMap;
Review Comment:
Make it private.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceManager.java:
##########
@@ -16,50 +19,85 @@
*/
public class GatewayServiceManager {
-
- HelixGatewayServiceService _helixGatewayServiceService;
-
- HelixGatewayServiceProcessor _helixGatewayServiceProcessor;
-
Map<String, HelixGatewayService> _helixGatewayServiceMap;
- // TODO: add thread pool for init
- // single thread tp for update
+ // a single thread tp for event processing
+ ExecutorService _participantStateTransitionResultUpdator;
- public enum EventType {
- CONNECT, // init connection to gateway service
- UPDATE, // update state transition result
- DISCONNECT // shutdown connection to gateway service.
- }
+ // a thread pool for participant connecting
+ ExecutorService _participantConnectingTp;
- public class GateWayServiceEvent {
- // event type
- EventType eventType;
- // event data
- String clusterName;
- String participantName;
-
- // todo: add more fields
- }
+ // grpc service
+ HelixGatewayServiceGrpcService _grpcService;
public GatewayServiceManager() {
_helixGatewayServiceMap = new ConcurrentHashMap<>();
+ _participantStateTransitionResultUpdator = Executors.newSingleThreadExecutor();
+ _participantConnectingTp = Executors.newFixedThreadPool(10); // todo: make it configurable
+ _grpcService = new HelixGatewayServiceGrpcService(this);
}
public AtomicBoolean sendTransitionRequestToApplicationInstance() {
return null;
}
- public void updateShardState() {
-
+ public void newGatewayServiceEvent(GatewayServiceEvent event) {
+ if (event.getEventType().equals(GatewayServiceEventType.UPDATE)) {
+ _participantStateTransitionResultUpdator.submit(new shardStateUpdator(event));
+ } else {
+ _participantConnectingTp.submit(new participantConnectionProcessor(event));
+ }
}
- public void newParticipantConnecting() {
-
+ /**
+ * Update shard state
+ */
+ class shardStateUpdator implements Runnable {
+
+ GatewayServiceEvent _event;
+
+ public shardStateUpdator(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService = _helixGatewayServiceMap.get(_event._clusterName);
+ if (helixGatewayService == null) {
+ // TODO: return error code and throw exception.
+ return;
+ }
+ helixGatewayService.receiveSTResponse();
+ }
}
- public void participantDisconnected() {
-
+ /**
+ * Create HelixGatewayService instance and register it to the manager.
+ * It includes waiting for ZK connection, and also wait for previous Liveinstance to expire.
+ */
+ class participantConnectionProcessor implements Runnable {
+ GatewayServiceEvent _event;
+
+ public participantConnectionProcessor(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService;
+ if (_event.getEventType().equals(GatewayServiceEventType.CONNECT)) {
+ if (_helixGatewayServiceMap.containsKey(_event._clusterName)) {
+ helixGatewayService = _helixGatewayServiceMap.get(_event._clusterName);
+ } else {
+ helixGatewayService = new HelixGatewayService(GatewayServiceManager.this, _event._clusterName);
+ _helixGatewayServiceMap.put(_event._clusterName, helixGatewayService);
+ }
+ helixGatewayService.registerParticipant();
+ } else { // disconnected
+ helixGatewayService = _helixGatewayServiceMap.get(_event._clusterName);
Review Comment:
NPE?
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceManager.java:
##########
@@ -16,50 +19,85 @@
*/
public class GatewayServiceManager {
-
- HelixGatewayServiceService _helixGatewayServiceService;
-
- HelixGatewayServiceProcessor _helixGatewayServiceProcessor;
-
Map<String, HelixGatewayService> _helixGatewayServiceMap;
- // TODO: add thread pool for init
- // single thread tp for update
+ // a single thread tp for event processing
+ ExecutorService _participantStateTransitionResultUpdator;
- public enum EventType {
- CONNECT, // init connection to gateway service
- UPDATE, // update state transition result
- DISCONNECT // shutdown connection to gateway service.
- }
+ // a thread pool for participant connecting
+ ExecutorService _participantConnectingTp;
- public class GateWayServiceEvent {
- // event type
- EventType eventType;
- // event data
- String clusterName;
- String participantName;
-
- // todo: add more fields
- }
+ // grpc service
+ HelixGatewayServiceGrpcService _grpcService;
public GatewayServiceManager() {
_helixGatewayServiceMap = new ConcurrentHashMap<>();
+ _participantStateTransitionResultUpdator = Executors.newSingleThreadExecutor();
+ _participantConnectingTp = Executors.newFixedThreadPool(10); // todo: make it configurable
+ _grpcService = new HelixGatewayServiceGrpcService(this);
}
public AtomicBoolean sendTransitionRequestToApplicationInstance() {
return null;
}
- public void updateShardState() {
-
+ public void newGatewayServiceEvent(GatewayServiceEvent event) {
+ if (event.getEventType().equals(GatewayServiceEventType.UPDATE)) {
+ _participantStateTransitionResultUpdator.submit(new shardStateUpdator(event));
+ } else {
+ _participantConnectingTp.submit(new participantConnectionProcessor(event));
+ }
}
- public void newParticipantConnecting() {
-
+ /**
+ * Update shard state
+ */
+ class shardStateUpdator implements Runnable {
+
+ GatewayServiceEvent _event;
+
+ public shardStateUpdator(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService = _helixGatewayServiceMap.get(_event._clusterName);
+ if (helixGatewayService == null) {
+ // TODO: return error code and throw exception.
+ return;
+ }
+ helixGatewayService.receiveSTResponse();
+ }
}
- public void participantDisconnected() {
-
+ /**
+ * Create HelixGatewayService instance and register it to the manager.
+ * It includes waiting for ZK connection, and also wait for previous Liveinstance to expire.
+ */
+ class participantConnectionProcessor implements Runnable {
+ GatewayServiceEvent _event;
+
+ public participantConnectionProcessor(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService;
+ if (_event.getEventType().equals(GatewayServiceEventType.CONNECT)) {
+ if (_helixGatewayServiceMap.containsKey(_event._clusterName)) {
+ helixGatewayService = _helixGatewayServiceMap.get(_event._clusterName);
+ } else {
+ helixGatewayService = new HelixGatewayService(GatewayServiceManager.this, _event._clusterName);
+ _helixGatewayServiceMap.put(_event._clusterName, helixGatewayService);
Review Comment:
if (! contains) {
new object
}
do xxxxxx
--
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 GatewayServiceManager [helix]
Posted by "frankmu (via GitHub)" <gi...@apache.org>.
frankmu commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1688378477
##########
helix-gateway/src/main/java/org/apache/helix/gateway/grpcservice/HelixGatewayServiceGrpcService.java:
##########
@@ -0,0 +1,92 @@
+package org.apache.helix.gateway.grpcservice;
+
+import io.grpc.stub.StreamObserver;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.helix.gateway.constant.GatewayServiceEventType;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import org.apache.helix.gateway.service.GatewayServiceManager;
+import org.apache.helix.gateway.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.util.StateTransitionMessageTranslateUtil;
+import proto.org.apache.helix.gateway.*;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.*;
+
+import java.util.Map;
+
+
+/**
+ * Helix Gateway Service GRPC UI implementation.
+ */
+public class HelixGatewayServiceGrpcService extends HelixGatewayServiceGrpc.HelixGatewayServiceImplBase
+ implements HelixGatewayServiceProcessor {
+
+ Map<String, StreamObserver<TransitionMessage>> _observerMap =
+ new ConcurrentHashMap<String, StreamObserver<TransitionMessage>>();
+ Map<StreamObserver<TransitionMessage>, String> _reversedObserverMap =
+ new ConcurrentHashMap<StreamObserver<TransitionMessage>, String>();
+
+ Map<String, String> instanceToClusterMap = new ConcurrentHashMap<String, String>();
+ GatewayServiceManager _manager;
+
+ public HelixGatewayServiceGrpcService(GatewayServiceManager manager) {
+ // register manager
+ _manager = manager;
+ }
+
+ @Override
+ public StreamObserver<proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardStateMessage> report(
+ StreamObserver<proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage> responseObserver) {
+
+ return new StreamObserver<ShardStateMessage>() {
+
+ @Override
+ public void onNext(ShardStateMessage request) {
+ // called when a client sends a message
+ //....
+ String instanceName = request.getInstanceName();
+ String clusterName = request.getClusterName();
+ GatewayServiceEvent event;
+ instanceToClusterMap.put(instanceName, clusterName);
+ if (!_observerMap.containsValue(instanceName)) {
+ // new connection
+ updateObserver(instanceName, responseObserver);
+ }
Review Comment:
Not sure if we want to make this class thread safe. If yes, this might cause race condition: https://stackoverflow.com/questions/15850803/concurrenthashmap-putifabsent-method-functionality
--
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 GatewayServiceManager [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1690626665
##########
helix-gateway/src/main/java/org/apache/helix/gateway/util/StateTransitionMessageTranslateUtil.java:
##########
@@ -1,17 +1,71 @@
package org.apache.helix.gateway.util;
-import org.apache.helix.gateway.service.GatewayServiceManager;
-import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.helix.gateway.constant.GatewayServiceEventType;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardState;
import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardStateMessage;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardTransitionStatus;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage;
public final class StateTransitionMessageTranslateUtil {
- public static TransitionMessage translateSTMsgToProto() {
+ public static TransitionMessage translateSTMsgToTransitionMessage() {
return null;
}
- public static GatewayServiceManager.GateWayServiceEvent translateProtoToSTMsg(ShardStateMessage message) {
- return null;
+ /**
+ * Translate from GRPC function call to Helix Gateway Service event.
+ */
+ public static GatewayServiceEvent translateShardStateMessageToEvent(ShardStateMessage request) {
+
+ if (request.hasShardState()) { // init connection to gateway service
+ ShardState shardState = request.getShardState();
+ String instanceName = shardState.getInstanceName();
+ String clusterName = shardState.getClusterName();
Review Comment:
TFTR. Updated
--
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 GatewayServiceManager [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu merged PR #2844:
URL: https://github.com/apache/helix/pull/2844
--
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 GatewayServiceManager [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1688449541
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceManager.java:
##########
@@ -26,40 +29,85 @@ public class GatewayServiceManager {
// TODO: add thread pool for init
// single thread tp for update
- public enum EventType {
- CONNECT, // init connection to gateway service
- UPDATE, // update state transition result
- DISCONNECT // shutdown connection to gateway service.
- }
+ // Map of cluster to HelixGatewayService
- public class GateWayServiceEvent {
- // event type
- EventType eventType;
- // event data
- String clusterName;
- String participantName;
+ // a single thread tp for event processing
+ ExecutorService _participantStateTransitionResultUpdator;
- // todo: add more fields
- }
+ // a thread pool for participant connecting
+ ExecutorService _participantConnectingTp;
+
+ // grpc service
+ HelixGatewayServiceGrpcService _grpcService;
public GatewayServiceManager() {
_helixGatewayServiceMap = new ConcurrentHashMap<>();
+ _participantStateTransitionResultUpdator = Executors.newSingleThreadExecutor();
+ _participantConnectingTp = Executors.newFixedThreadPool(10); // todo: make it configurable
+ _grpcService = new HelixGatewayServiceGrpcService(this);
}
public AtomicBoolean sendTransitionRequestToApplicationInstance() {
return null;
}
- public void updateShardState() {
-
+ public void newGatewayServiceEvent(GatewayServiceEvent event) {
+ if (event.getEventType().equals(GatewayServiceEventType.UPDATE)) {
+ _participantStateTransitionResultUpdator.submit(new shardStateUpdator(event));
+ } else {
+ _participantConnectingTp.submit(new participantConnectionProcessor(event));
+ }
}
- public void newParticipantConnecting() {
-
+ /**
+ * Update shard state
+ */
+ class shardStateUpdator implements Runnable {
+
+ GatewayServiceEvent _event;
+
+ public shardStateUpdator(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService = _helixGatewayServiceMap.get(_event._clusterName);
+ if (helixGatewayService == null) {
+ // TODO: return error code and throw exception.
+ return;
+ }
+ helixGatewayService.receiveSTResponse();
+ }
}
- public void participantDisconnected() {
-
+ /**
+ * Create HelixGatewayService instance and register it to the manager.
+ * It includes waiting for ZK connection, and also wait for previous Liveinstance to expire.
+ */
+ class participantConnectionProcessor implements Runnable {
+ GatewayServiceEvent _event;
+
+ public participantConnectionProcessor(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService;
+ if (_event.getEventType().equals(GatewayServiceEventType.CONNECT)) {
Review Comment:
Updated. Commits are messed up when I split 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] Implement GatewayServiceManager [helix]
Posted by "xyuanlu (via GitHub)" <gi...@apache.org>.
xyuanlu commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1688448617
##########
helix-gateway/src/main/java/org/apache/helix/gateway/constant/GatewayServiceEventType.java:
##########
@@ -0,0 +1,7 @@
+package org.apache.helix.gateway.constant;
+
+public enum GatewayServiceEventType {
Review Comment:
TFTR. I moved the enum to constant folder
--
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 GatewayServiceManager [helix]
Posted by "junkaixue (via GitHub)" <gi...@apache.org>.
junkaixue commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1690590639
##########
helix-gateway/src/main/java/org/apache/helix/gateway/util/StateTransitionMessageTranslateUtil.java:
##########
@@ -1,17 +1,71 @@
package org.apache.helix.gateway.util;
-import org.apache.helix.gateway.service.GatewayServiceManager;
-import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.helix.gateway.constant.GatewayServiceEventType;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardState;
import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardStateMessage;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardTransitionStatus;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage;
public final class StateTransitionMessageTranslateUtil {
- public static TransitionMessage translateSTMsgToProto() {
+ public static TransitionMessage translateSTMsgToTransitionMessage() {
return null;
}
- public static GatewayServiceManager.GateWayServiceEvent translateProtoToSTMsg(ShardStateMessage message) {
- return null;
+ /**
+ * Translate from GRPC function call to Helix Gateway Service event.
+ */
+ public static GatewayServiceEvent translateShardStateMessageToEvent(ShardStateMessage request) {
+
+ if (request.hasShardState()) { // init connection to gateway service
+ ShardState shardState = request.getShardState();
+ String instanceName = shardState.getInstanceName();
+ String clusterName = shardState.getClusterName();
+ Map<String, String> shardStateMap = new HashMap<>();
+ for (HelixGatewayServiceOuterClass.SingleResourceState resourceState : shardState.getResourceStateList()) {
+ for (HelixGatewayServiceOuterClass.SingleShardState state : resourceState.getShardStatesList()) {
+ shardStateMap.put(resourceState.getResource() + "_" + state.getShardName(), state.getCurrentState());
+ }
+ }
+ GatewayServiceEvent.GateWayServiceEventBuilder builder =
+ new GatewayServiceEvent.GateWayServiceEventBuilder(GatewayServiceEventType.CONNECT).setClusterName(
+ clusterName).setParticipantName(instanceName);
+ return builder.build();
+ } else {
Review Comment:
if there is a return, no else clause needed.
##########
helix-gateway/src/main/java/org/apache/helix/gateway/util/StateTransitionMessageTranslateUtil.java:
##########
@@ -1,17 +1,71 @@
package org.apache.helix.gateway.util;
-import org.apache.helix.gateway.service.GatewayServiceManager;
-import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.helix.gateway.constant.GatewayServiceEventType;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardState;
import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardStateMessage;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardTransitionStatus;
+import proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.TransitionMessage;
public final class StateTransitionMessageTranslateUtil {
- public static TransitionMessage translateSTMsgToProto() {
+ public static TransitionMessage translateSTMsgToTransitionMessage() {
return null;
}
- public static GatewayServiceManager.GateWayServiceEvent translateProtoToSTMsg(ShardStateMessage message) {
- return null;
+ /**
+ * Translate from GRPC function call to Helix Gateway Service event.
+ */
+ public static GatewayServiceEvent translateShardStateMessageToEvent(ShardStateMessage request) {
+
+ if (request.hasShardState()) { // init connection to gateway service
+ ShardState shardState = request.getShardState();
+ String instanceName = shardState.getInstanceName();
+ String clusterName = shardState.getClusterName();
Review Comment:
No need to define the variable when it only uses once.
--
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 GatewayServiceManager [helix]
Posted by "junkaixue (via GitHub)" <gi...@apache.org>.
junkaixue commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1688759371
##########
helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceManager.java:
##########
@@ -26,40 +29,85 @@ public class GatewayServiceManager {
// TODO: add thread pool for init
// single thread tp for update
- public enum EventType {
- CONNECT, // init connection to gateway service
- UPDATE, // update state transition result
- DISCONNECT // shutdown connection to gateway service.
- }
+ // Map of cluster to HelixGatewayService
- public class GateWayServiceEvent {
- // event type
- EventType eventType;
- // event data
- String clusterName;
- String participantName;
+ // a single thread tp for event processing
+ ExecutorService _participantStateTransitionResultUpdator;
- // todo: add more fields
- }
+ // a thread pool for participant connecting
+ ExecutorService _participantConnectingTp;
+
+ // grpc service
+ HelixGatewayServiceGrpcService _grpcService;
public GatewayServiceManager() {
_helixGatewayServiceMap = new ConcurrentHashMap<>();
+ _participantStateTransitionResultUpdator = Executors.newSingleThreadExecutor();
+ _participantConnectingTp = Executors.newFixedThreadPool(10); // todo: make it configurable
+ _grpcService = new HelixGatewayServiceGrpcService(this);
}
public AtomicBoolean sendTransitionRequestToApplicationInstance() {
return null;
}
- public void updateShardState() {
-
+ public void newGatewayServiceEvent(GatewayServiceEvent event) {
+ if (event.getEventType().equals(GatewayServiceEventType.UPDATE)) {
+ _participantStateTransitionResultUpdator.submit(new shardStateUpdator(event));
+ } else {
+ _participantConnectingTp.submit(new participantConnectionProcessor(event));
+ }
}
- public void newParticipantConnecting() {
-
+ /**
+ * Update shard state
+ */
+ class shardStateUpdator implements Runnable {
+
+ GatewayServiceEvent _event;
+
+ public shardStateUpdator(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService = _helixGatewayServiceMap.get(_event._clusterName);
+ if (helixGatewayService == null) {
+ // TODO: return error code and throw exception.
+ return;
+ }
+ helixGatewayService.receiveSTResponse();
+ }
}
- public void participantDisconnected() {
-
+ /**
+ * Create HelixGatewayService instance and register it to the manager.
+ * It includes waiting for ZK connection, and also wait for previous Liveinstance to expire.
+ */
+ class participantConnectionProcessor implements Runnable {
+ GatewayServiceEvent _event;
+
+ public participantConnectionProcessor(GatewayServiceEvent event) {
+ _event = event;
+ }
+
+ @Override
+ public void run() {
+ HelixGatewayService helixGatewayService;
+ if (_event.getEventType().equals(GatewayServiceEventType.CONNECT)) {
Review Comment:
No... UPDATE still falls back to DISCONNECT path.
--
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 GatewayServiceManager [helix]
Posted by "junkaixue (via GitHub)" <gi...@apache.org>.
junkaixue commented on code in PR #2844:
URL: https://github.com/apache/helix/pull/2844#discussion_r1692049083
##########
helix-gateway/src/main/java/org/apache/helix/gateway/grpcservice/HelixGatewayServiceGrpcService.java:
##########
@@ -0,0 +1,118 @@
+package org.apache.helix.gateway.grpcservice;
+
+import io.grpc.stub.StreamObserver;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.helix.gateway.service.GatewayServiceEvent;
+import org.apache.helix.gateway.service.GatewayServiceManager;
+import org.apache.helix.gateway.service.HelixGatewayServiceProcessor;
+import org.apache.helix.gateway.util.PerKeyLockRegistry;
Review Comment:
Is this implemented?
--
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