[HELIX-599] Support creating/maintaining/routing resources with same names in different instance groups.

helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java

@@ -70,7 +70,6 @@ public void process(ClusterEvent event) throws Exception {
 
     for (String resourceName : resourceMap.keySet()) {
       Resource resource = resourceMap.get(resourceName);
-      int bucketSize = resource.getBucketSize();
 
       StateModelDefinition stateModelDef = cache.getStateModelDef(resource.getStateModelDefRef());
 
@@ -125,9 +124,8 @@ public void process(ClusterEvent event) throws Exception {
           } else {
 
             Message message =
-                createMessage(manager, resourceName, partition.getPartitionName(), instanceName,
-                    currentState, nextState, sessionIdMap.get(instanceName), stateModelDef.getId(),
-                    resource.getStateModelFactoryname(), bucketSize);
+                createMessage(manager, resource, partition.getPartitionName(), instanceName,
+                    currentState, nextState, sessionIdMap.get(instanceName), stateModelDef.getId());
 
             IdealState idealState = cache.getIdealState(resourceName);
             if (idealState != null
@@ -188,23 +186,30 @@ public void process(ClusterEvent event) throws Exception {
     event.addAttribute(AttributeName.MESSAGES_ALL.toString(), output);
   }
 
-  private Message createMessage(HelixManager manager, String resourceName, String partitionName,
+  private Message createMessage(HelixManager manager, Resource resource, String partitionName,
       String instanceName, String currentState, String nextState, String sessionId,
-      String stateModelDefName, String stateModelFactoryName, int bucketSize) {
+      String stateModelDefName) {
     String uuid = UUID.randomUUID().toString();
     Message message = new Message(MessageType.STATE_TRANSITION, uuid);
     message.setSrcName(manager.getInstanceName());
     message.setTgtName(instanceName);
     message.setMsgState(MessageState.NEW);
     message.setPartitionName(partitionName);
-    message.setResourceName(resourceName);
+    message.setResourceName(resource.getResourceName());
     message.setFromState(currentState);
     message.setToState(nextState);
     message.setTgtSessionId(sessionId);
     message.setSrcSessionId(manager.getSessionId());
     message.setStateModelDef(stateModelDefName);
-    message.setStateModelFactoryName(stateModelFactoryName);
-    message.setBucketSize(bucketSize);
+    message.setStateModelFactoryName(resource.getStateModelFactoryname());
+    message.setBucketSize(resource.getBucketSize());
+
+    if (resource.getResourceGroupName() != null) {
+      message.setResourceGroupName(resource.getResourceGroupName());
+    }
+    if (resource.getResourceTag() != null) {
+      message.setResourceTag(resource.getResourceTag());
+    }
 
     return message;
   }

helix-core/src/main/java/org/apache/helix/controller/stages/ResourceComputationStage.java

@@ -62,6 +62,8 @@ public void process(ClusterEvent event) throws Exception {
           resource.setStateModelFactoryName(idealState.getStateModelFactoryName());
           resource.setBucketSize(idealState.getBucketSize());
           resource.setBatchMessageMode(idealState.getBatchMessageMode());
+          resource.setResourceGroupName(idealState.getResourceGroupName());
+          resource.setResourceTag(idealState.getInstanceGroupTag());
         }
 
         for (String partition : partitionSet) {
@@ -102,6 +104,12 @@ public void process(ClusterEvent event) throws Exception {
             resource.setStateModelFactoryName(currentState.getStateModelFactoryName());
             resource.setBucketSize(currentState.getBucketSize());
             resource.setBatchMessageMode(currentState.getBatchMessageMode());
+
+            IdealState idealState = idealStates.get(resourceName);
+            if (idealState != null) {
+              resource.setResourceGroupName(idealState.getResourceGroupName());
+              resource.setResourceTag(idealState.getInstanceGroupTag());
+            }
           }
 
           if (currentState.getStateModelDefRef() == null) {

helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java

@@ -404,6 +404,13 @@ public void resetPartition(String clusterName, String instanceName, String resou
       message.setToState(stateModel.getInitialState());
       message.setStateModelFactoryName(idealState.getStateModelFactoryName());
 
+      if (idealState.getResourceGroupName() != null) {
+        message.setResourceGroupName(idealState.getResourceGroupName());
+      }
+      if (idealState.getInstanceGroupTag() != null) {
+        message.setResourceTag(idealState.getInstanceGroupTag());
+      }
+
       resetMessages.add(message);
       messageKeys.add(keyBuilder.message(instanceName, message.getId()));
     }

helix-core/src/main/java/org/apache/helix/model/ExternalView.java

@@ -31,6 +31,16 @@
  * of current states for the partitions in a resource
  */
 public class ExternalView extends HelixProperty {
+
+  /**
+   * Properties that are persisted and are queryable for an external view
+   */
+  public enum ExternalViewProperty {
+    INSTANCE_GROUP_TAG,
+    RESOURCE_GROUP_NAME,
+    GROUP_ROUTING_ENABLED
+  }
+
   /**
    * Instantiate an external view with the resource it corresponds to
    * @param resource the name of the resource
@@ -95,6 +105,32 @@ public String getResourceName() {
     return _record.getId();
   }
 
+  /**
+   * Get the resource group name
+   *
+   * @return the name of the resource group this resource belongs to.
+   */
+  public String getResourceGroupName() {
+    return _record.getSimpleField(ExternalViewProperty.RESOURCE_GROUP_NAME.toString());
+  }
+
+  /**
+   * Check whether the group routing is enabled for this resource.
+   *
+   * @return true if the group routing enabled for this resource; false otherwise
+   */
+  public boolean isGroupRoutingEnabled() {
+    return _record.getBooleanField(ExternalViewProperty.GROUP_ROUTING_ENABLED.name(), false);
+  }
+
+  /**
+   * Check for a group tag of this resource
+   * @return the group tag, or null if none is present
+   */
+  public String getInstanceGroupTag() {
+    return _record.getSimpleField(ExternalViewProperty.INSTANCE_GROUP_TAG.toString());
+  }
+
   @Override
   public boolean isValid() {
     return true;

helix-core/src/main/java/org/apache/helix/model/IdealState.java

@@ -55,7 +55,9 @@ public enum IdealStateProperty {
     MAX_PARTITIONS_PER_INSTANCE,
     INSTANCE_GROUP_TAG,
     REBALANCER_CLASS_NAME,
-    HELIX_ENABLED
+    HELIX_ENABLED,
+    RESOURCE_GROUP_NAME,
+    GROUP_ROUTING_ENABLED
   }
 
   public static final String QUERY_LIST = "PREFERENCE_LIST_QUERYS";
@@ -113,7 +115,7 @@ public String getResourceName() {
   }
 
   /**
-   * Get the rebalance mode of the ideal state
+   * Set the rebalance mode of the ideal state
    * @param mode {@link IdealStateModeProperty}
    */
   @Deprecated
@@ -124,7 +126,7 @@ public void setIdealStateMode(String mode) {
   }
 
   /**
-   * Get the rebalance mode of the resource
+   * Set the rebalance mode of the resource
    * @param rebalancerType
    */
   public void setRebalanceMode(RebalanceMode rebalancerType) {
@@ -159,6 +161,43 @@ public String getRebalancerClassName() {
     return _record.getSimpleField(IdealStateProperty.REBALANCER_CLASS_NAME.toString());
   }
 
+  /**
+   * Set the resource group name
+   * @param resourceGroupName
+   */
+  public void setResourceGroupName(String resourceGroupName) {
+    _record.setSimpleField(IdealStateProperty.RESOURCE_GROUP_NAME.toString(), resourceGroupName);
+  }
+
+  /**
+   * Get the resource group name
+   *
+   * @return
+   */
+  public String getResourceGroupName() {
+    return _record.getSimpleField(IdealStateProperty.RESOURCE_GROUP_NAME.toString());
+  }
+
+  /**
+   * Get if the resource group routing feature is enabled or not
+   * By default, it's disabled
+   *
+   * @return true if enabled; false otherwise
+   */
+  public boolean isResourceGroupEnabled() {
+    return _record.getBooleanField(IdealStateProperty.GROUP_ROUTING_ENABLED.name(), false);
+  }
+
+  /**
+   * Enable/Disable the aggregated routing on resource group.
+   *
+   * @param enabled
+   */
+  public void enableGroupRouting(boolean enabled) {
+    _record.setSimpleField(IdealStateProperty.GROUP_ROUTING_ENABLED.name(),
+        Boolean.toString(enabled));
+  }
+
   /**
    * Set the maximum number of partitions of this resource that an instance can serve
    * @param max the maximum number of partitions supported

helix-core/src/main/java/org/apache/helix/model/Message.java

@@ -64,6 +64,8 @@ public enum Attributes {
     MSG_STATE,
     PARTITION_NAME,
     RESOURCE_NAME,
+    RESOURCE_GROUP_NAME,
+    RESOURCE_TAG,
     FROM_STATE,
     TO_STATE,
     STATE_MODEL_DEF,
@@ -396,6 +398,42 @@ public String getResourceName() {
     return _record.getSimpleField(Attributes.RESOURCE_NAME.toString());
   }
 
+  /**
+   * Set the resource group associated with this message
+   *
+   * @param resourceGroupName resource group name to set
+   */
+  public void setResourceGroupName(String resourceGroupName) {
+    _record.setSimpleField(Attributes.RESOURCE_GROUP_NAME.toString(), resourceGroupName);
+  }
+
+  /**
+   * Get the resource group name associated with this message
+   *
+   * @return resource group name
+   */
+  public String getResourceGroupName() {
+    return _record.getSimpleField(Attributes.RESOURCE_GROUP_NAME.toString());
+  }
+
+  /**
+   * Set the resource tag associated with this message
+   *
+   * @param resourceTag resource tag to set
+   */
+  public void setResourceTag(String resourceTag) {
+    _record.setSimpleField(Attributes.RESOURCE_TAG.toString(), resourceTag);
+  }
+
+  /**
+   * Get the resource tag associated with this message
+   *
+   * @return resource tag
+   */
+  public String getResourceTag() {
+    return _record.getSimpleField(Attributes.RESOURCE_TAG.toString());
+  }
+
   /**
    * Get the resource partition associated with this message
    * @return partition name

helix-core/src/main/java/org/apache/helix/model/Resource.java

@@ -38,6 +38,8 @@ public class Resource {
   private String _stateModelFactoryName;
   private int _bucketSize = 0;
   private boolean _batchMessageMode = false;
+  private String _resourceGroupName;
+  private String _resourceTag;
 
   /**
    * Instantiate a resource by its name
@@ -149,6 +151,39 @@ public boolean getBatchMessageMode() {
     return _batchMessageMode;
   }
 
+  /**
+   * Get the resource tag assigned to this resource
+   *
+   * @return the name of the tag
+   */
+  public String getResourceTag() {
+    return _resourceTag;
+  }
+
+  /**
+   * Set the resource tag
+   * @param resourceTag
+   */
+  public void setResourceTag(String resourceTag) {
+    _resourceTag = resourceTag;
+  }
+
+  /**
+   * Get resource group name
+   * @return the resource group name
+   */
+  public String getResourceGroupName() {
+    return _resourceGroupName;
+  }
+
+  /**
+   * Set resource group name
+   * @param resourceGroupName
+   */
+  public void setResourceGroupName(String resourceGroupName) {
+    _resourceGroupName = resourceGroupName;
+  }
+
   @Override
   public String toString() {
     StringBuilder sb = new StringBuilder();