Add HelixGatewayParticipant and util tests.

helix-gateway/pom.xml

@@ -86,6 +86,12 @@
       <groupId>org.apache.helix</groupId>
       <artifactId>helix-core</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.helix</groupId>
+      <artifactId>helix-core</artifactId>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-services</artifactId>

helix-gateway/src/main/java/org/apache/helix/gateway/service/HelixGatewayServiceProcessor.java → helix-gateway/src/main/java/org/apache/helix/gateway/api/participant/HelixStateTransitionProcessor.java

@@ -1,4 +1,4 @@
-package org.apache.helix.gateway.service;
+package org.apache.helix.gateway.api.participant;
 
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
@@ -19,11 +19,26 @@
  * under the License.
  */
 
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateTransitionError;
+
 /**
- * Translate from/to GRPC function call to Helix Gateway Service event.
+ * Process state transition message.
  */
-public interface HelixGatewayServiceProcessor {
-
-  public boolean sendStateTransitionMessage( String instanceName);
+public interface HelixStateTransitionProcessor {
+  /**
+   * Process state transition message.
+   * @param message state transition message
+   * @throws Exception if failed to process the message
+   */
+  void processStateTransitionMessage(Message message) throws Exception;
 
+  /**
+   * Handle state transition error. This results from state transition handler throwing an exception or
+   * timing out.
+   *
+   * @param message state transition message
+   * @param error state transition error
+   */
+  void handleStateTransitionError(Message message, StateTransitionError error);
 }

helix-gateway/src/main/java/org/apache/helix/gateway/api/service/HelixGatewayServiceProcessor.java

@@ -1,15 +1,40 @@
 package org.apache.helix.gateway.api.service;
 
-import org.apache.helix.gateway.constant.MessageType;
+/*
+ * 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.model.Message;
 
 /**
  * Helix Gateway Service Processor interface allows sending state transition messages to
- * participants through services implementing this interface.
+ * participants through service implementing this interface.
  */
 public interface HelixGatewayServiceProcessor {
 
-  public void sendStateTransitionMessage(String instanceName, MessageType messageType,
+  /**
+   * Send a state transition message to a remote participant.
+   *
+   * @param instanceName the name of the participant
+   * @param currentState the current state of the participant
+   * @param message      the message to send
+   */
+  void sendStateTransitionMessage(String instanceName, String currentState,
       Message message);
 
 }

helix-gateway/src/main/java/org/apache/helix/gateway/grpcservice/HelixGatewayServiceGrpcService.java

@@ -24,7 +24,6 @@
 import java.util.Map;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.apache.commons.lang3.tuple.Pair;
-import org.apache.helix.gateway.constant.MessageType;
 import org.apache.helix.gateway.service.GatewayServiceEvent;
 import org.apache.helix.gateway.service.GatewayServiceManager;
 import org.apache.helix.gateway.api.service.HelixGatewayServiceProcessor;
@@ -61,8 +60,9 @@ public HelixGatewayServiceGrpcService(GatewayServiceManager manager) {
   /**
    * Grpc service end pint.
    * Application instances Report the state of the shard or result of transition request to the gateway service.
-   * @param responseObserver
-   * @return
+   *
+   * @param responseObserver the observer to send the response to the client
+   * @return the observer to receive the state of the shard or result of transition request
    */
   @Override
   public StreamObserver<proto.org.apache.helix.gateway.HelixGatewayServiceOuterClass.ShardStateMessage> report(
@@ -94,18 +94,19 @@ public void onCompleted() {
   /**
    * Send state transition message to the instance.
    * The instance must already have established a connection to the gateway service.
+   *
    * @param instanceName the instance name to send the message to
-   * @param messageType the type of the message
+   * @param currentState the current state of shard
    * @param message the message to convert to the transition message
    */
   @Override
-  public void sendStateTransitionMessage(String instanceName, MessageType messageType,
+  public void sendStateTransitionMessage(String instanceName, String currentState,
       Message message) {
     StreamObserver<TransitionMessage> observer;
     observer = _observerMap.get(instanceName);
     if (observer != null) {
       observer.onNext(
-          StateTransitionMessageTranslateUtil.translateSTMsgToTransitionMessage(messageType,
+          StateTransitionMessageTranslateUtil.translateSTMsgToTransitionMessage(currentState,
               message));
     }
   }

helix-gateway/src/main/java/org/apache/helix/gateway/participant/HelixGatewayParticipant.java

@@ -1,23 +1,49 @@
 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.HelixGatewayMultiTopStateStateTransitionProcessor;
+import org.apache.helix.gateway.api.participant.HelixStateTransitionProcessor;
 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 {
+/**
+ * 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;
@@ -32,7 +58,7 @@ private HelixGatewayParticipant(HelixGatewayServiceProcessor gatewayServiceProce
   }
 
   @Override
-  public void processMultiTopStateModelStateTransitionMessage(Message message) throws Exception {
+  public void processStateTransitionMessage(Message message) throws Exception {
     String transitionId = message.getMsgId();
     String resourceId = message.getResourceName();
     String shardId = message.getPartitionName();
@@ -45,9 +71,8 @@ public void processMultiTopStateModelStateTransitionMessage(Message message) thr
 
       CompletableFuture<Boolean> future = new CompletableFuture<>();
       _stateTransitionResultMap.put(transitionId, future);
-      _gatewayServiceProcessor.sendStateTransitionMessage(
-          _participantManager.getInstanceName(),
-          determineTransitionType(resourceId, shardId, toState), message);
+      _gatewayServiceProcessor.sendStateTransitionMessage(_participantManager.getInstanceName(),
+          getCurrentState(resourceId, shardId), message);
 
       boolean success = future.get();
       if (!success) {
@@ -79,6 +104,15 @@ public void handleStateTransitionError(Message message, StateTransitionError err
     // 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.
    *
@@ -94,33 +128,37 @@ public void completeStateTransition(String transitionId, boolean 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());
+    return getCurrentState(resourceId, shardId).equals(targetState);
   }
 
-  private MessageType determineTransitionType(String resourceId, String shardId,
-      String toState) {
-    boolean containsShard = _shardStateMap.containsKey(resourceId) && _shardStateMap.get(resourceId)
-        .containsKey(shardId);
-    if (toState.equals(HelixDefinedState.DROPPED.name())) {
-      return containsShard ? MessageType.DELETE : MessageType.ADD;
-    } else {
-      return containsShard ? MessageType.CHANGE_ROLE : MessageType.ADD;
-    }
+  @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())) {
-      _shardStateMap.computeIfPresent(resourceId, (k, v) -> {
+      getShardStateMap().computeIfPresent(resourceId, (k, v) -> {
         v.remove(shardId);
         if (v.isEmpty()) {
           return null;
         }
         return v;
       });
     } else {
-      _shardStateMap.computeIfAbsent(resourceId, k -> new ConcurrentHashMap<>())
+      getShardStateMap().computeIfAbsent(resourceId, k -> new ConcurrentHashMap<>())
           .put(shardId, state);
     }
   }
@@ -156,7 +194,7 @@ public Builder(HelixGatewayServiceProcessor helixGatewayServiceProcessor, String
      * @return the builder
      */
     public Builder addMultiTopStateStateModelDefinition(String stateModelDefinitionName) {
-      // TODO: Add validation that thi state model definition is a multi-top state model
+      // TODO: Add validation that the state model definition is a multi-top state model
       _multiTopStateModelDefinitions.add(stateModelDefinitionName);
       return this;
     }
@@ -169,7 +207,9 @@ public Builder addMultiTopStateStateModelDefinition(String stateModelDefinitionN
      * @param initialShardStateMap the initial shard state map to add
      * @return the Builder
      */
-    public Builder addInitialShardState(Map<String, Map<String, String>> initialShardStateMap) {
+    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));

helix-gateway/src/main/java/org/apache/helix/gateway/service/GatewayServiceManager.java

@@ -132,7 +132,7 @@ private void createHelixGatewayParticipant(String clusterName, String instanceNa
     // Create and add the participant to the participant map
     HelixGatewayParticipant.Builder participantBuilder =
         new HelixGatewayParticipant.Builder(_gatewayServiceProcessor, instanceName, clusterName,
-            _zkAddress).addInitialShardState(initialShardStateMap);
+            _zkAddress).setInitialShardState(initialShardStateMap);
     SUPPORTED_MULTI_STATE_MODEL_TYPES.forEach(
         participantBuilder::addMultiTopStateStateModelDefinition);
     _helixGatewayParticipantMap.computeIfAbsent(clusterName, k -> new ConcurrentHashMap<>())