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: Add SharedZkClient/InnerSharedZkClient implementation (#796)
Date Thu, 27 Feb 2020 22:55:49 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 5fc6305  Add SharedZkClient/InnerSharedZkClient implementation (#796)
5fc6305 is described below

commit 5fc6305d7b8df8f668d38ab6be3bf0ba59b8d2bd
Author: kaisun2000 <52840222+kaisun2000@users.noreply.github.com>
AuthorDate: Thu Feb 27 14:55:41 2020 -0800

    Add SharedZkClient/InnerSharedZkClient implementation (#796)
    
    Refactor the original SharedZkClient to InnerSharedZkClient. Add
    SharedZkClient implementation. The implementation use composition
    pattern. It would check the ZkPath validity and delegate the
    implementation to InnerSharedZkClient. In sum, InnerSharedZkClient
    is shared ZkClient but not realm aware. SharedZkClient is truely
    realm aware ZkClient.
---
 .../helix/manager/zk/client/SharedZkClient.java    |  16 +-
 .../helix/zookeeper/api/client/HelixZkClient.java  |  69 +--
 .../zookeeper/api/client/RealmAwareZkClient.java   |  14 +-
 .../api/factory/RealmAwareZkClientFactory.java     |   1 +
 .../zookeeper/impl/client/SharedZkClient.java      | 523 ++++++++++++++++++---
 .../impl/factory/SharedZkClientFactory.java        | 107 ++++-
 .../impl/client/RealmAwareZkClientTestBase.java    |   8 +-
 .../zookeeper/impl/client/TestSharedZkClient.java  |  47 ++
 8 files changed, 635 insertions(+), 150 deletions(-)

diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/client/SharedZkClient.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/client/SharedZkClient.java
index 52a2c17..be6b54d 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/client/SharedZkClient.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/client/SharedZkClient.java
@@ -19,19 +19,17 @@ package org.apache.helix.manager.zk.client;
  * under the License.
  */
 
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.helix.zookeeper.impl.factory.ZkConnectionManager;
+
+
 /**
  * Deprecated; use SharedZkClient in zookeeper-api instead.
  */
 @Deprecated
-class SharedZkClient extends org.apache.helix.zookeeper.impl.client.SharedZkClient {
-  /**
-   * Construct a shared RealmAwareZkClient that uses a shared ZkConnection.
-   *  @param connectionManager     The manager of the shared ZkConnection.
-   * @param clientConfig          ZkClientConfig details to create the shared RealmAwareZkClient.
-   * @param callback              Clean up logic when the shared RealmAwareZkClient is closed.
-   */
-  protected SharedZkClient(ZkConnectionManager connectionManager, ZkClientConfig clientConfig,
-      org.apache.helix.zookeeper.impl.client.SharedZkClient.OnCloseCallback callback) {
+class SharedZkClient extends SharedZkClientFactory.InnerSharedZkClient {
+  SharedZkClient(ZkConnectionManager connectionManager, ZkClientConfig clientConfig,
+      SharedZkClientFactory.OnCloseCallback callback) {
     super(connectionManager, clientConfig, callback);
   }
 }
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/HelixZkClient.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/HelixZkClient.java
index 9a1a69d..03bf000 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/HelixZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/HelixZkClient.java
@@ -19,9 +19,9 @@ package org.apache.helix.zookeeper.api.client;
  * under the License.
  */
 
+import org.apache.helix.zookeeper.zkclient.ZkClient;
 import org.apache.helix.zookeeper.zkclient.serialize.BasicZkSerializer;
 import org.apache.helix.zookeeper.zkclient.serialize.PathBasedZkSerializer;
-import org.apache.helix.zookeeper.zkclient.serialize.SerializableSerializer;
 import org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer;
 
 
@@ -42,7 +42,7 @@ public interface HelixZkClient extends RealmAwareZkClient {
   class ZkConnectionConfig {
     // Connection configs
     private final String _zkServers;
-    private int _sessionTimeout = HelixZkClient.DEFAULT_SESSION_TIMEOUT;
+    private int _sessionTimeout = DEFAULT_SESSION_TIMEOUT;
 
     public ZkConnectionConfig(String zkServers) {
       _zkServers = zkServers;
@@ -53,10 +53,10 @@ public interface HelixZkClient extends RealmAwareZkClient {
       if (obj == this) {
         return true;
       }
-      if (!(obj instanceof HelixZkClient.ZkConnectionConfig)) {
+      if (!(obj instanceof ZkConnectionConfig)) {
         return false;
       }
-      HelixZkClient.ZkConnectionConfig configObj = (HelixZkClient.ZkConnectionConfig) obj;
+      ZkConnectionConfig configObj = (ZkConnectionConfig) obj;
       return (_zkServers == null && configObj._zkServers == null || _zkServers !=
null && _zkServers
           .equals(configObj._zkServers)) && _sessionTimeout == configObj._sessionTimeout;
     }
@@ -71,7 +71,7 @@ public interface HelixZkClient extends RealmAwareZkClient {
       return (_zkServers + "_" + _sessionTimeout).replaceAll("[\\W]", "_");
     }
 
-    public HelixZkClient.ZkConnectionConfig setSessionTimeout(Integer sessionTimeout) {
+    public ZkConnectionConfig setSessionTimeout(Integer sessionTimeout) {
       this._sessionTimeout = sessionTimeout;
       return this;
     }
@@ -89,32 +89,16 @@ public interface HelixZkClient extends RealmAwareZkClient {
    * Deprecated - please use RealmAwareZkClient and RealmAwareZkClientConfig instead.
    *
    * Configuration for creating a new HelixZkClient with serializer and monitor.
-   *
-   * TODO: If possible, try to merge with RealmAwareZkClient's RealmAwareZkClientConfig to
reduce duplicate logic/code (without breaking backward-compatibility).
-   * Simply making this a subclass of RealmAwareZkClientConfig will break backward-compatiblity.
    */
   @Deprecated
-  class ZkClientConfig {
-    // For client to init the connection
-    private long _connectInitTimeout = DEFAULT_CONNECTION_TIMEOUT;
-
-    // Data access configs
-    private long _operationRetryTimeout = DEFAULT_OPERATION_TIMEOUT;
-
-    // Others
-    private PathBasedZkSerializer _zkSerializer;
-
-    // Monitoring
-    private String _monitorType;
-    private String _monitorKey;
-    private String _monitorInstanceName = null;
-    private boolean _monitorRootPathOnly = true;
-
+  class ZkClientConfig extends RealmAwareZkClientConfig {
+    @Override
     public ZkClientConfig setZkSerializer(PathBasedZkSerializer zkSerializer) {
       this._zkSerializer = zkSerializer;
       return this;
     }
 
+    @Override
     public ZkClientConfig setZkSerializer(ZkSerializer zkSerializer) {
       this._zkSerializer = new BasicZkSerializer(zkSerializer);
       return this;
@@ -125,6 +109,7 @@ public interface HelixZkClient extends RealmAwareZkClient {
      *
      * @param monitorType
      */
+    @Override
     public ZkClientConfig setMonitorType(String monitorType) {
       this._monitorType = monitorType;
       return this;
@@ -135,6 +120,7 @@ public interface HelixZkClient extends RealmAwareZkClient {
      *
      * @param monitorKey
      */
+    @Override
     public ZkClientConfig setMonitorKey(String monitorKey) {
       this._monitorKey = monitorKey;
       return this;
@@ -145,55 +131,28 @@ public interface HelixZkClient extends RealmAwareZkClient {
      *
      * @param instanceName
      */
+    @Override
     public ZkClientConfig setMonitorInstanceName(String instanceName) {
       this._monitorInstanceName = instanceName;
       return this;
     }
 
+    @Override
     public ZkClientConfig setMonitorRootPathOnly(Boolean monitorRootPathOnly) {
       this._monitorRootPathOnly = monitorRootPathOnly;
       return this;
     }
 
+    @Override
     public ZkClientConfig setOperationRetryTimeout(Long operationRetryTimeout) {
       this._operationRetryTimeout = operationRetryTimeout;
       return this;
     }
 
+    @Override
     public ZkClientConfig setConnectInitTimeout(long _connectInitTimeout) {
       this._connectInitTimeout = _connectInitTimeout;
       return this;
     }
-
-    public PathBasedZkSerializer getZkSerializer() {
-      if (_zkSerializer == null) {
-        _zkSerializer = new BasicZkSerializer(new SerializableSerializer());
-      }
-      return _zkSerializer;
-    }
-
-    public long getOperationRetryTimeout() {
-      return _operationRetryTimeout;
-    }
-
-    public String getMonitorType() {
-      return _monitorType;
-    }
-
-    public String getMonitorKey() {
-      return _monitorKey;
-    }
-
-    public String getMonitorInstanceName() {
-      return _monitorInstanceName;
-    }
-
-    public boolean isMonitorRootPathOnly() {
-      return _monitorRootPathOnly;
-    }
-
-    public long getConnectInitTimeout() {
-      return _connectInitTimeout;
-    }
   }
 }
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/RealmAwareZkClient.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/RealmAwareZkClient.java
index aa8bf7e..e466d36 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/RealmAwareZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/RealmAwareZkClient.java
@@ -380,19 +380,19 @@ public interface RealmAwareZkClient {
    */
   class RealmAwareZkClientConfig {
     // For client to init the connection
-    private long _connectInitTimeout = DEFAULT_CONNECTION_TIMEOUT;
+    protected long _connectInitTimeout = DEFAULT_CONNECTION_TIMEOUT;
 
     // Data access configs
-    private long _operationRetryTimeout = DEFAULT_OPERATION_TIMEOUT;
+    protected long _operationRetryTimeout = DEFAULT_OPERATION_TIMEOUT;
 
     // Others
-    private PathBasedZkSerializer _zkSerializer;
+    protected PathBasedZkSerializer _zkSerializer;
 
     // Monitoring
-    private String _monitorType;
-    private String _monitorKey;
-    private String _monitorInstanceName = null;
-    private boolean _monitorRootPathOnly = true;
+    protected String _monitorType;
+    protected String _monitorKey;
+    protected String _monitorInstanceName = null;
+    protected boolean _monitorRootPathOnly = true;
 
     public RealmAwareZkClientConfig setZkSerializer(PathBasedZkSerializer zkSerializer) {
       this._zkSerializer = zkSerializer;
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/factory/RealmAwareZkClientFactory.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/factory/RealmAwareZkClientFactory.java
index 8c1f7a3..cdfa778 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/factory/RealmAwareZkClientFactory.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/factory/RealmAwareZkClientFactory.java
@@ -45,6 +45,7 @@ public interface RealmAwareZkClientFactory {
    * @param metadataStoreRoutingData
    * @return RealmAwareZkClient
    */
+
   // TODO: remove MetadataStoreRoutingData
   default RealmAwareZkClient buildZkClient(
       RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig,
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/SharedZkClient.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/SharedZkClient.java
index 70d58a8..f2d9416 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/SharedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/SharedZkClient.java
@@ -20,13 +20,27 @@ package org.apache.helix.zookeeper.impl.client;
  */
 
 import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.concurrent.TimeUnit;
 
+import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
 import org.apache.helix.zookeeper.api.client.HelixZkClient;
-import org.apache.helix.zookeeper.impl.factory.ZkConnectionManager;
-import org.apache.helix.zookeeper.zkclient.IZkConnection;
-import org.apache.helix.zookeeper.zkclient.ZkConnection;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.helix.zookeeper.zkclient.DataUpdater;
+import org.apache.helix.zookeeper.zkclient.IZkChildListener;
+import org.apache.helix.zookeeper.zkclient.IZkDataListener;
+import org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks;
+import org.apache.helix.zookeeper.zkclient.deprecated.IZkStateListener;
+import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
+import org.apache.helix.zookeeper.zkclient.serialize.PathBasedZkSerializer;
+import org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer;
 import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.OpResult;
+import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -37,79 +51,460 @@ import org.slf4j.LoggerFactory;
  * HelixZkClient that uses shared ZkConnection.
  * A SharedZkClient won't manipulate the shared ZkConnection directly.
  */
-public class SharedZkClient extends ZkClient implements HelixZkClient {
+public class SharedZkClient implements RealmAwareZkClient {
   private static Logger LOG = LoggerFactory.getLogger(SharedZkClient.class);
-  /*
-   * Since we cannot really disconnect the ZkConnection, we need a dummy ZkConnection placeholder.
-   * This is to ensure connection field is never null even the shared RealmAwareZkClient
instance is closed so as to avoid NPE.
-   */
-  private final static ZkConnection IDLE_CONNECTION = new ZkConnection("Dummy_ZkServers");
-  private final OnCloseCallback _onCloseCallback;
-  private final ZkConnectionManager _connectionManager;
-
-  public interface OnCloseCallback {
-    /**
-     * Triggered after the RealmAwareZkClient is closed.
-     */
-    void onClose();
-  }
-
-  /**
-   * Construct a shared RealmAwareZkClient that uses a shared ZkConnection.
-   *
-   * @param connectionManager     The manager of the shared ZkConnection.
-   * @param clientConfig          ZkClientConfig details to create the shared RealmAwareZkClient.
-   * @param callback              Clean up logic when the shared RealmAwareZkClient is closed.
-   */
-  public SharedZkClient(ZkConnectionManager connectionManager, ZkClientConfig clientConfig,
-      OnCloseCallback callback) {
-    super(connectionManager.getConnection(), 0, clientConfig.getOperationRetryTimeout(),
-        clientConfig.getZkSerializer(), clientConfig.getMonitorType(), clientConfig.getMonitorKey(),
-        clientConfig.getMonitorInstanceName(), clientConfig.isMonitorRootPathOnly());
-    _connectionManager = connectionManager;
-    // Register to the base dedicated RealmAwareZkClient
-    _connectionManager.registerWatcher(this);
-    _onCloseCallback = callback;
+
+  private final HelixZkClient _innerSharedZkClient;
+  private final String _zkRealmShardingKey;
+  private final MetadataStoreRoutingData _metadataStoreRoutingData;
+  private final String _zkRealmAddress;
+
+  public SharedZkClient(RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig,
+      RealmAwareZkClient.RealmAwareZkClientConfig clientConfig,
+      MetadataStoreRoutingData metadataStoreRoutingData) {
+
+    if (connectionConfig == null) {
+      throw new IllegalArgumentException("RealmAwareZkConnectionConfig cannot be null!");
+    }
+    _zkRealmShardingKey = connectionConfig.getZkRealmShardingKey();
+
+    if (metadataStoreRoutingData == null) {
+      throw new IllegalArgumentException("MetadataStoreRoutingData cannot be null!");
+    }
+    _metadataStoreRoutingData = metadataStoreRoutingData;
+
+    // TODO: use _zkRealmShardingKey to generate zkRealmAddress. This can done the same way
of pull 765 once @hunter check it in.
+    // Get the ZkRealm address based on the ZK path sharding key
+    String zkRealmAddress = _metadataStoreRoutingData.getMetadataStoreRealm(_zkRealmShardingKey);
+    if (zkRealmAddress == null || zkRealmAddress.isEmpty()) {
+      throw new IllegalArgumentException(
+          "ZK realm address for the given ZK realm sharding key is invalid! ZK realm address:
"
+              + zkRealmAddress + " ZK realm sharding key: " + _zkRealmShardingKey);
+    }
+    _zkRealmAddress = zkRealmAddress;
+
+    // Create an InnerSharedZkClient to actually serve ZK requests
+    // TODO: Rename HelixZkClient in the future or remove it entirely - this will be a backward-compatibility
breaking change because HelixZkClient is being used by Helix users.
+
+    // Note, here delegate _innerSharedZkClient would share the same connectionManager. Once
the close() API of
+    // SharedZkClient is invoked, we can just call the close() API of delegate _innerSharedZkClient.
This would follow
+    // exactly the pattern of innerSharedZkClient closing logic, which would close the connectionManager
when the last
+    // sharedInnerZkClient is closed.
+    HelixZkClient.ZkConnectionConfig zkConnectionConfig =
+        new HelixZkClient.ZkConnectionConfig(zkRealmAddress)
+            .setSessionTimeout(connectionConfig.getSessionTimeout());
+    HelixZkClient.ZkClientConfig zkClientConfig = new HelixZkClient.ZkClientConfig();
+    zkClientConfig.setZkSerializer(clientConfig.getZkSerializer())
+        .setConnectInitTimeout(clientConfig.getConnectInitTimeout())
+        .setOperationRetryTimeout(clientConfig.getOperationRetryTimeout())
+        .setMonitorInstanceName(clientConfig.getMonitorInstanceName())
+        .setMonitorKey(clientConfig.getMonitorKey()).setMonitorType(clientConfig.getMonitorType())
+        .setMonitorRootPathOnly(clientConfig.isMonitorRootPathOnly());
+    _innerSharedZkClient =
+        SharedZkClientFactory.getInstance().buildZkClient(zkConnectionConfig, zkClientConfig);
+  }
+
+  @Override
+  public List<String> subscribeChildChanges(String path, IZkChildListener listener)
{
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.subscribeChildChanges(path, listener);
+  }
+
+  @Override
+  public void unsubscribeChildChanges(String path, IZkChildListener listener) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.unsubscribeChildChanges(path, listener);
+  }
+
+  @Override
+  public void subscribeDataChanges(String path, IZkDataListener listener) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.subscribeDataChanges(path, listener);
+  }
+
+  @Override
+  public void unsubscribeDataChanges(String path, IZkDataListener listener) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.unsubscribeDataChanges(path, listener);
+  }
+
+  @Override
+  public void subscribeStateChanges(IZkStateListener listener) {
+    _innerSharedZkClient.subscribeStateChanges(listener);
+  }
+
+  @Override
+  public void unsubscribeStateChanges(IZkStateListener listener) {
+    _innerSharedZkClient.unsubscribeStateChanges(listener);
+  }
+
+  @Override
+  public void unsubscribeAll() {
+    _innerSharedZkClient.unsubscribeAll();
+  }
+
+  @Override
+  public void createPersistent(String path) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.createPersistent(path);
+  }
+
+  @Override
+  public void createPersistent(String path, boolean createParents) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.createPersistent(path, createParents);
+  }
+
+  @Override
+  public void createPersistent(String path, boolean createParents, List<ACL> acl) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.createPersistent(path, createParents, acl);
+  }
+
+  @Override
+  public void createPersistent(String path, Object data) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.createPersistent(path, data);
+  }
+
+  @Override
+  public void createPersistent(String path, Object data, List<ACL> acl) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.createPersistent(path, data, acl);
+  }
+
+  @Override
+  public String createPersistentSequential(String path, Object data) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.createPersistentSequential(path, data);
+  }
+
+  @Override
+  public String createPersistentSequential(String path, Object data, List<ACL> acl)
{
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.createPersistentSequential(path, data, acl);
+  }
+
+  @Override
+  public void createEphemeral(String path) {
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public void createEphemeral(String path, String sessionId) {
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public void createEphemeral(String path, List<ACL> acl) {
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public void createEphemeral(String path, List<ACL> acl, String sessionId) {
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public String create(String path, Object data, CreateMode mode) {
+    checkIfPathContainsShardingKey(path);
+    // delegate to _innerSharedZkClient is fine as _innerSharedZkClient would not allow creating
ephemeral node.
+    // this still keeps the same behavior.
+    return _innerSharedZkClient.create(path, data, mode);
+  }
+
+  @Override
+  public String create(String path, Object datat, List<ACL> acl, CreateMode mode) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.create(path, datat, acl, mode);
+  }
+
+  @Override
+  public void createEphemeral(String path, Object data) {
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public void createEphemeral(String path, Object data, String sessionId) {
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public void createEphemeral(String path, Object data, List<ACL> acl) {
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public void createEphemeral(String path, Object data, List<ACL> acl, String sessionId)
{
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public String createEphemeralSequential(String path, Object data) {
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public String createEphemeralSequential(String path, Object data, List<ACL> acl)
{
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public String createEphemeralSequential(String path, Object data, String sessionId) {
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public String createEphemeralSequential(String path, Object data, List<ACL> acl,
+      String sessionId) {
+    throw new UnsupportedOperationException(
+        "Creating ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+            + " is not supported.");
+  }
+
+  @Override
+  public List<String> getChildren(String path) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.getChildren(path);
+  }
+
+  @Override
+  public int countChildren(String path) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.countChildren(path);
+  }
+
+  @Override
+  public boolean exists(String path) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.exists(path);
+  }
+
+  @Override
+  public Stat getStat(String path) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.getStat(path);
+  }
+
+  @Override
+  public boolean waitUntilExists(String path, TimeUnit timeUnit, long time) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.waitUntilExists(path, timeUnit, time);
+  }
+
+  @Override
+  public void deleteRecursively(String path) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.deleteRecursively(path);
+  }
+
+  @Override
+  public boolean delete(String path) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.delete(path);
+  }
+
+  @Override
+  public <T> T readData(String path) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.readData(path);
+  }
+
+  @Override
+  public <T> T readData(String path, boolean returnNullIfPathNotExists) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.readData(path, returnNullIfPathNotExists);
+  }
+
+  @Override
+  public <T> T readData(String path, Stat stat) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.readData(path, stat);
+  }
+
+  @Override
+  public <T> T readData(String path, Stat stat, boolean watch) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.readData(path, stat, watch);
+  }
+
+  @Override
+  public <T> T readDataAndStat(String path, Stat stat, boolean returnNullIfPathNotExists)
{
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.readDataAndStat(path, stat, returnNullIfPathNotExists);
+  }
+
+  @Override
+  public void writeData(String path, Object object) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.writeData(path, object);
+  }
+
+  @Override
+  public <T> void updateDataSerialized(String path, DataUpdater<T> updater) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.updateDataSerialized(path, updater);
+  }
+
+  @Override
+  public void writeData(String path, Object datat, int expectedVersion) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.writeDataReturnStat(path, datat, expectedVersion);
+  }
+
+  @Override
+  public Stat writeDataReturnStat(String path, Object datat, int expectedVersion) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.writeDataReturnStat(path, datat, expectedVersion);
+  }
+
+  @Override
+  public Stat writeDataGetStat(String path, Object datat, int expectedVersion) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.writeDataReturnStat(path, datat, expectedVersion);
+  }
+
+  @Override
+  public void asyncCreate(String path, Object datat, CreateMode mode,
+      ZkAsyncCallbacks.CreateCallbackHandler cb) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.asyncCreate(path, datat, mode, cb);
+  }
+
+  @Override
+  public void asyncSetData(String path, Object datat, int version,
+      ZkAsyncCallbacks.SetDataCallbackHandler cb) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.asyncSetData(path, datat, version, cb);
+  }
+
+  @Override
+  public void asyncGetData(String path, ZkAsyncCallbacks.GetDataCallbackHandler cb) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.asyncGetData(path, cb);
+  }
+
+  @Override
+  public void asyncExists(String path, ZkAsyncCallbacks.ExistsCallbackHandler cb) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.asyncExists(path, cb);
+  }
+
+  @Override
+  public void asyncDelete(String path, ZkAsyncCallbacks.DeleteCallbackHandler cb) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.asyncDelete(path, cb);
+  }
+
+  @Override
+  public void watchForData(String path) {
+    checkIfPathContainsShardingKey(path);
+    _innerSharedZkClient.watchForData(path);
+  }
+
+  @Override
+  public List<String> watchForChilds(String path) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.watchForChilds(path);
+  }
+
+  @Override
+  public long getCreationTime(String path) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.getCreationTime(path);
+  }
+
+  @Override
+  public List<OpResult> multi(Iterable<Op> ops) {
+    return _innerSharedZkClient.multi(ops);
+  }
+
+  @Override
+  public boolean waitUntilConnected(long time, TimeUnit timeUnit) {
+    return _innerSharedZkClient.waitUntilConnected(time, timeUnit);
+  }
+
+  @Override
+  public String getServers() {
+    return _innerSharedZkClient.getServers();
+  }
+
+  @Override
+  public long getSessionId() {
+    return _innerSharedZkClient.getSessionId();
   }
 
   @Override
   public void close() {
-    super.close();
-    if (isClosed()) {
-      // Note that if register is not done while constructing, these private fields may not
be init yet.
-      if (_connectionManager != null) {
-        _connectionManager.unregisterWatcher(this);
-      }
-      if (_onCloseCallback != null) {
-        _onCloseCallback.onClose();
-      }
-    }
+    _innerSharedZkClient.close();
   }
 
   @Override
-  public IZkConnection getConnection() {
-    if (isClosed()) {
-      return IDLE_CONNECTION;
-    }
-    return super.getConnection();
+  public boolean isClosed() {
+    return _innerSharedZkClient.isClosed();
   }
 
-  /**
-   * Since ZkConnection session is shared in this RealmAwareZkClient, do not create ephemeral
node using a SharedZKClient.
-   */
   @Override
-  public String create(final String path, Object datat, final List<ACL> acl,
-      final CreateMode mode) {
-    if (mode.isEphemeral()) {
-      throw new UnsupportedOperationException(
-          "Create ephemeral nodes using a " + SharedZkClient.class.getSimpleName()
-              + " is not supported.");
-    }
-    return super.create(path, datat, acl, mode);
+  public byte[] serialize(Object data, String path) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.serialize(data, path);
+  }
+
+  @Override
+  public <T> T deserialize(byte[] data, String path) {
+    checkIfPathContainsShardingKey(path);
+    return _innerSharedZkClient.deserialize(data, path);
+  }
+
+  @Override
+  public void setZkSerializer(ZkSerializer zkSerializer) {
+    _innerSharedZkClient.setZkSerializer(zkSerializer);
+  }
+
+  @Override
+  public void setZkSerializer(PathBasedZkSerializer zkSerializer) {
+    _innerSharedZkClient.setZkSerializer(zkSerializer);
   }
 
   @Override
-  protected boolean isManagingZkConnection() {
-    return false;
+  public PathBasedZkSerializer getZkSerializer() {
+    return _innerSharedZkClient.getZkSerializer();
+  }
+
+  private void checkIfPathContainsShardingKey(String path) {
+    // TODO: replace with the singleton MetadataStoreRoutingData
+    try {
+      String zkRealmForPath = _metadataStoreRoutingData.getMetadataStoreRealm(path);
+      if (!_zkRealmAddress.equals(zkRealmForPath)) {
+        throw new IllegalArgumentException("Given path: " + path + "'s ZK realm: " + zkRealmForPath
+            + " does not match the ZK realm: " + _zkRealmAddress + " and sharding key: "
+            + _zkRealmShardingKey + " for this DedicatedZkClient!");
+      }
+    } catch (NoSuchElementException e) {
+      throw new IllegalArgumentException(
+          "Given path: " + path + " does not have a valid sharding key!");
+    }
   }
-}
+}
\ No newline at end of file
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/SharedZkClientFactory.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/SharedZkClientFactory.java
index 1801614..80c58bf 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/SharedZkClientFactory.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/SharedZkClientFactory.java
@@ -20,12 +20,18 @@ package org.apache.helix.zookeeper.impl.factory;
  */
 
 import java.util.HashMap;
+import java.util.List;
 
 import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
 import org.apache.helix.zookeeper.api.client.HelixZkClient;
 import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
 import org.apache.helix.zookeeper.exception.ZkClientException;
 import org.apache.helix.zookeeper.impl.client.SharedZkClient;
+import org.apache.helix.zookeeper.impl.client.ZkClient;
+import org.apache.helix.zookeeper.zkclient.IZkConnection;
+import org.apache.helix.zookeeper.zkclient.ZkConnection;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.data.ACL;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -39,6 +45,12 @@ public class SharedZkClientFactory extends HelixZkClientFactory {
   private final HashMap<HelixZkClient.ZkConnectionConfig, ZkConnectionManager>
       _connectionManagerPool = new HashMap<>();
 
+  /*
+   * Since we cannot really disconnect the ZkConnection, we need a dummy ZkConnection placeholder.
+   * This is to ensure connection field is never null even the shared RealmAwareZkClient
instance is closed so as to avoid NPE.
+   */
+  private final static ZkConnection IDLE_CONNECTION = new ZkConnection("Dummy_ZkServers");
+
   protected SharedZkClientFactory() {
   }
 
@@ -47,9 +59,8 @@ public class SharedZkClientFactory extends HelixZkClientFactory {
       RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig,
       RealmAwareZkClient.RealmAwareZkClientConfig clientConfig,
       MetadataStoreRoutingData metadataStoreRoutingData) {
-    // TODO: Implement the logic
-    // Return an instance of SharedZkClient
-    return null;
+    // Note, the logic sharing connectionManager logic is inside SharedZkClient, similar
to innerSharedZkClient.
+    return new SharedZkClient(connectionConfig, clientConfig, metadataStoreRoutingData);
   }
 
   @Override
@@ -84,14 +95,14 @@ public class SharedZkClientFactory extends HelixZkClientFactory {
       if (zkConnectionManager == null) {
         throw new ZkClientException("Failed to create a connection manager in the pool to
share.");
       }
-      LOG.info("Sharing ZkConnection {} to a new SharedZkClient.", connectionConfig.toString());
-      return new SharedZkClient(zkConnectionManager, clientConfig,
-          new SharedZkClient.OnCloseCallback() {
-            @Override
-            public void onClose() {
-              cleanupConnectionManager(zkConnectionManager);
-            }
-          });
+      LOG.info("Sharing ZkConnection {} to a new InnerSharedZkClient.",
+          connectionConfig.toString());
+      return new InnerSharedZkClient(zkConnectionManager, clientConfig, new OnCloseCallback()
{
+        @Override
+        public void onClose() {
+          cleanupConnectionManager(zkConnectionManager);
+        }
+      });
     }
   }
 
@@ -128,4 +139,78 @@ public class SharedZkClientFactory extends HelixZkClientFactory {
     }
     return count;
   }
+
+  public interface OnCloseCallback {
+    /**
+     * Triggered after the SharedZkClient is closed.
+     */
+    void onClose();
+  }
+
+  /**
+   * NOTE: do NOT use this class directly. Please use SharedZkClientFactory to create an
instance of SharedZkClient.
+   * InnerSharedZkClient is a ZkClient used by SharedZkClient to power ZK operations against
a single ZK realm.
+   *
+   * NOTE2: current InnerSharedZkClient replace the original SharedZKClient. We intend to
keep the behavior of original
+   * SharedZkClient intact. (Think of rename the original SharedZkClient as InnerSharedZkClient.
This would maintain
+   * backward compatibility.
+   */
+  public static class InnerSharedZkClient extends ZkClient implements HelixZkClient {
+
+    private final OnCloseCallback _onCloseCallback;
+    private final ZkConnectionManager _connectionManager;
+
+    public InnerSharedZkClient(ZkConnectionManager connectionManager, ZkClientConfig clientConfig,
+        OnCloseCallback callback) {
+      super(connectionManager.getConnection(), 0, clientConfig.getOperationRetryTimeout(),
+          clientConfig.getZkSerializer(), clientConfig.getMonitorType(),
+          clientConfig.getMonitorKey(), clientConfig.getMonitorInstanceName(),
+          clientConfig.isMonitorRootPathOnly());
+      _connectionManager = connectionManager;
+      // Register to the base dedicated RealmAwareZkClient
+      _connectionManager.registerWatcher(this);
+      _onCloseCallback = callback;
+    }
+
+    @Override
+    public void close() {
+      super.close();
+      if (isClosed()) {
+        // Note that if register is not done while constructing, these private fields may
not be init yet.
+        if (_connectionManager != null) {
+          _connectionManager.unregisterWatcher(this);
+        }
+        if (_onCloseCallback != null) {
+          _onCloseCallback.onClose();
+        }
+      }
+    }
+
+    @Override
+    public IZkConnection getConnection() {
+      if (isClosed()) {
+        return IDLE_CONNECTION;
+      }
+      return super.getConnection();
+    }
+
+    /**
+     * Since ZkConnection session is shared in this HelixZkClient, do not create ephemeral
node using a SharedZKClient.
+     */
+    @Override
+    public String create(final String path, Object datat, final List<ACL> acl,
+        final CreateMode mode) {
+      if (mode.isEphemeral()) {
+        throw new UnsupportedOperationException(
+            "Create ephemeral nodes using " + SharedZkClient.class.getSimpleName()
+                + " is not supported.");
+      }
+      return super.create(path, datat, acl, mode);
+    }
+
+    @Override
+    protected boolean isManagingZkConnection() {
+      return false;
+    }
+  }
 }
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/RealmAwareZkClientTestBase.java
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/RealmAwareZkClientTestBase.java
index cd74975..d500ce4 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/RealmAwareZkClientTestBase.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/RealmAwareZkClientTestBase.java
@@ -38,16 +38,16 @@ import org.testng.annotations.Test;
 
 
 public abstract class RealmAwareZkClientTestBase extends ZkTestBase {
-  private static final String ZK_SHARDING_KEY_PREFIX = "/TEST_SHARDING_KEY";
-  private static final String TEST_VALID_PATH = ZK_SHARDING_KEY_PREFIX + "_" + 0 + "/a/b/c";
-  private static final String TEST_INVALID_PATH = ZK_SHARDING_KEY_PREFIX + "_invalid" + "/a/b/c";
+  protected static final String ZK_SHARDING_KEY_PREFIX = "/TEST_SHARDING_KEY";
+  protected static final String TEST_VALID_PATH = ZK_SHARDING_KEY_PREFIX + "_" + 0 + "/a/b/c";
+  protected static final String TEST_INVALID_PATH = ZK_SHARDING_KEY_PREFIX + "_invalid" +
"/a/b/c";
 
   // <Realm, List of sharding keys> Mapping
   private static final Map<String, List<String>> RAW_ROUTING_DATA = new HashMap<>();
 
   // The following RealmAwareZkClientFactory is to be defined in subclasses
   protected RealmAwareZkClientFactory _realmAwareZkClientFactory;
-  private RealmAwareZkClient _realmAwareZkClient;
+  protected RealmAwareZkClient _realmAwareZkClient;
   private MetadataStoreRoutingData _metadataStoreRoutingData;
 
   @BeforeClass
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestSharedZkClient.java
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestSharedZkClient.java
new file mode 100644
index 0000000..364f66f
--- /dev/null
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestSharedZkClient.java
@@ -0,0 +1,47 @@
+package org.apache.helix.zookeeper.impl.client;
+
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.zookeeper.CreateMode;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class TestSharedZkClient extends RealmAwareZkClientTestBase {
+  @BeforeClass
+  public void beforeClass() throws Exception {
+    super.beforeClass();
+    // Set the factory to SharedZkClientFactory
+    _realmAwareZkClientFactory = SharedZkClientFactory.getInstance();
+  }
+
+  @Test
+  public void testCreateEphemeralFailure() {
+    _realmAwareZkClient.setZkSerializer(new ZNRecordSerializer());
+
+    // Create a dummy ZNRecord
+    ZNRecord znRecord = new ZNRecord("DummyRecord");
+    znRecord.setSimpleField("Dummy", "Value");
+
+    // test createEphemeral should fail
+    try {
+      _realmAwareZkClient.createEphemeral(TEST_VALID_PATH);
+      Assert.fail(
+          "sharedReamlAwareZkClient is not expected to be able to create ephemeral node via
createEphemeral");
+    } catch (UnsupportedOperationException e) {
+      // this is expected
+    }
+
+    // test creating Ephemeral via creat would also fail
+    try {
+      _realmAwareZkClient.create(TEST_VALID_PATH, znRecord, CreateMode.EPHEMERAL);
+      Assert.fail(
+          "sharedRealmAwareZkClient is not expected to be able to create ephmeral node via
create");
+    } catch (UnsupportedOperationException e) {
+      // this is expected.
+    }
+  }
+}
\ No newline at end of file


Mime
View raw message