Misc minor fixes in Helix tool, ReplicationEngine and Composite Manager

ambry-clustermap/src/main/java/com.github.ambry.clustermap/CompositeClusterManager.java

@@ -129,7 +129,7 @@ public byte getLocalDatacenterId() {
   @Override
   public String getDatacenterName(byte id) {
     String dcName = staticClusterManager.getDatacenterName(id);
-    if (!dcName.equals(helixClusterManager.getDatacenterName(id))) {
+    if (helixClusterManager != null && !dcName.equals(helixClusterManager.getDatacenterName(id))) {
       helixClusterManagerMetrics.getDatacenterNameMismatchCount.inc();
     }
     return dcName;

ambry-clustermap/src/main/java/com.github.ambry.clustermap/HelixBootstrapUpgradeUtil.java

@@ -421,7 +421,7 @@ private void populateInstancesAndPartitionsMap() {
   private void updateClusterMapInHelix(boolean startValidatingClusterManager) throws IOException {
     info("Initializing admins and possibly adding cluster in Helix (if non-existent)");
     maybeAddCluster();
-    info("Initialized, starting validating cluster manager");
+    info("Validating cluster manager is {}", startValidatingClusterManager ? "ENABLED" : "DISABLED");
     if (startValidatingClusterManager) {
       startClusterManager();
     }
@@ -1007,6 +1007,13 @@ private void logSummary() {
     } else {
       info("========No updates were done to the cluster in Helix========");
     }
+    if (validatingHelixClusterManager != null) {
+      info("========Validating HelixClusterManager metrics========");
+      info("Instance config change count: {}",
+          validatingHelixClusterManager.helixClusterManagerMetrics.instanceConfigChangeTriggerCount.getCount());
+      info("Instance config update ignored count: {}",
+          validatingHelixClusterManager.helixClusterManagerMetrics.ignoredUpdatesCount.getCount());
+    }
   }
 }
 

ambry-clustermap/src/main/java/com.github.ambry.clustermap/HelixClusterManager.java

@@ -38,11 +38,11 @@
 import org.I0Itec.zkclient.IZkDataListener;
 import org.apache.helix.AccessOption;
 import org.apache.helix.HelixManager;
-import org.apache.helix.InstanceConfigChangeListener;
 import org.apache.helix.InstanceType;
-import org.apache.helix.LiveInstanceChangeListener;
 import org.apache.helix.NotificationContext;
 import org.apache.helix.ZNRecord;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.LiveInstanceChangeListener;
 import org.apache.helix.model.InstanceConfig;
 import org.apache.helix.model.LiveInstance;
 import org.apache.helix.store.zk.ZkHelixPropertyStore;
@@ -519,7 +519,7 @@ private void updateStateOfReplicas(List<InstanceConfig> instanceConfigs) {
             AmbryDataNode node = instanceNameToAmbryDataNode.get(instanceName);
             if (instanceName.equals(selfInstanceName) || instanceXid <= currentXid.get()) {
               if (node == null) {
-                logger.info("Dynamic addition of new nodes is not yet supported, ignoring InstanceConfig {}",
+                logger.trace("Dynamic addition of new nodes is not yet supported, ignoring InstanceConfig {}",
                     instanceConfig);
               } else {
                 Set<String> sealedReplicas = new HashSet<>(getSealedReplicas(instanceConfig));
@@ -531,6 +531,7 @@ private void updateStateOfReplicas(List<InstanceConfig> instanceConfigs) {
                     logger.trace(
                         "Ignoring instanceConfig change for partition {} on instance {} because partition override is enabled",
                         partitionId, instanceName);
+                    helixClusterManagerMetrics.ignoredUpdatesCount.inc();
                   } else {
                     replica.setSealedState(sealedReplicas.contains(partitionId));
                     replica.setStoppedState(stoppedReplicas.contains(partitionId));

ambry-clustermap/src/test/java/com.github.ambry.clustermap/MockHelixManager.java

@@ -19,26 +19,26 @@
 import org.apache.helix.AccessOption;
 import org.apache.helix.ClusterMessagingService;
 import org.apache.helix.ConfigAccessor;
-import org.apache.helix.ControllerChangeListener;
-import org.apache.helix.CurrentStateChangeListener;
-import org.apache.helix.ExternalViewChangeListener;
 import org.apache.helix.HelixAdmin;
 import org.apache.helix.HelixDataAccessor;
 import org.apache.helix.HelixManager;
 import org.apache.helix.HelixManagerProperties;
-import org.apache.helix.IdealStateChangeListener;
-import org.apache.helix.InstanceConfigChangeListener;
 import org.apache.helix.InstanceType;
-import org.apache.helix.LiveInstanceChangeListener;
 import org.apache.helix.LiveInstanceInfoProvider;
-import org.apache.helix.MessageListener;
 import org.apache.helix.NotificationContext;
 import org.apache.helix.PreConnectCallback;
 import org.apache.helix.PropertyKey;
-import org.apache.helix.ScopedConfigChangeListener;
 import org.apache.helix.ZNRecord;
 import org.apache.helix.api.listeners.ClusterConfigChangeListener;
+import org.apache.helix.api.listeners.ControllerChangeListener;
+import org.apache.helix.api.listeners.CurrentStateChangeListener;
+import org.apache.helix.api.listeners.ExternalViewChangeListener;
+import org.apache.helix.api.listeners.IdealStateChangeListener;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.LiveInstanceChangeListener;
+import org.apache.helix.api.listeners.MessageListener;
 import org.apache.helix.api.listeners.ResourceConfigChangeListener;
+import org.apache.helix.api.listeners.ScopedConfigChangeListener;
 import org.apache.helix.healthcheck.ParticipantHealthReportCollector;
 import org.apache.helix.model.HelixConfigScope;
 import org.apache.helix.model.LiveInstance;
@@ -108,32 +108,21 @@ public void disconnect() {
   }
 
   @Override
-  public void addLiveInstanceChangeListener(LiveInstanceChangeListener listener) throws Exception {
-    if (beBadException != null) {
-      throw beBadException;
-    }
-    liveInstanceChangeListener = listener;
-    triggerLiveInstanceNotification(true);
+  public void addLiveInstanceChangeListener(org.apache.helix.LiveInstanceChangeListener listener) throws Exception {
+    // deprecated LiveInstanceChangeListener is no longer supported in testing
+    throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addInstanceConfigChangeListener(InstanceConfigChangeListener listener) throws Exception {
-    if (beBadException != null) {
-      throw beBadException;
-    }
-    instanceConfigChangeListener = listener;
-    triggerConfigChangeNotification(true);
+  public void addInstanceConfigChangeListener(org.apache.helix.InstanceConfigChangeListener listener) throws Exception {
+    // deprecated InstanceConfigChangeListener is no longer supported in testing
+    throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addExternalViewChangeListener(ExternalViewChangeListener listener) throws Exception {
-    if (beBadException != null) {
-      throw beBadException;
-    }
-    externalViewChangeListener = listener;
-    NotificationContext notificationContext = new NotificationContext(this);
-    notificationContext.setType(NotificationContext.Type.INIT);
-    externalViewChangeListener.onExternalViewChange(Collections.EMPTY_LIST, notificationContext);
+  public void addExternalViewChangeListener(org.apache.helix.ExternalViewChangeListener listener) throws Exception {
+    // deprecated ExternalViewChangeListener is no longer supported in testing
+    throw new IllegalStateException("Not implemented");
   }
 
   @Override
@@ -191,25 +180,27 @@ void triggerConfigChangeNotification(boolean init) {
   // Not implemented.
   //****************************
   @Override
-  public void addIdealStateChangeListener(IdealStateChangeListener listener) throws Exception {
+  public void addIdealStateChangeListener(org.apache.helix.IdealStateChangeListener listener) throws Exception {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addIdealStateChangeListener(
-      org.apache.helix.api.listeners.IdealStateChangeListener idealStateChangeListener) throws Exception {
+  public void addIdealStateChangeListener(IdealStateChangeListener idealStateChangeListener) throws Exception {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addLiveInstanceChangeListener(
-      org.apache.helix.api.listeners.LiveInstanceChangeListener liveInstanceChangeListener) throws Exception {
-    throw new IllegalStateException("Not implemented");
+  public void addLiveInstanceChangeListener(LiveInstanceChangeListener liveInstanceChangeListener) throws Exception {
+    if (beBadException != null) {
+      throw beBadException;
+    }
+    this.liveInstanceChangeListener = liveInstanceChangeListener;
+    triggerLiveInstanceNotification(true);
   }
 
   @Override
-  public void addConfigChangeListener(ScopedConfigChangeListener listener, HelixConfigScope.ConfigScopeProperty scope)
-      throws Exception {
+  public void addConfigChangeListener(org.apache.helix.ScopedConfigChangeListener listener,
+      HelixConfigScope.ConfigScopeProperty scope) throws Exception {
     throw new IllegalStateException("Not implemented");
   }
 
@@ -220,9 +211,13 @@ public void addConfigChangeListener(org.apache.helix.api.listeners.ConfigChangeL
   }
 
   @Override
-  public void addInstanceConfigChangeListener(
-      org.apache.helix.api.listeners.InstanceConfigChangeListener instanceConfigChangeListener) throws Exception {
-    throw new IllegalStateException("Not implemented");
+  public void addInstanceConfigChangeListener(InstanceConfigChangeListener instanceConfigChangeListener)
+      throws Exception {
+    if (beBadException != null) {
+      throw beBadException;
+    }
+    this.instanceConfigChangeListener = instanceConfigChangeListener;
+    triggerConfigChangeNotification(true);
   }
 
   @Override
@@ -237,65 +232,67 @@ public void addClusterfigChangeListener(ClusterConfigChangeListener clusterConfi
   }
 
   @Override
-  public void addConfigChangeListener(
-      org.apache.helix.api.listeners.ScopedConfigChangeListener scopedConfigChangeListener,
+  public void addConfigChangeListener(ScopedConfigChangeListener scopedConfigChangeListener,
       HelixConfigScope.ConfigScopeProperty configScopeProperty) throws Exception {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addMessageListener(org.apache.helix.api.listeners.MessageListener messageListener, String s)
-      throws Exception {
+  public void addMessageListener(MessageListener messageListener, String s) throws Exception {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addMessageListener(MessageListener listener, String instanceName) throws Exception {
+  public void addMessageListener(org.apache.helix.MessageListener listener, String instanceName) throws Exception {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addCurrentStateChangeListener(
-      org.apache.helix.api.listeners.CurrentStateChangeListener currentStateChangeListener, String s, String s1)
+  public void addCurrentStateChangeListener(CurrentStateChangeListener currentStateChangeListener, String s, String s1)
       throws Exception {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addCurrentStateChangeListener(CurrentStateChangeListener listener, String instanceName, String sessionId)
-      throws Exception {
+  public void addCurrentStateChangeListener(org.apache.helix.CurrentStateChangeListener listener, String instanceName,
+      String sessionId) throws Exception {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addExternalViewChangeListener(
-      org.apache.helix.api.listeners.ExternalViewChangeListener externalViewChangeListener) throws Exception {
-    throw new IllegalStateException("Not implemented");
+  public void addExternalViewChangeListener(ExternalViewChangeListener externalViewChangeListener) throws Exception {
+    if (beBadException != null) {
+      throw beBadException;
+    }
+    this.externalViewChangeListener = externalViewChangeListener;
+    NotificationContext notificationContext = new NotificationContext(this);
+    notificationContext.setType(NotificationContext.Type.INIT);
+    this.externalViewChangeListener.onExternalViewChange(Collections.emptyList(), notificationContext);
   }
 
   @Override
-  public void addTargetExternalViewChangeListener(
-      org.apache.helix.api.listeners.ExternalViewChangeListener externalViewChangeListener) throws Exception {
+  public void addTargetExternalViewChangeListener(ExternalViewChangeListener externalViewChangeListener)
+      throws Exception {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addControllerListener(ControllerChangeListener listener) {
+  public void addControllerListener(org.apache.helix.ControllerChangeListener listener) {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addControllerListener(org.apache.helix.api.listeners.ControllerChangeListener controllerChangeListener) {
+  public void addControllerListener(ControllerChangeListener controllerChangeListener) {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addControllerMessageListener(org.apache.helix.api.listeners.MessageListener messageListener) {
+  public void addControllerMessageListener(MessageListener messageListener) {
     throw new IllegalStateException("Not implemented");
   }
 
   @Override
-  public void addControllerMessageListener(MessageListener listener) {
+  public void addControllerMessageListener(org.apache.helix.MessageListener listener) {
     throw new IllegalStateException("Not implemented");
   }
 

ambry-replication/src/main/java/com.github.ambry.replication/ReplicationEngine.java

@@ -228,6 +228,7 @@ private RemoteReplicaInfo getRemoteReplicaInfo(PartitionId partitionId, String h
           .getHostname()
           .equals(hostName)) {
         foundRemoteReplicaInfo = remoteReplicaInfo;
+        break;
       }
     }
     // TODO: replace replicaPath.contains("vcr").

ambry-tools/src/main/java/com.github.ambry/clustermap/HelixBootstrapUpgradeTool.java

@@ -158,6 +158,9 @@ public static void main(String args[]) throws Exception {
     OptionSpecBuilder dryRun =
         parser.accepts("dryRun", "(Optional argument) Dry run, do not modify the cluster map in Helix.");
 
+    OptionSpecBuilder disableValidatingClusterManager = parser.accepts("disableVCM",
+        "(Optional argument) whether to disable validating cluster manager(VCM) in Helix bootstrap tool.");
+
     OptionSet options = parser.parse(args);
     String hardwareLayoutPath = options.valueOf(hardwareLayoutPathOpt);
     String partitionLayoutPath = options.valueOf(partitionLayoutPathOpt);
@@ -207,8 +210,11 @@ public static void main(String args[]) throws Exception {
           clusterNamePrefix, dcs,
           options.valueOf(maxPartitionsInOneResourceOpt) == null ? DEFAULT_MAX_PARTITIONS_PER_RESOURCE
               : Integer.valueOf(options.valueOf(maxPartitionsInOneResourceOpt)), options.has(dryRun),
-          options.has(forceRemove), new HelixAdminFactory(), true);
+          options.has(forceRemove), new HelixAdminFactory(), !options.has(disableValidatingClusterManager));
     }
+    System.out.println("======== HelixBootstrapUpgradeTool completed successfully! ========");
+    System.out.println("( If program doesn't exit, please use Ctrl-c to terminate. )");
+    System.exit(0);
   }
 }