[GitHub] [helix] dasahcc commented on a change in pull request #1178: Feature: ZooScalability Improvements

[GitBox <gi...@apache.org>]

dasahcc commented on a change in pull request #1178:
URL: https://github.com/apache/helix/pull/1178#discussion_r465340200



##########
File path: zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
##########
@@ -550,17 +546,71 @@ private ZkClient getZkClient(String path) {
   }
 
   private String getZkRealm(String path) {
+    if (_routingDataUpdateOnCacheMissEnabled) {
+      try {
+        return updateRoutingDataOnCacheMiss(path);
+      } catch (InvalidRoutingDataException e) {
+        LOG.error(
+            "FederatedZkClient::getZkRealm: Failed to update routing data due to invalid routing "
+                + "data!", e);
+        throw new MultiZkException(e);
+      }
+    }
+    return _metadataStoreRoutingData.getMetadataStoreRealm(path);
+  }
+
+  /**
+   * Perform a 2-tier routing data cache update:
+   * 1. Do an in-memory update from the singleton RoutingDataManager
+   * 2. Do an I/O based read from the routing data source by resetting RoutingDataManager
+   * @param path
+   * @return
+   * @throws InvalidRoutingDataException
+   */
+  private String updateRoutingDataOnCacheMiss(String path) throws InvalidRoutingDataException {
     String zkRealm;
     try {
       zkRealm = _metadataStoreRoutingData.getMetadataStoreRealm(path);
-    } catch (NoSuchElementException ex) {
-      throw new NoSuchElementException("Cannot find ZK realm for the path: " + path);
-    }
-
-    if (zkRealm == null || zkRealm.isEmpty()) {
-      throw new NoSuchElementException("Cannot find ZK realm for the path: " + path);
+    } catch (NoSuchElementException e1) {
+      synchronized (this) {
+        try {
+          zkRealm = _metadataStoreRoutingData.getMetadataStoreRealm(path);
+        } catch (NoSuchElementException e2) {
+          // Try 1) Refresh MetadataStoreRoutingData from RoutingDataManager
+          // This is an in-memory refresh from the Singleton RoutingDataManager - other
+          // FederatedZkClient objects may have triggered a cache refresh, so we first update the
+          // in-memory reference. This refresh only affects this object/thread, so we synchronize
+          // on "this".
+          _metadataStoreRoutingData =
+              RealmAwareZkClient.getMetadataStoreRoutingData(_connectionConfig);
+          try {
+            zkRealm = _metadataStoreRoutingData.getMetadataStoreRealm(path);
+          } catch (NoSuchElementException e3) {
+            synchronized (FederatedZkClient.class) {
+              try {
+                zkRealm = _metadataStoreRoutingData.getMetadataStoreRealm(path);
+              } catch (NoSuchElementException e4) {
+                if (shouldThrottleRead()) {
+                  // If routing data update from routing data source has taken place recently,
+                  // then just skip the update and throw the exception
+                  throw e4;
+                }
+                // Try 2) Reset RoutingDataManager and re-read the routing data from routing data
+                // source via I/O. Since RoutingDataManager's cache doesn't have it either, so we
+                // synchronize on all threads by locking on FederatedZkClient.class.
+                RoutingDataManager.getInstance().reset();

Review comment:
       Can we simplify the code here?
   Fundamentally, it is check path in _metadataStoreRoutingData -> get exception -> update _metadataStoreRoutingData from different level.
   
   There are several comment code. we can have a loop to wrap of it and based on the retried times to do different level update.

##########
File path: zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/RealmAwareZkClient.java
##########
@@ -486,6 +500,13 @@ private void validate() {
           throw new IllegalArgumentException(
               "RealmAwareZkConnectionConfig.Builder: ZK sharding key must be set on single-realm mode!");
         }
+        if ((_routingDataSourceEndpoint == null && _routingDataSourceType != null) || (
+            _routingDataSourceEndpoint != null && _routingDataSourceType == null)) {
+          // For routing data source type and endpoint, if one is set and not the other, it is invalid

Review comment:
       What if some other customer they dont set up the default MSDS? They are using our module for ZKClient use case. So will path ZK address still work?

##########
File path: zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
##########
@@ -0,0 +1,210 @@
+package org.apache.helix.zookeeper.routing;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
+import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
+import org.apache.helix.msdcommon.datamodel.TrieRoutingData;
+import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
+import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
+import org.apache.helix.zookeeper.exception.MultiZkException;
+
+
+/**
+ * RoutingDataManager is a Singleton that
+ * 1. resolves RoutingDataReader based on the system config given
+ * 2. caches routing data
+ * 3. provides public methods for reading routing data from various sources (configurable)
+ */
+public class RoutingDataManager {
+  /** HTTP call to MSDS is used to fetch routing data by default */
+  private String _defaultMsdsEndpoint =
+      System.getProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY);
+
+  /** Double-checked locking requires that the following fields be final (volatile) */
+  // The following map stands for (RoutingDataReaderType_endpoint ID, Raw Routing Data)
+  private final Map<String, Map<String, List<String>>> _rawRoutingDataMap =
+      new ConcurrentHashMap<>();
+  // The following map stands for (RoutingDataReaderType_endpoint ID, MetadataStoreRoutingData)
+  private final Map<String, MetadataStoreRoutingData> _metadataStoreRoutingDataMap =
+      new ConcurrentHashMap<>();
+
+  // Tracks the time at which reset() was called last. Used to throttle reset()
+  private volatile long _lastResetTimestamp;
+
+  // Singleton instance
+  private static RoutingDataManager _instance;
+
+  /**
+   * This class is a Singleton.
+   */
+  private RoutingDataManager() {
+    // Private constructor for Singleton
+  }
+
+  /**
+   * Lazy initialization with double-checked locking.
+   * @return
+   */
+  public static RoutingDataManager getInstance() {
+    if (_instance == null) {
+      synchronized (RoutingDataManager.class) {
+        if (_instance == null) {
+          _instance = new RoutingDataManager();
+        }
+      }
+    }
+    return _instance;
+  }
+
+  /**
+   * Fetches routing data from the data source via HTTP by querying the MSDS configured in the JVM
+   * config.
+   * @return
+   */
+  public Map<String, List<String>> getRawRoutingData() {
+    if (_defaultMsdsEndpoint == null || _defaultMsdsEndpoint.isEmpty()) {
+      throw new IllegalStateException(
+          "HttpRoutingDataReader was unable to find a valid MSDS endpoint String in System "
+              + "Properties!");
+    }
+    return getRawRoutingData(RoutingDataReaderType.HTTP, _defaultMsdsEndpoint);
+  }
+
+  /**
+   * Fetches routing data from the data source via HTTP.
+   * @return a mapping from "metadata store realm addresses" to lists of
+   * "metadata store sharding keys", where the sharding keys in a value list all route to
+   * the realm address in the key disallows a meaningful mapping to be returned.
+   * @param routingDataReaderType
+   * @param endpoint
+   */
+  public Map<String, List<String>> getRawRoutingData(RoutingDataReaderType routingDataReaderType,
+      String endpoint) {
+    String routingDataCacheKey = getRoutingDataCacheKey(routingDataReaderType, endpoint);
+    Map<String, List<String>> rawRoutingData = _rawRoutingDataMap.get(routingDataCacheKey);
+    if (rawRoutingData == null) {
+      synchronized (RoutingDataManager.class) {
+        rawRoutingData = _rawRoutingDataMap.get(routingDataCacheKey);
+        if (rawRoutingData == null) {
+          RoutingDataReader reader = resolveRoutingDataReader(routingDataReaderType);
+          rawRoutingData = reader.getRawRoutingData(endpoint);
+          _rawRoutingDataMap.put(routingDataCacheKey, rawRoutingData);
+        }
+      }
+    }
+    return rawRoutingData;
+  }
+
+  /**
+   * Returns the routing data read from MSDS in a MetadataStoreRoutingData format by querying the
+   * MSDS configured in the JVM config.
+   * @return MetadataStoreRoutingData
+   */
+  public MetadataStoreRoutingData getMetadataStoreRoutingData() throws InvalidRoutingDataException {
+    if (_defaultMsdsEndpoint == null || _defaultMsdsEndpoint.isEmpty()) {
+      throw new IllegalStateException(
+          "HttpRoutingDataReader was unable to find a valid MSDS endpoint String in System "
+              + "Properties!");
+    }
+    return getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, _defaultMsdsEndpoint);
+  }
+
+  /**
+   * Returns the routing data read from MSDS as a MetadataStoreRoutingData object.
+   * @param routingDataReaderType
+   * @param endpoint
+   * @return
+   * @throws IOException
+   * @throws InvalidRoutingDataException
+   */
+  public MetadataStoreRoutingData getMetadataStoreRoutingData(
+      RoutingDataReaderType routingDataReaderType, String endpoint)
+      throws InvalidRoutingDataException {
+    String routingDataCacheKey = getRoutingDataCacheKey(routingDataReaderType, endpoint);
+    MetadataStoreRoutingData metadataStoreRoutingData =
+        _metadataStoreRoutingDataMap.get(routingDataCacheKey);
+    if (metadataStoreRoutingData == null) {
+      synchronized (RoutingDataManager.class) {
+        metadataStoreRoutingData = _metadataStoreRoutingDataMap.get(routingDataCacheKey);
+        if (metadataStoreRoutingData == null) {
+          metadataStoreRoutingData =
+              new TrieRoutingData(getRawRoutingData(routingDataReaderType, endpoint));
+          _metadataStoreRoutingDataMap.put(routingDataCacheKey, metadataStoreRoutingData);
+        }
+      }
+    }
+    return metadataStoreRoutingData;
+  }
+
+  /**
+   * Clears the statically-cached routing data and private fields.
+   */
+  public synchronized void reset() {
+    _rawRoutingDataMap.clear();
+    _metadataStoreRoutingDataMap.clear();
+    _defaultMsdsEndpoint =
+        System.getProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY);
+    _lastResetTimestamp = System.currentTimeMillis();
+  }
+
+  /**
+   * Returns the timestamp for the last reset().
+   * @return
+   */
+  public long getLastResetTimestamp() {
+    return _lastResetTimestamp;
+  }
+
+  /**
+   * Returns an appropriate instance of RoutingDataReader given the type.
+   * @param routingDataReaderType
+   * @return
+   */
+  private RoutingDataReader resolveRoutingDataReader(RoutingDataReaderType routingDataReaderType) {
+    // Instantiate an instance of routing data reader using the type
+    try {
+      return (RoutingDataReader) Class.forName(routingDataReaderType.getClassName()).newInstance();
+    } catch (ClassNotFoundException | IllegalAccessException | InstantiationException e) {
+      throw new MultiZkException(

Review comment:
       error log here?




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org