helix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From hu...@apache.org
Subject [helix] branch zooscalability updated: Make ZkHelixClusterVerifier and its child classes realm-aware (#867)
Date Thu, 12 Mar 2020 03:06:24 GMT
This is an automated email from the ASF dual-hosted git repository.

hulee pushed a commit to branch zooscalability
in repository https://gitbox.apache.org/repos/asf/helix.git


The following commit(s) were added to refs/heads/zooscalability by this push:
     new 7077098  Make ZkHelixClusterVerifier and its child classes realm-aware (#867)
7077098 is described below

commit 7077098fc9d6ab4a77958a846775942ea62cffc5
Author: Hunter Lee <hulee@linkedin.com>
AuthorDate: Wed Mar 11 20:06:18 2020 -0700

    Make ZkHelixClusterVerifier and its child classes realm-aware (#867)
    
    Changelist:
    Make sure constructors accept RealmAwareZkClient
    Add Builders in each child class of ZkHelixClusterVerifier so that ZkClient configs are
configurable and uses realm-aware ZkClient APIs
---
 .../BestPossibleExternalViewVerifier.java          |  81 +++++++++----
 .../ClusterVerifiers/ClusterLiveNodesVerifier.java |  46 +++++++-
 .../StrictMatchExternalViewVerifier.java           |  83 ++++++++-----
 .../ClusterVerifiers/ZkHelixClusterVerifier.java   | 130 ++++++++++++++++++---
 .../impl/factory/HelixZkClientFactory.java         |   2 +
 5 files changed, 274 insertions(+), 68 deletions(-)

diff --git a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java
b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java
index 0efd187..7f57fa9 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java
@@ -50,8 +50,6 @@ import org.apache.helix.controller.stages.CurrentStateComputationStage;
 import org.apache.helix.controller.stages.CurrentStateOutput;
 import org.apache.helix.controller.stages.ResourceComputationStage;
 import org.apache.helix.manager.zk.ZkBucketDataAccessor;
-import org.apache.helix.zookeeper.impl.client.ZkClient;
-import org.apache.helix.zookeeper.api.client.HelixZkClient;
 import org.apache.helix.model.ClusterConfig;
 import org.apache.helix.model.ExternalView;
 import org.apache.helix.model.IdealState;
@@ -60,6 +58,7 @@ import org.apache.helix.model.Resource;
 import org.apache.helix.model.ResourceAssignment;
 import org.apache.helix.model.StateModelDefinition;
 import org.apache.helix.task.TaskConstants;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -75,6 +74,15 @@ public class BestPossibleExternalViewVerifier extends ZkHelixClusterVerifier
{
   private final Set<String> _expectLiveInstances;
   private final ResourceControllerDataProvider _dataProvider;
 
+  /**
+   * Deprecated - please use the Builder to construct this class.
+   * @param zkAddr
+   * @param clusterName
+   * @param resources
+   * @param errStates
+   * @param expectLiveInstances
+   */
+  @Deprecated
   public BestPossibleExternalViewVerifier(String zkAddr, String clusterName, Set<String>
resources,
       Map<String, Map<String, String>> errStates, Set<String> expectLiveInstances)
{
     super(zkAddr, clusterName);
@@ -84,7 +92,16 @@ public class BestPossibleExternalViewVerifier extends ZkHelixClusterVerifier
{
     _dataProvider = new ResourceControllerDataProvider();
   }
 
-  public BestPossibleExternalViewVerifier(HelixZkClient zkClient, String clusterName,
+  /**
+   * Deprecated - please use the Builder to construct this class.
+   * @param zkClient
+   * @param clusterName
+   * @param resources
+   * @param errStates
+   * @param expectLiveInstances
+   */
+  @Deprecated
+  public BestPossibleExternalViewVerifier(RealmAwareZkClient zkClient, String clusterName,
       Set<String> resources, Map<String, Map<String, String>> errStates,
       Set<String> expectLiveInstances) {
     super(zkClient, clusterName);
@@ -94,20 +111,31 @@ public class BestPossibleExternalViewVerifier extends ZkHelixClusterVerifier
{
     _dataProvider = new ResourceControllerDataProvider();
   }
 
-  public static class Builder {
-    private String _clusterName;
+  private BestPossibleExternalViewVerifier(Builder builder) {
+    super(builder);
+    // Deep copy data from Builder
+    _errStates = new HashMap<>();
+    if (builder._errStates != null) {
+      builder._errStates.forEach((k, v) -> _errStates.put(k, new HashMap<>(v)));
+    }
+    _resources = new HashSet<>(builder._resources);
+    _expectLiveInstances = new HashSet<>(builder._expectLiveInstances);
+    _dataProvider = new ResourceControllerDataProvider();
+  }
+
+  public static class Builder extends ZkHelixClusterVerifier.Builder {
+    private final String _clusterName;
     private Map<String, Map<String, String>> _errStates;
     private Set<String> _resources;
     private Set<String> _expectLiveInstances;
-    private String _zkAddr;
-    private HelixZkClient _zkClient;
+    private RealmAwareZkClient _zkClient;
 
     public Builder(String clusterName) {
       _clusterName = clusterName;
     }
 
     public BestPossibleExternalViewVerifier build() {
-      if (_clusterName == null || (_zkAddr == null && _zkClient == null)) {
+      if (_clusterName == null || (_zkAddress == null && _zkClient == null)) {
         throw new IllegalArgumentException("Cluster name or zookeeper info is missing!");
       }
 
@@ -115,8 +143,15 @@ public class BestPossibleExternalViewVerifier extends ZkHelixClusterVerifier
{
         return new BestPossibleExternalViewVerifier(_zkClient, _clusterName, _resources,
_errStates,
             _expectLiveInstances);
       }
-      return new BestPossibleExternalViewVerifier(_zkAddr, _clusterName, _resources, _errStates,
-          _expectLiveInstances);
+
+      if (_realmAwareZkConnectionConfig == null || _realmAwareZkClientConfig == null) {
+        // For backward-compatibility
+        return new BestPossibleExternalViewVerifier(_zkAddress, _clusterName, _resources,
+            _errStates, _expectLiveInstances);
+      }
+
+      validate();
+      return new BestPossibleExternalViewVerifier(this);
     }
 
     public String getClusterName() {
@@ -151,25 +186,29 @@ public class BestPossibleExternalViewVerifier extends ZkHelixClusterVerifier
{
     }
 
     public String getZkAddr() {
-      return _zkAddr;
+      return _zkAddress;
     }
 
-    public Builder setZkAddr(String zkAddr) {
-      _zkAddr = zkAddr;
+    public Builder setZkClient(RealmAwareZkClient zkClient) {
+      _zkClient = zkClient;
       return this;
     }
 
-    public HelixZkClient getHelixZkClient() {
-      return _zkClient;
+    @Override
+    public Builder setZkAddr(String zkAddress) {
+      return (Builder) super.setZkAddr(zkAddress);
     }
 
-    @Deprecated
-    public ZkClient getZkClient() {
-      return (ZkClient) getHelixZkClient();
+    @Override
+    public Builder setRealmAwareZkConnectionConfig(
+        RealmAwareZkClient.RealmAwareZkConnectionConfig realmAwareZkConnectionConfig) {
+      return (Builder) super.setRealmAwareZkConnectionConfig(realmAwareZkConnectionConfig);
     }
-    public Builder setZkClient(HelixZkClient zkClient) {
-      _zkClient = zkClient;
-      return this;
+
+    @Override
+    public Builder setRealmAwareZkClientConfig(
+        RealmAwareZkClient.RealmAwareZkClientConfig realmAwareZkClientConfig) {
+      return (Builder) super.setRealmAwareZkClientConfig(realmAwareZkClientConfig);
     }
   }
 
diff --git a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ClusterLiveNodesVerifier.java
b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ClusterLiveNodesVerifier.java
index 0c284ff..8aeb64a 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ClusterLiveNodesVerifier.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ClusterLiveNodesVerifier.java
@@ -24,18 +24,25 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 
-import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+
 
 public class ClusterLiveNodesVerifier extends ZkHelixClusterVerifier {
 
   final Set<String> _expectLiveNodes;
 
-  public ClusterLiveNodesVerifier(HelixZkClient zkclient, String clusterName,
+  @Deprecated
+  public ClusterLiveNodesVerifier(RealmAwareZkClient zkclient, String clusterName,
       List<String> expectLiveNodes) {
     super(zkclient, clusterName);
     _expectLiveNodes = new HashSet<>(expectLiveNodes);
   }
 
+  private ClusterLiveNodesVerifier(Builder builder) {
+    super(builder);
+    _expectLiveNodes = new HashSet<>(builder._expectLiveNodes);
+  }
+
   @Override
   public boolean verifyByZkCallback(long timeout) {
     List<ClusterVerifyTrigger> triggers = new ArrayList<ClusterVerifyTrigger>();
@@ -51,4 +58,39 @@ public class ClusterLiveNodesVerifier extends ZkHelixClusterVerifier {
     return _expectLiveNodes.equals(actualLiveNodes);
   }
 
+  public static class Builder extends ZkHelixClusterVerifier.Builder {
+    private final String _clusterName; // This is the ZK path sharding key
+    private final Set<String> _expectLiveNodes;
+
+    public Builder(String clusterName, Set<String> expectLiveNodes) {
+      _clusterName = clusterName;
+      _expectLiveNodes = expectLiveNodes;
+    }
+
+    public ClusterLiveNodesVerifier build() {
+      if (_clusterName == null || _clusterName.isEmpty()) {
+        throw new IllegalArgumentException("Cluster name is missing!");
+      }
+
+      validate();
+      return new ClusterLiveNodesVerifier(this);
+    }
+
+    @Override
+    public Builder setZkAddr(String zkAddress) {
+      return (Builder) super.setZkAddr(zkAddress);
+    }
+
+    @Override
+    public Builder setRealmAwareZkConnectionConfig(
+        RealmAwareZkClient.RealmAwareZkConnectionConfig realmAwareZkConnectionConfig) {
+      return (Builder) super.setRealmAwareZkConnectionConfig(realmAwareZkConnectionConfig);
+    }
+
+    @Override
+    public Builder setRealmAwareZkClientConfig(
+        RealmAwareZkClient.RealmAwareZkClientConfig realmAwareZkClientConfig) {
+      return (Builder) super.setRealmAwareZkClientConfig(realmAwareZkClientConfig);
+    }
+  }
 }
diff --git a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/StrictMatchExternalViewVerifier.java
b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/StrictMatchExternalViewVerifier.java
index fcc7261..9c4a587 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/StrictMatchExternalViewVerifier.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/StrictMatchExternalViewVerifier.java
@@ -23,6 +23,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -33,8 +34,6 @@ import org.apache.helix.HelixException;
 import org.apache.helix.PropertyKey;
 import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
 import org.apache.helix.controller.rebalancer.AbstractRebalancer;
-import org.apache.helix.zookeeper.impl.client.ZkClient;
-import org.apache.helix.zookeeper.api.client.HelixZkClient;
 import org.apache.helix.model.ClusterConfig;
 import org.apache.helix.model.ExternalView;
 import org.apache.helix.model.IdealState;
@@ -42,6 +41,7 @@ import org.apache.helix.model.Partition;
 import org.apache.helix.model.StateModelDefinition;
 import org.apache.helix.task.TaskConstants;
 import org.apache.helix.util.HelixUtil;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -64,11 +64,12 @@ public class StrictMatchExternalViewVerifier extends ZkHelixClusterVerifier
{
   }
 
   @Deprecated
-  public StrictMatchExternalViewVerifier(HelixZkClient zkClient, String clusterName,
+  public StrictMatchExternalViewVerifier(RealmAwareZkClient zkClient, String clusterName,
       Set<String> resources, Set<String> expectLiveInstances) {
     this(zkClient, clusterName, resources, expectLiveInstances, false);
   }
 
+  @Deprecated
   private StrictMatchExternalViewVerifier(String zkAddr, String clusterName, Set<String>
resources,
       Set<String> expectLiveInstances, boolean isDeactivatedNodeAware) {
     super(zkAddr, clusterName);
@@ -77,7 +78,8 @@ public class StrictMatchExternalViewVerifier extends ZkHelixClusterVerifier
{
     _isDeactivatedNodeAware = isDeactivatedNodeAware;
   }
 
-  private StrictMatchExternalViewVerifier(HelixZkClient zkClient, String clusterName,
+  @Deprecated
+  private StrictMatchExternalViewVerifier(RealmAwareZkClient zkClient, String clusterName,
       Set<String> resources, Set<String> expectLiveInstances, boolean isDeactivatedNodeAware)
{
     super(zkClient, clusterName);
     _resources = resources;
@@ -85,17 +87,23 @@ public class StrictMatchExternalViewVerifier extends ZkHelixClusterVerifier
{
     _isDeactivatedNodeAware = isDeactivatedNodeAware;
   }
 
-  public static class Builder {
-    private String _clusterName;
+  private StrictMatchExternalViewVerifier(Builder builder) {
+    super(builder);
+    _resources = new HashSet<>(builder._resources);
+    _expectLiveInstances = new HashSet<>(builder._expectLiveInstances);
+    _isDeactivatedNodeAware = builder._isDeactivatedNodeAware;
+  }
+
+  public static class Builder extends ZkHelixClusterVerifier.Builder {
+    private final String _clusterName; // This is the ZK path sharding key
     private Set<String> _resources;
     private Set<String> _expectLiveInstances;
-    private String _zkAddr;
-    private HelixZkClient _zkClient;
+    private RealmAwareZkClient _zkClient;
     // For backward compatibility, set the default isDeactivatedNodeAware to be false.
     private boolean _isDeactivatedNodeAware = false;
 
     public StrictMatchExternalViewVerifier build() {
-      if (_clusterName == null || (_zkAddr == null && _zkClient == null)) {
+      if (_clusterName == null || (_zkAddress == null && _zkClient == null)) {
         throw new IllegalArgumentException("Cluster name or zookeeper info is missing!");
       }
 
@@ -103,8 +111,15 @@ public class StrictMatchExternalViewVerifier extends ZkHelixClusterVerifier
{
         return new StrictMatchExternalViewVerifier(_zkClient, _clusterName, _resources,
             _expectLiveInstances, _isDeactivatedNodeAware);
       }
-      return new StrictMatchExternalViewVerifier(_zkAddr, _clusterName, _resources,
-          _expectLiveInstances, _isDeactivatedNodeAware);
+
+      if (_realmAwareZkConnectionConfig == null || _realmAwareZkClientConfig == null) {
+        // For backward-compatibility
+        return new StrictMatchExternalViewVerifier(_zkAddress, _clusterName, _resources,
+            _expectLiveInstances, _isDeactivatedNodeAware);
+      }
+
+      validate();
+      return new StrictMatchExternalViewVerifier(this);
     }
 
     public Builder(String clusterName) {
@@ -133,25 +148,12 @@ public class StrictMatchExternalViewVerifier extends ZkHelixClusterVerifier
{
       return this;
     }
 
-    public String getZkAddr() {
-      return _zkAddr;
-    }
-
-    public Builder setZkAddr(String zkAddr) {
-      _zkAddr = zkAddr;
-      return this;
-    }
-
-    public HelixZkClient getHelixZkClient() {
-      return _zkClient;
+    public String getZkAddress() {
+      return _zkAddress;
     }
 
     @Deprecated
-    public ZkClient getZkClient() {
-      return (ZkClient) getHelixZkClient();
-    }
-
-    public Builder setZkClient(HelixZkClient zkClient) {
+    public Builder setZkClient(RealmAwareZkClient zkClient) {
       _zkClient = zkClient;
       return this;
     }
@@ -164,6 +166,33 @@ public class StrictMatchExternalViewVerifier extends ZkHelixClusterVerifier
{
       _isDeactivatedNodeAware = isDeactivatedNodeAware;
       return this;
     }
+
+    @Override
+    public Builder setZkAddr(String zkAddress) {
+      return (Builder) super.setZkAddr(zkAddress);
+    }
+
+    @Override
+    public Builder setRealmAwareZkConnectionConfig(
+        RealmAwareZkClient.RealmAwareZkConnectionConfig realmAwareZkConnectionConfig) {
+      return (Builder) super.setRealmAwareZkConnectionConfig(realmAwareZkConnectionConfig);
+    }
+
+    @Override
+    public Builder setRealmAwareZkClientConfig(
+        RealmAwareZkClient.RealmAwareZkClientConfig realmAwareZkClientConfig) {
+      return (Builder) super.setRealmAwareZkClientConfig(realmAwareZkClientConfig);
+    }
+
+    protected void validate() {
+      super.validate();
+      if (!_clusterName.equals(_realmAwareZkConnectionConfig.getZkRealmShardingKey())) {
+        throw new IllegalArgumentException(
+            "StrictMatchExternalViewVerifier: Cluster name: " + _clusterName
+                + " and ZK realm sharding key: " + _realmAwareZkConnectionConfig
+                .getZkRealmShardingKey() + " do not match!");
+      }
+    }
   }
 
   @Override
diff --git a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ZkHelixClusterVerifier.java
b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ZkHelixClusterVerifier.java
index e82fccf..449b659 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ZkHelixClusterVerifier.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ZkHelixClusterVerifier.java
@@ -19,6 +19,7 @@ package org.apache.helix.tools.ClusterVerifiers;
  * under the License.
  */
 
+import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -26,14 +27,17 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
 import org.apache.helix.PropertyKey;
+import org.apache.helix.SystemPropertyKeys;
 import org.apache.helix.api.listeners.PreFetch;
 import org.apache.helix.manager.zk.ZKHelixDataAccessor;
-import org.apache.helix.manager.zk.ZNRecordSerializer;
 import org.apache.helix.manager.zk.ZkBaseDataAccessor;
-import org.apache.helix.zookeeper.impl.client.ZkClient;
-import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
+import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
 import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
 import org.apache.helix.zookeeper.zkclient.IZkChildListener;
 import org.apache.helix.zookeeper.zkclient.IZkDataListener;
 import org.slf4j.Logger;
@@ -45,8 +49,8 @@ public abstract class ZkHelixClusterVerifier
   protected static int DEFAULT_TIMEOUT = 300 * 1000;
   protected static int DEFAULT_PERIOD = 500;
 
-  protected final HelixZkClient _zkClient;
-  // true if ZkHelixClusterVerifier was instantiated with a HelixZkClient, false otherwise
+  protected final RealmAwareZkClient _zkClient;
+  // true if ZkHelixClusterVerifier was instantiated with a RealmAwareZkClient, false otherwise
   // This is used for close() to determine how ZkHelixClusterVerifier should close the underlying
   // ZkClient
   private final boolean _usesExternalZkClient;
@@ -89,7 +93,13 @@ public abstract class ZkHelixClusterVerifier
     }
   }
 
-  public ZkHelixClusterVerifier(HelixZkClient zkClient, String clusterName) {
+  /**
+   * Deprecated because we discourage passing in a ZkClient directly.
+   * @param zkClient
+   * @param clusterName
+   */
+  @Deprecated
+  public ZkHelixClusterVerifier(RealmAwareZkClient zkClient, String clusterName) {
     if (zkClient == null || clusterName == null) {
       throw new IllegalArgumentException("requires zkClient|clusterName");
     }
@@ -100,12 +110,33 @@ public abstract class ZkHelixClusterVerifier
     _keyBuilder = _accessor.keyBuilder();
   }
 
+  @Deprecated
   public ZkHelixClusterVerifier(String zkAddr, String clusterName) {
-    if (zkAddr == null || clusterName == null) {
-      throw new IllegalArgumentException("requires zkAddr|clusterName");
+    if (clusterName == null || clusterName.isEmpty()) {
+      throw new IllegalArgumentException("ZkHelixClusterVerifier: clusterName is null or
empty!");
+    }
+    // If the multi ZK config is enabled, use DedicatedZkClient on multi-realm mode
+    if (Boolean.parseBoolean(System.getProperty(SystemPropertyKeys.MULTI_ZK_ENABLED))) {
+      try {
+        RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder connectionConfigBuilder =
+            new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder();
+        connectionConfigBuilder.setZkRealmShardingKey("/" + clusterName);
+        RealmAwareZkClient.RealmAwareZkClientConfig clientConfig =
+            new RealmAwareZkClient.RealmAwareZkClientConfig();
+        _zkClient = DedicatedZkClientFactory.getInstance()
+            .buildZkClient(connectionConfigBuilder.build(), clientConfig);
+      } catch (IOException | InvalidRoutingDataException | IllegalStateException e) {
+        // Note: IllegalStateException is for HttpRoutingDataReader if MSDS endpoint cannot
be
+        // found
+        throw new HelixException("ZkHelixClusterVerifier: failed to create ZkClient!", e);
+      }
+    } else {
+      if (zkAddr == null) {
+        throw new IllegalArgumentException("ZkHelixClusterVerifier: ZkAddress is null or
empty!");
+      }
+      _zkClient = DedicatedZkClientFactory.getInstance()
+          .buildZkClient(new HelixZkClient.ZkConnectionConfig(zkAddr));
     }
-    _zkClient = DedicatedZkClientFactory.getInstance()
-        .buildZkClient(new HelixZkClient.ZkConnectionConfig(zkAddr));
     _usesExternalZkClient = false;
     _zkClient.setZkSerializer(new ZNRecordSerializer());
     _clusterName = clusterName;
@@ -113,6 +144,27 @@ public abstract class ZkHelixClusterVerifier
     _keyBuilder = _accessor.keyBuilder();
   }
 
+  protected ZkHelixClusterVerifier(Builder builder) {
+    if (Boolean.parseBoolean(System.getProperty(SystemPropertyKeys.MULTI_ZK_ENABLED))) {
+      try {
+        // First, try to create a RealmAwareZkClient that's a DedicatedZkClient
+        _zkClient = DedicatedZkClientFactory.getInstance()
+            .buildZkClient(builder._realmAwareZkConnectionConfig,
+                builder._realmAwareZkClientConfig);
+      } catch (IOException | InvalidRoutingDataException | IllegalStateException e) {
+        throw new HelixException("ZkHelixClusterVerifier: failed to create ZkClient!", e);
+      }
+    } else {
+      _zkClient = DedicatedZkClientFactory.getInstance()
+          .buildZkClient(new HelixZkClient.ZkConnectionConfig(builder._zkAddress));
+    }
+    _usesExternalZkClient = false;
+    _zkClient.setZkSerializer(new ZNRecordSerializer());
+    _clusterName = builder._realmAwareZkConnectionConfig.getZkRealmShardingKey();
+    _accessor = new ZKHelixDataAccessor(_clusterName, new ZkBaseDataAccessor<>(_zkClient));
+    _keyBuilder = _accessor.keyBuilder();
+  }
+
   /**
    * Verify the cluster.
    * The method will be blocked at most {@code timeout}.
@@ -289,16 +341,58 @@ public abstract class ZkHelixClusterVerifier
     }
   }
 
-  public HelixZkClient getHelixZkClient() {
-    return _zkClient;
+  public String getClusterName() {
+    return _clusterName;
   }
 
-  @Deprecated
-  public ZkClient getZkClient() {
-    return (ZkClient) getHelixZkClient();
-  }
+  // TODO: refactor Builders for Java APIs
+  protected abstract static class Builder {
+    // Note: ZkHelixClusterVerifier is a single-realm API, so RealmMode is assumed to be
+    // SINGLE-REALM
+    protected String _zkAddress;
+    protected RealmAwareZkClient.RealmAwareZkConnectionConfig _realmAwareZkConnectionConfig;
+    protected RealmAwareZkClient.RealmAwareZkClientConfig _realmAwareZkClientConfig;
 
-  public String getClusterName() {
-    return _clusterName;
+    public Builder() {
+    }
+
+    public Builder setZkAddr(String zkAddress) {
+      _zkAddress = zkAddress;
+      return this;
+    }
+
+    public Builder setRealmAwareZkConnectionConfig(
+        RealmAwareZkClient.RealmAwareZkConnectionConfig realmAwareZkConnectionConfig) {
+      _realmAwareZkConnectionConfig = realmAwareZkConnectionConfig;
+      return this;
+    }
+
+    public Builder setRealmAwareZkClientConfig(
+        RealmAwareZkClient.RealmAwareZkClientConfig realmAwareZkClientConfig) {
+      _realmAwareZkClientConfig = realmAwareZkClientConfig;
+      return this;
+    }
+
+    protected void validate() {
+      // Validate that either ZkAddress or ZkRealmShardingKey is set.
+      if (_zkAddress == null || _zkAddress.isEmpty()) {
+        if (_realmAwareZkConnectionConfig == null
+            || _realmAwareZkConnectionConfig.getZkRealmShardingKey() == null
+            || _realmAwareZkConnectionConfig.getZkRealmShardingKey().isEmpty()) {
+          throw new IllegalArgumentException(
+              "ZkHelixClusterVerifier: one of either ZkAddress or ZkRealmShardingKey must
be set! ZkAddress: "
+                  + _zkAddress + " RealmAwareZkConnectionConfig: " + _realmAwareZkConnectionConfig);
+        }
+      }
+
+      // Resolve all default values
+      if (_realmAwareZkConnectionConfig == null) {
+        _realmAwareZkConnectionConfig =
+            new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder().build();
+      }
+      if (_realmAwareZkClientConfig == null) {
+        _realmAwareZkClientConfig = new RealmAwareZkClient.RealmAwareZkClientConfig();
+      }
+    }
   }
 }
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/HelixZkClientFactory.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/HelixZkClientFactory.java
index 9584c57..ca13321 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/HelixZkClientFactory.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/HelixZkClientFactory.java
@@ -38,6 +38,7 @@ abstract class HelixZkClientFactory implements RealmAwareZkClientFactory
{
    * @param clientConfig
    * @return HelixZkClient
    */
+  @Deprecated
   public abstract HelixZkClient buildZkClient(HelixZkClient.ZkConnectionConfig connectionConfig,
       HelixZkClient.ZkClientConfig clientConfig);
 
@@ -47,6 +48,7 @@ abstract class HelixZkClientFactory implements RealmAwareZkClientFactory
{
    * @param connectionConfig
    * @return HelixZkClient
    */
+  @Deprecated
   public HelixZkClient buildZkClient(HelixZkClient.ZkConnectionConfig connectionConfig) {
     return buildZkClient(connectionConfig, new HelixZkClient.ZkClientConfig());
   }


Mime
View raw message