helix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From hu...@apache.org
Subject [helix] branch master updated: Add an option for using dedicated ZkConnection in ZkBaseDataAccessor (#655)
Date Fri, 13 Dec 2019 00:24:35 GMT
This is an automated email from the ASF dual-hosted git repository.

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


The following commit(s) were added to refs/heads/master by this push:
     new 2416922  Add an option for using dedicated ZkConnection in ZkBaseDataAccessor (#655)
2416922 is described below

commit 2416922793e0138ef054186710714853db2ddb4e
Author: Hunter Lee <hulee@linkedin.com>
AuthorDate: Thu Dec 12 16:24:26 2019 -0800

    Add an option for using dedicated ZkConnection in ZkBaseDataAccessor (#655)
    
    There are usage patterns where a user needs to create ephemeral nodes using a ZkBaseDataAccessor.
We need to support this use case since we want users to avoid using ZkClient directly, so
we do it in ZkBaseDataAccessor, which is Helix's wrapper API around the raw ZkClient.
    
    The default behavior would be to use a shared ZK connection resource.
---
 .../helix/manager/zk/ZkBaseDataAccessor.java       | 107 +++++++++++++++++++--
 .../helix/manager/zk/ZkCacheBaseDataAccessor.java  |  29 ++++--
 .../helix/store/zk/ZkHelixPropertyStore.java       |   5 +
 3 files changed, 124 insertions(+), 17 deletions(-)

diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
index fb2749e..c1abccb 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
@@ -44,6 +44,7 @@ import org.apache.helix.manager.zk.ZkAsyncCallbacks.DeleteCallbackHandler;
 import org.apache.helix.manager.zk.ZkAsyncCallbacks.ExistsCallbackHandler;
 import org.apache.helix.manager.zk.ZkAsyncCallbacks.GetDataCallbackHandler;
 import org.apache.helix.manager.zk.ZkAsyncCallbacks.SetDataCallbackHandler;
+import org.apache.helix.manager.zk.client.DedicatedZkClientFactory;
 import org.apache.helix.manager.zk.client.HelixZkClient;
 import org.apache.helix.manager.zk.client.SharedZkClientFactory;
 import org.apache.helix.store.zk.ZNode;
@@ -56,6 +57,24 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class ZkBaseDataAccessor<T> implements BaseDataAccessor<T> {
+
+  // Designates which mode ZkBaseDataAccessor should be created in. If not specified, it
will be
+  // created on SHARED mode.
+  public enum ZkClientType {
+    /*
+     * When ZkBaseDataAccessor is created with the DEDICATED type, it supports ephemeral
node
+     * creation, callback functionality, and session management. But note that this is more
+     * resource-heavy since it creates a dedicated ZK connection so should be used sparingly
only
+     * when the aforementioned features are needed.
+     */
+    DEDICATED,
+    /*
+     * When ZkBaseDataAccessor is created with the SHARED type, it only supports CRUD
+     * functionalities. This will be the default mode of creation.
+     */
+    SHARED
+  }
+
   enum RetCode {
     OK,
     NODE_EXISTS,
@@ -103,28 +122,30 @@ public class ZkBaseDataAccessor<T> implements BaseDataAccessor<T>
{
 
   /**
    * The ZkBaseDataAccessor with custom serializer support of ZkSerializer type.
+   * Note: This constructor will use a shared ZkConnection.
+   * Do NOT use this for ephemeral node creation/callbacks/session management.
+   * Do use this for simple CRUD operations to ZooKeeper.
    * @param zkAddress The zookeeper address
    */
   public ZkBaseDataAccessor(String zkAddress, ZkSerializer zkSerializer) {
-    _zkClient = SharedZkClientFactory.getInstance().buildZkClient(
-        new HelixZkClient.ZkConnectionConfig(zkAddress),
-        new HelixZkClient.ZkClientConfig().setZkSerializer(zkSerializer));
-    _usesExternalZkClient = false;
+    this(zkAddress, zkSerializer, ZkClientType.SHARED);
   }
 
   /**
    * The ZkBaseDataAccessor with custom serializer support of PathBasedZkSerializer type.
+   * Note: This constructor will use a shared ZkConnection.
+   * Do NOT use this for ephemeral node creation/callbacks/session management.
+   * Do use this for simple CRUD operations to ZooKeeper.
    * @param zkAddress The zookeeper address
    */
-  public ZkBaseDataAccessor(String zkAddress, PathBasedZkSerializer zkSerializer) {
-    _zkClient = SharedZkClientFactory.getInstance().buildZkClient(
-        new HelixZkClient.ZkConnectionConfig(zkAddress),
-        new HelixZkClient.ZkClientConfig().setZkSerializer(zkSerializer));
-    _usesExternalZkClient = false;
+  public ZkBaseDataAccessor(String zkAddress, PathBasedZkSerializer pathBasedZkSerializer)
{
+    this(zkAddress, pathBasedZkSerializer, ZkClientType.SHARED);
   }
 
   /**
-   * The default ZkBaseDataAccessor with {@link org.apache.helix.ZNRecord} as the data model;
+   * Creates a ZkBaseDataAccessor with {@link org.apache.helix.ZNRecord} as the data model.
+   * Uses a shared ZkConnection resource.
+   * Does NOT support ephemeral node creation, callbacks, or session management.
    * Uses {@link ZNRecordSerializer} serializer
    * @param zkAddress The zookeeper address
    */
@@ -133,6 +154,72 @@ public class ZkBaseDataAccessor<T> implements BaseDataAccessor<T>
{
   }
 
   /**
+   * Creates a ZkBaseDataAccessor with {@link org.apache.helix.ZNRecord} as the data model.
+   * If DEDICATED, it will use a dedicated ZkConnection, which allows ephemeral
+   * node creation, callbacks, and session management.
+   * If SHARED, it will use a shared ZkConnection, which only allows simple
+   * CRUD operations to ZooKeeper.
+   * @param zkAddress
+   * @param zkClientType
+   */
+  public ZkBaseDataAccessor(String zkAddress, ZkClientType zkClientType) {
+    this(zkAddress, new ZNRecordSerializer(), zkClientType);
+  }
+
+  /**
+   * Creates a ZkBaseDataAccessor with a custom implementation of ZkSerializer.
+   * If DEDICATED, it will use a dedicated ZkConnection, which allows ephemeral
+   * node creation, callbacks, and session management.
+   * If SHARED, it will use a shared ZkConnection, which only allows simple
+   * CRUD operations to ZooKeeper.
+   * @param zkAddress
+   * @param zkSerializer
+   */
+  public ZkBaseDataAccessor(String zkAddress, ZkSerializer zkSerializer,
+      ZkClientType zkClientType) {
+    switch (zkClientType) {
+    case DEDICATED:
+      _zkClient = DedicatedZkClientFactory.getInstance().buildZkClient(
+          new HelixZkClient.ZkConnectionConfig(zkAddress),
+          new HelixZkClient.ZkClientConfig().setZkSerializer(zkSerializer));
+      break;
+    case SHARED:
+    default:
+      _zkClient = SharedZkClientFactory.getInstance().buildZkClient(
+          new HelixZkClient.ZkConnectionConfig(zkAddress),
+          new HelixZkClient.ZkClientConfig().setZkSerializer(zkSerializer));
+    }
+    _usesExternalZkClient = false;
+  }
+
+  /**
+   * Creates a ZkBaseDataAccessor with a custom implementation of PathBasedZkSerializer.
+   * If created with DEDICATED mode, it will use a dedicated ZkConnection, which allows ephemeral
+   * node creation, callbacks, and session management.
+   * If SHARED, it will use a shared ZkConnection, which only allows simple
+   * CRUD operations to ZooKeeper.
+   * @param zkAddress
+   * @param pathBasedZkSerializer
+   * @param zkClientType
+   */
+  public ZkBaseDataAccessor(String zkAddress, PathBasedZkSerializer pathBasedZkSerializer,
+      ZkClientType zkClientType) {
+    switch (zkClientType) {
+    case DEDICATED:
+      _zkClient = DedicatedZkClientFactory.getInstance().buildZkClient(
+          new HelixZkClient.ZkConnectionConfig(zkAddress),
+          new HelixZkClient.ZkClientConfig().setZkSerializer(pathBasedZkSerializer));
+      break;
+    case SHARED:
+    default:
+      _zkClient = SharedZkClientFactory.getInstance().buildZkClient(
+          new HelixZkClient.ZkConnectionConfig(zkAddress),
+          new HelixZkClient.ZkClientConfig().setZkSerializer(pathBasedZkSerializer));
+    }
+    _usesExternalZkClient = false;
+  }
+
+  /**
    * sync create
    */
   @Override
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
index ce4fc94..b230827 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
@@ -38,6 +38,7 @@ import org.apache.helix.AccessOption;
 import org.apache.helix.HelixException;
 import org.apache.helix.manager.zk.ZkAsyncCallbacks.CreateCallbackHandler;
 import org.apache.helix.manager.zk.ZkBaseDataAccessor.RetCode;
+import org.apache.helix.manager.zk.client.DedicatedZkClientFactory;
 import org.apache.helix.manager.zk.client.HelixZkClient;
 import org.apache.helix.manager.zk.client.SharedZkClientFactory;
 import org.apache.helix.store.HelixPropertyListener;
@@ -106,18 +107,32 @@ public class ZkCacheBaseDataAccessor<T> implements HelixPropertyStore<T>
{
 
   public ZkCacheBaseDataAccessor(String zkAddress, ZkSerializer serializer, String chrootPath,
       List<String> wtCachePaths, List<String> zkCachePaths) {
-    this(zkAddress, serializer, chrootPath, wtCachePaths, zkCachePaths, null, null);
+    this(zkAddress, serializer, chrootPath, wtCachePaths, zkCachePaths, null, null,
+        ZkBaseDataAccessor.ZkClientType.SHARED);
   }
 
   public ZkCacheBaseDataAccessor(String zkAddress, ZkSerializer serializer, String chrootPath,
       List<String> wtCachePaths, List<String> zkCachePaths, String monitorType,
String monitorkey) {
-    HelixZkClient.ZkClientConfig clientConfig = new HelixZkClient.ZkClientConfig();
-    clientConfig.setZkSerializer(serializer)
-        .setMonitorType(monitorType)
-        .setMonitorKey(monitorkey);
-    _zkClient = SharedZkClientFactory.getInstance()
-        .buildZkClient(new HelixZkClient.ZkConnectionConfig(zkAddress), clientConfig);
+    this(zkAddress, serializer, chrootPath, wtCachePaths, zkCachePaths, monitorType, monitorkey,
+        ZkBaseDataAccessor.ZkClientType.SHARED);
+  }
 
+  public ZkCacheBaseDataAccessor(String zkAddress, ZkSerializer serializer, String chrootPath,
+      List<String> wtCachePaths, List<String> zkCachePaths, String monitorType,
String monitorkey,
+      ZkBaseDataAccessor.ZkClientType zkClientType) {
+    HelixZkClient.ZkClientConfig clientConfig = new HelixZkClient.ZkClientConfig();
+    clientConfig.setZkSerializer(serializer).setMonitorType(monitorType).setMonitorKey(monitorkey);
+    switch (zkClientType) {
+    case DEDICATED:
+      _zkClient = DedicatedZkClientFactory.getInstance().buildZkClient(
+          new HelixZkClient.ZkConnectionConfig(zkAddress),
+          new HelixZkClient.ZkClientConfig().setZkSerializer(serializer));
+      break;
+    case SHARED:
+    default:
+      _zkClient = SharedZkClientFactory.getInstance()
+          .buildZkClient(new HelixZkClient.ZkConnectionConfig(zkAddress), clientConfig);
+    }
     _zkClient.waitUntilConnected(HelixZkClient.DEFAULT_CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS);
     _baseAccessor = new ZkBaseDataAccessor<>(_zkClient);
 
diff --git a/helix-core/src/main/java/org/apache/helix/store/zk/ZkHelixPropertyStore.java
b/helix-core/src/main/java/org/apache/helix/store/zk/ZkHelixPropertyStore.java
index 0e91314..a6e7a9c 100644
--- a/helix-core/src/main/java/org/apache/helix/store/zk/ZkHelixPropertyStore.java
+++ b/helix-core/src/main/java/org/apache/helix/store/zk/ZkHelixPropertyStore.java
@@ -41,4 +41,9 @@ public class ZkHelixPropertyStore<T> extends ZkCacheBaseDataAccessor<T>
{
   public ZkHelixPropertyStore(String zkAddress, ZkSerializer serializer, String chrootPath)
{
     super(zkAddress, serializer, chrootPath, null, null, MONITOR_TYPE, chrootPath);
   }
+
+  public ZkHelixPropertyStore(String zkAddress, ZkSerializer serializer, String chrootPath,
+      ZkBaseDataAccessor.ZkClientType zkClientType) {
+    super(zkAddress, serializer, chrootPath, null, null, MONITOR_TYPE, chrootPath, zkClientType);
+  }
 }


Mime
View raw message