You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by hu...@apache.org on 2020/08/14 18:13:55 UTC
[helix] branch master updated (918039b -> 28dc1cf)
This is an automated email from the ASF dual-hosted git repository.
hulee pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git.
from 918039b Add latency metric support for customized view aggregation (#1187)
new 5e85638 Implement RoutingDataManager to replace HttpRoutingDataReader
new a686d58 Fix string
new 0a95840 Remove unnecessary IOException
new 3bfc766 Change interface for RoutingDataReader
new 0ff7e95 Add HttpZkFallbackRoutingDataReader
new d3477be Implement ZkRoutingDataReader
new 6014df6 Modify realm-aware ZkClient and Helix API for configurable routing source
new f001cde Make RoutingDataManager a pure Singleton with double-checked locking
new c876649 Refactor RealmAwareZkClient code to remove duplicate code
new 59cd674 Implement routing data update upon cache miss for FederatedZkClient
new 8c2059c Implement throttling for routing data update on cache miss
new 28dc1cf Change UPDATE_ROUTING_DATA_ON_CACHE_MISS and remove unused imports
The 12 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../main/java/org/apache/helix/ConfigAccessor.java | 7 +-
.../manager/zk/GenericBaseDataAccessorBuilder.java | 3 +-
.../helix/manager/zk/GenericZkHelixApiBuilder.java | 19 +-
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 27 +-
.../java/org/apache/helix/manager/zk/ZKUtil.java | 3 +-
.../helix/manager/zk/ZkBaseDataAccessor.java | 7 +-
.../helix/manager/zk/ZkBucketDataAccessor.java | 2 +-
.../java/org/apache/helix/tools/ClusterSetup.java | 8 +-
.../ClusterVerifiers/ZkHelixClusterVerifier.java | 5 +-
.../multizk/TestMultiZkHelixJavaApis.java | 92 +++++-
.../apache/helix/rest/server/ServerContext.java | 11 +-
.../zookeeper/api/client/RealmAwareZkClient.java | 67 ++++-
.../zookeeper/constant/RoutingDataConstants.java | 23 +-
.../zookeeper/constant/RoutingDataReaderType.java | 54 ++++
.../constant/RoutingSystemPropertyKeys.java | 24 +-
...kClientException.java => MultiZkException.java} | 13 +-
.../zookeeper/impl/client/DedicatedZkClient.java | 16 +-
.../zookeeper/impl/client/FederatedZkClient.java | 124 ++++++--
.../zookeeper/impl/client/SharedZkClient.java | 15 +-
.../impl/factory/DedicatedZkClientFactory.java | 4 +-
.../impl/factory/SharedZkClientFactory.java | 3 +-
.../zookeeper/routing/HttpRoutingDataReader.java | 115 ++++++++
.../routing/HttpZkFallbackRoutingDataReader.java | 81 ++++++
.../zookeeper/routing/RoutingDataManager.java | 217 ++++++++++++++
.../helix/zookeeper/routing/RoutingDataReader.java | 53 ++++
.../zookeeper/routing/ZkRoutingDataReader.java | 83 ++++++
.../zookeeper/util/HttpRoutingDataReader.java | 197 -------------
.../helix/zookeeper/constant/TestConstants.java | 16 +-
.../impl/client/RealmAwareZkClientTestBase.java | 8 +-
.../impl/client/TestFederatedZkClient.java | 319 ++++++++++++++++++++-
.../TestHttpZkFallbackRoutingDataReader.java | 73 +++++
.../zookeeper/routing/TestZkRoutingDataReader.java | 77 +++++
...DataReader.java => TestRoutingDataManager.java} | 33 ++-
33 files changed, 1405 insertions(+), 394 deletions(-)
copy helix-core/src/main/java/org/apache/helix/api/cloud/CloudInstanceInformation.java => zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataConstants.java (63%)
create mode 100644 zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java
copy helix-core/src/main/java/org/apache/helix/api/listeners/CustomizedViewChangeListener.java => zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java (58%)
copy zookeeper-api/src/main/java/org/apache/helix/zookeeper/exception/{ZkClientException.java => MultiZkException.java} (74%)
create mode 100644 zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpRoutingDataReader.java
create mode 100644 zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpZkFallbackRoutingDataReader.java
create mode 100644 zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
create mode 100644 zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataReader.java
create mode 100644 zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/ZkRoutingDataReader.java
delete mode 100644 zookeeper-api/src/main/java/org/apache/helix/zookeeper/util/HttpRoutingDataReader.java
create mode 100644 zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestHttpZkFallbackRoutingDataReader.java
create mode 100644 zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestZkRoutingDataReader.java
rename zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/{TestHttpRoutingDataReader.java => TestRoutingDataManager.java} (77%)
[helix] 11/12: Implement throttling for routing data update on
cache miss
Posted by hu...@apache.org.
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
commit 8c2059c78130f8138d6295bef27d76ac5a413aaf
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Mon Jul 27 18:36:39 2020 -0700
Implement throttling for routing data update on cache miss
This commit implements throttling for routing data update by using a timestamp for last time the cache was reset in RoutingDataManager. It defines a default interval (5 seconds) but makes this interval configurable by way of System Properties config.
---
...PropertyKeys.java => RoutingDataConstants.java} | 14 ++---
.../constant/RoutingSystemPropertyKeys.java | 5 ++
.../zookeeper/impl/client/FederatedZkClient.java | 38 ++++++++++++
.../zookeeper/routing/RoutingDataManager.java | 12 ++++
.../impl/client/TestFederatedZkClient.java | 70 +++++++++++++++++++++-
5 files changed, 128 insertions(+), 11 deletions(-)
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataConstants.java
similarity index 66%
copy from zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
copy to zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataConstants.java
index a57075b..164c543 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataConstants.java
@@ -19,15 +19,13 @@ package org.apache.helix.zookeeper.constant;
* under the License.
*/
-/**
- * This class contains various routing-related system property keys for multi-zk clients.
- */
-public class RoutingSystemPropertyKeys {
+public class RoutingDataConstants {
/**
- * If enabled, FederatedZkClient (multiZkClient) will invalidate the cached routing data and
- * re-read the routing data from the routing data source upon ZK path sharding key cache miss.
+ * Default interval that defines how frequently RoutingDataManager's routing data should be
+ * updated from the routing data source. This exists to apply throttling to the rate at which
+ * the ZkClient pulls routing data from the routing data source to avoid overloading the routing
+ * data source.
*/
- public static final String UPDATE_ROUTING_DATA_ON_CACHE_MISS =
- "update.routing.data.on.cache.miss.enabled";
+ public static final long DEFAULT_ROUTING_DATA_UPDATE_INTERVAL_MS = 5 * 1000L; // 5 seconds
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
index a57075b..e22ad08 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
@@ -30,4 +30,9 @@ public class RoutingSystemPropertyKeys {
*/
public static final String UPDATE_ROUTING_DATA_ON_CACHE_MISS =
"update.routing.data.on.cache.miss.enabled";
+
+ /**
+ * The interval to use between routing data updates from the routing data source.
+ */
+ public static final String ROUTING_DATA_UPDATE_INTERVAL_MS = "routing.data.update.interval.ms";
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
index 4354537..dc55d53 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
@@ -30,6 +30,7 @@ import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.api.client.ChildrenSubscribeResult;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.constant.RoutingDataConstants;
import org.apache.helix.zookeeper.constant.RoutingSystemPropertyKeys;
import org.apache.helix.zookeeper.exception.MultiZkException;
import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
@@ -86,6 +87,7 @@ public class FederatedZkClient implements RealmAwareZkClient {
private PathBasedZkSerializer _pathBasedZkSerializer;
private final boolean _routingDataUpdateOnCacheMissEnabled = Boolean.parseBoolean(
System.getProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS));
+ private long _routingDataUpdateInterval;
// TODO: support capacity of ZkClient number in one FederatedZkClient and do garbage collection.
public FederatedZkClient(RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig,
@@ -102,6 +104,7 @@ public class FederatedZkClient implements RealmAwareZkClient {
_clientConfig = clientConfig;
_pathBasedZkSerializer = clientConfig.getZkSerializer();
_zkRealmToZkClientMap = new ConcurrentHashMap<>();
+ getRoutingDataUpdateInterval();
}
@Override
@@ -587,6 +590,11 @@ public class FederatedZkClient implements RealmAwareZkClient {
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.
@@ -626,4 +634,34 @@ public class FederatedZkClient implements RealmAwareZkClient {
+ ". Instead, please use " + DEDICATED_ZK_CLIENT_FACTORY
+ " to create a dedicated RealmAwareZkClient for this operation.");
}
+
+ /**
+ * Resolves the routing data update interval value from System Properties.
+ */
+ private void getRoutingDataUpdateInterval() {
+ try {
+ _routingDataUpdateInterval = Long.parseLong(
+ System.getProperty(RoutingSystemPropertyKeys.ROUTING_DATA_UPDATE_INTERVAL_MS));
+ if (_routingDataUpdateInterval < 0) {
+ LOG.warn("FederatedZkClient::shouldThrottleRead(): invalid value: {} given for "
+ + "ROUTING_DATA_UPDATE_INTERVAL_MS, using the default value (5 sec) instead!",
+ _routingDataUpdateInterval);
+ _routingDataUpdateInterval = RoutingDataConstants.DEFAULT_ROUTING_DATA_UPDATE_INTERVAL_MS;
+ }
+ } catch (NumberFormatException e) {
+ LOG.warn("FederatedZkClient::shouldThrottleRead(): failed to parse "
+ + "ROUTING_DATA_UPDATE_INTERVAL_MS, using the default value (5 sec) instead!", e);
+ _routingDataUpdateInterval = RoutingDataConstants.DEFAULT_ROUTING_DATA_UPDATE_INTERVAL_MS;
+ }
+ }
+
+ /**
+ * Return whether the read request to routing data source should be throttled using the default
+ * routing data update interval.
+ * @return
+ */
+ private boolean shouldThrottleRead() {
+ return System.currentTimeMillis() - RoutingDataManager.getInstance().getLastResetTimestamp()
+ < _routingDataUpdateInterval;
+ }
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
index 6df9616..853bd5c 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
@@ -51,6 +51,9 @@ public class RoutingDataManager {
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;
@@ -164,6 +167,15 @@ public class RoutingDataManager {
_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;
}
/**
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java
index 93e5892..e201905 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java
@@ -279,6 +279,8 @@ public class TestFederatedZkClient extends RealmAwareZkClientTestBase {
throws IOException, InvalidRoutingDataException {
// Enable routing data update upon cache miss
System.setProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS, "true");
+ // Set the routing data update interval to 0 so there's no delay in testing
+ System.setProperty(RoutingSystemPropertyKeys.ROUTING_DATA_UPDATE_INTERVAL_MS, "0");
RoutingDataManager.getInstance().getMetadataStoreRoutingData();
_msdsServer.stopServer();
@@ -375,7 +377,8 @@ public class TestFederatedZkClient extends RealmAwareZkClientTestBase {
// Shut down MSDS
_msdsServer.stopServer();
// Disable System property
- System.setProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS, "false");
+ System.clearProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS);
+ System.clearProperty(RoutingSystemPropertyKeys.ROUTING_DATA_UPDATE_INTERVAL_MS);
}
/**
@@ -402,6 +405,8 @@ public class TestFederatedZkClient extends RealmAwareZkClientTestBase {
// Enable routing data update upon cache miss
System.setProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS, "true");
+ // Set the routing data update interval to 0 so there's no delay in testing
+ System.setProperty(RoutingSystemPropertyKeys.ROUTING_DATA_UPDATE_INTERVAL_MS, "0");
RoutingDataManager.getInstance().reset();
RoutingDataManager.getInstance().getMetadataStoreRoutingData(RoutingDataReaderType.ZK, zkRealm);
@@ -496,7 +501,66 @@ public class TestFederatedZkClient extends RealmAwareZkClientTestBase {
zkClient.deleteRecursively(MetadataStoreRoutingConstants.ROUTING_DATA_PATH);
zkClient.close();
// Disable System property
- System.setProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS, "false");
+ System.clearProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS);
+ System.clearProperty(RoutingSystemPropertyKeys.ROUTING_DATA_UPDATE_INTERVAL_MS);
+ }
+
+ /**
+ * Test that throttle based on last reset timestamp works correctly. Here, we use ZK as the
+ * routing data source.
+ * Test scenario: set the throttle value to a high value and check that routing data update from
+ * the routing data source does NOT happen (because it would be throttled).
+ */
+ @Test(dependsOnMethods = "testUpdateRoutingDataOnCacheMissZK")
+ public void testRoutingDataUpdateThrottle() throws InvalidRoutingDataException {
+ // Call reset to set the last reset() timestamp in RoutingDataManager
+ RoutingDataManager.getInstance().reset();
+
+ // Set up routing data in ZK with empty sharding key list
+ String zkRealm = "localhost:2127";
+ String newShardingKey = "/throttle";
+ ZkClient zkClient =
+ new ZkClient.Builder().setZkServer(zkRealm).setZkSerializer(new ZNRecordSerializer())
+ .build();
+ zkClient.create(MetadataStoreRoutingConstants.ROUTING_DATA_PATH, null, CreateMode.PERSISTENT);
+ ZNRecord zkRealmRecord = new ZNRecord(zkRealm);
+ zkRealmRecord.setListField(MetadataStoreRoutingConstants.ZNRECORD_LIST_FIELD_KEY,
+ new ArrayList<>(TestConstants.TEST_KEY_LIST_1));
+ zkClient.create(MetadataStoreRoutingConstants.ROUTING_DATA_PATH + "/" + zkRealm, zkRealmRecord,
+ CreateMode.PERSISTENT);
+
+ // Enable routing data update upon cache miss
+ System.setProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS, "true");
+ // Set the throttle value to a very long value
+ System.setProperty(RoutingSystemPropertyKeys.ROUTING_DATA_UPDATE_INTERVAL_MS,
+ String.valueOf(Integer.MAX_VALUE));
+
+ // Create a new FederatedZkClient, whose _routingDataUpdateInterval should be MAX_VALUE
+ FederatedZkClient federatedZkClient = new FederatedZkClient(
+ new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder()
+ .setRoutingDataSourceType(RoutingDataReaderType.ZK.name())
+ .setRoutingDataSourceEndpoint(zkRealm).build(),
+ new RealmAwareZkClient.RealmAwareZkClientConfig());
+
+ // Add newShardingKey to ZK's routing data
+ zkRealmRecord.getListField(MetadataStoreRoutingConstants.ZNRECORD_LIST_FIELD_KEY)
+ .add(newShardingKey);
+ zkClient
+ .writeData(MetadataStoreRoutingConstants.ROUTING_DATA_PATH + "/" + zkRealm, zkRealmRecord);
+
+ try {
+ Assert.assertFalse(federatedZkClient.exists(newShardingKey));
+ Assert.fail("NoSuchElementException expected!");
+ } catch (NoSuchElementException e) {
+ // Expected because it should not read from the routing data source because of the throttle
+ }
+
+ // Clean up
+ zkClient.deleteRecursively(MetadataStoreRoutingConstants.ROUTING_DATA_PATH);
+ zkClient.close();
+ federatedZkClient.close();
+ System.clearProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS);
+ System.clearProperty(RoutingSystemPropertyKeys.ROUTING_DATA_UPDATE_INTERVAL_MS);
}
/*
@@ -504,7 +568,7 @@ public class TestFederatedZkClient extends RealmAwareZkClientTestBase {
* TODO: test that all raw zkClients are closed after FederatedZkClient close() is called. This
* could help avoid ZkClient leakage.
*/
- @Test(dependsOnMethods = "testUpdateRoutingDataOnCacheMissZK")
+ @Test(dependsOnMethods = "testRoutingDataUpdateThrottle")
public void testClose() {
Assert.assertFalse(_realmAwareZkClient.isClosed());
[helix] 09/12: Refactor RealmAwareZkClient code to remove duplicate
code
Posted by hu...@apache.org.
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
commit c876649818675c546f69083cdf75cd1cb2f4351b
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Fri Jul 24 18:38:24 2020 -0700
Refactor RealmAwareZkClient code to remove duplicate code
There was a piece of code that resolves MetadataStoreRoutingData based on RealmAwareZkConnectionConfig in all implementations of RealmAwareZkClient. This commit refactors that logic into a common static method in RealmAwareZkClient.
---
.../helix/manager/zk/GenericZkHelixApiBuilder.java | 9 +--------
.../zookeeper/api/client/RealmAwareZkClient.java | 21 +++++++++++++++++++++
.../constant/RoutingSystemPropertyKeys.java | 6 ++++--
.../zookeeper/impl/client/DedicatedZkClient.java | 13 +------------
.../zookeeper/impl/client/FederatedZkClient.java | 13 +------------
.../helix/zookeeper/impl/client/SharedZkClient.java | 13 +------------
6 files changed, 29 insertions(+), 46 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java b/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
index 1a4a69b..0447eee 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
@@ -200,15 +200,8 @@ public abstract class GenericZkHelixApiBuilder<B extends GenericZkHelixApiBuilde
private String resolveZkAddressWithShardingKey(
RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig)
throws InvalidRoutingDataException {
- boolean isRoutingDataSourceEndpointSet =
- connectionConfig.getRoutingDataSourceEndpoint() != null && !connectionConfig
- .getRoutingDataSourceEndpoint().isEmpty();
MetadataStoreRoutingData routingData =
- isRoutingDataSourceEndpointSet ? RoutingDataManager.getInstance()
- .getMetadataStoreRoutingData(
- RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
- connectionConfig.getRoutingDataSourceEndpoint())
- : RoutingDataManager.getInstance().getMetadataStoreRoutingData();
+ RealmAwareZkClient.getMetadataStoreRoutingData(connectionConfig);
return routingData.getMetadataStoreRealm(connectionConfig.getZkRealmShardingKey());
}
}
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 e8c91f1..d006420 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
@@ -23,7 +23,10 @@ import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
+import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
+import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
+import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
@@ -640,4 +643,22 @@ public interface RealmAwareZkClient {
dataListener);
}
}
+
+ /**
+ * Based on the RealmAwareZkConnectionConfig given, return MetadataStoreRoutingData.
+ * @param connectionConfig
+ * @return
+ */
+ static MetadataStoreRoutingData getMetadataStoreRoutingData(
+ RealmAwareZkConnectionConfig connectionConfig) throws InvalidRoutingDataException {
+ String routingDataSourceEndpoint = connectionConfig.getRoutingDataSourceEndpoint();
+ if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
+ // If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
+ return RoutingDataManager.getInstance().getMetadataStoreRoutingData();
+ } else {
+ return RoutingDataManager.getInstance().getMetadataStoreRoutingData(
+ RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
+ routingDataSourceEndpoint);
+ }
+ }
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
index 23b8ebc..a57075b 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
@@ -25,7 +25,9 @@ package org.apache.helix.zookeeper.constant;
public class RoutingSystemPropertyKeys {
/**
- * This static constant is used to refer to which implementation of RoutingDataReader to use.
+ * If enabled, FederatedZkClient (multiZkClient) will invalidate the cached routing data and
+ * re-read the routing data from the routing data source upon ZK path sharding key cache miss.
*/
- public static final String ROUTING_DATA_READER_TYPE = "routing.data.reader.type";
+ public static final String UPDATE_ROUTING_DATA_ON_CACHE_MISS =
+ "update.routing.data.on.cache.miss.enabled";
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
index 4527fe8..cc37853 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
@@ -80,18 +80,7 @@ public class DedicatedZkClient implements RealmAwareZkClient {
}
_connectionConfig = connectionConfig;
_clientConfig = clientConfig;
-
- // Get MetadataStoreRoutingData
- String routingDataSourceEndpoint = connectionConfig.getRoutingDataSourceEndpoint();
- if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
- // If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
- _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData();
- } else {
- _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData(
- RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
- routingDataSourceEndpoint);
- }
-
+ _metadataStoreRoutingData = RealmAwareZkClient.getMetadataStoreRoutingData(connectionConfig);
_zkRealmShardingKey = connectionConfig.getZkRealmShardingKey();
if (_zkRealmShardingKey == null || _zkRealmShardingKey.isEmpty()) {
throw new IllegalArgumentException(
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
index 11d0291..679d4e7 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
@@ -93,18 +93,7 @@ public class FederatedZkClient implements RealmAwareZkClient {
if (clientConfig == null) {
throw new IllegalArgumentException("RealmAwareZkClientConfig cannot be null!");
}
-
- // Get MetadataStoreRoutingData
- String routingDataSourceEndpoint = connectionConfig.getRoutingDataSourceEndpoint();
- if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
- // If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
- _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData();
- } else {
- _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData(
- RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
- routingDataSourceEndpoint);
- }
-
+ _metadataStoreRoutingData = RealmAwareZkClient.getMetadataStoreRoutingData(connectionConfig);
_isClosed = false;
_connectionConfig = connectionConfig;
_clientConfig = clientConfig;
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 14a4794..bcc4b12 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
@@ -74,23 +74,12 @@ public class SharedZkClient implements RealmAwareZkClient {
_connectionConfig = connectionConfig;
_clientConfig = clientConfig;
- // Get MetadataStoreRoutingData
- String routingDataSourceEndpoint = connectionConfig.getRoutingDataSourceEndpoint();
- if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
- // If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
- _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData();
- } else {
- _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData(
- RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
- routingDataSourceEndpoint);
- }
-
_zkRealmShardingKey = connectionConfig.getZkRealmShardingKey();
if (_zkRealmShardingKey == null || _zkRealmShardingKey.isEmpty()) {
throw new IllegalArgumentException(
"RealmAwareZkConnectionConfig's ZK realm sharding key cannot be null or empty for SharedZkClient!");
}
-
+ _metadataStoreRoutingData = RealmAwareZkClient.getMetadataStoreRoutingData(connectionConfig);
// Get the ZkRealm address based on the ZK path sharding key
String zkRealmAddress = _metadataStoreRoutingData.getMetadataStoreRealm(_zkRealmShardingKey);
if (zkRealmAddress == null || zkRealmAddress.isEmpty()) {
[helix] 07/12: Modify realm-aware ZkClient and Helix API for
configurable routing source
Posted by hu...@apache.org.
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
commit 6014df6ddc89cb62163b8448cc408075b1691dc4
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Fri Jul 24 15:01:04 2020 -0700
Modify realm-aware ZkClient and Helix API for configurable routing source
This commit changes old MSDS-based interfaces and replaces them with a more generic configurable routing data source interfaces. This commit also adds test cases for Helix API.
---
.../main/java/org/apache/helix/ConfigAccessor.java | 7 +-
.../manager/zk/GenericBaseDataAccessorBuilder.java | 3 +-
.../helix/manager/zk/GenericZkHelixApiBuilder.java | 21 +++---
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 14 ++--
.../java/org/apache/helix/manager/zk/ZKUtil.java | 3 +-
.../helix/manager/zk/ZkBaseDataAccessor.java | 7 +-
.../helix/manager/zk/ZkBucketDataAccessor.java | 2 +-
.../java/org/apache/helix/tools/ClusterSetup.java | 8 +-
.../multizk/TestMultiZkHelixJavaApis.java | 88 ++++++++++++++++++++--
.../apache/helix/rest/server/ServerContext.java | 5 +-
.../zookeeper/api/client/RealmAwareZkClient.java | 44 ++++++++---
.../zookeeper/constant/RoutingDataReaderType.java | 14 ++++
.../zookeeper/impl/client/DedicatedZkClient.java | 18 ++---
.../zookeeper/impl/client/FederatedZkClient.java | 17 ++---
.../zookeeper/impl/client/SharedZkClient.java | 17 ++---
15 files changed, 183 insertions(+), 85 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/ConfigAccessor.java b/helix-core/src/main/java/org/apache/helix/ConfigAccessor.java
index 48bcfd4..4885f31 100644
--- a/helix-core/src/main/java/org/apache/helix/ConfigAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/ConfigAccessor.java
@@ -19,7 +19,6 @@ package org.apache.helix;
* under the License.
*/
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -30,8 +29,8 @@ import java.util.TreeMap;
import org.apache.helix.manager.zk.GenericZkHelixApiBuilder;
import org.apache.helix.manager.zk.ZKUtil;
-import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.CloudConfig;
+import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.ConfigScope;
import org.apache.helix.model.CustomizedStateConfig;
import org.apache.helix.model.HelixConfigScope;
@@ -40,8 +39,8 @@ import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.RESTConfig;
import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
-import org.apache.helix.util.HelixUtil;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
+import org.apache.helix.util.HelixUtil;
import org.apache.helix.util.StringTemplate;
import org.apache.helix.zookeeper.api.client.HelixZkClient;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
@@ -120,7 +119,7 @@ public class ConfigAccessor {
new RealmAwareZkClient.RealmAwareZkClientConfig()
.setZkSerializer(new ZNRecordSerializer()));
return;
- } catch (IOException | InvalidRoutingDataException | IllegalStateException e) {
+ } catch (InvalidRoutingDataException | IllegalStateException e) {
throw new HelixException("Failed to create ConfigAccessor!", e);
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/GenericBaseDataAccessorBuilder.java b/helix-core/src/main/java/org/apache/helix/manager/zk/GenericBaseDataAccessorBuilder.java
index 054e693..a75c8cf 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/GenericBaseDataAccessorBuilder.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/GenericBaseDataAccessorBuilder.java
@@ -19,7 +19,6 @@ package org.apache.helix.manager.zk;
* under the License.
*/
-import java.io.IOException;
import java.util.concurrent.TimeUnit;
import org.apache.helix.HelixException;
@@ -83,7 +82,7 @@ public class GenericBaseDataAccessorBuilder<B extends GenericBaseDataAccessorBui
case MULTI_REALM:
try {
zkClient = new FederatedZkClient(connectionConfig, clientConfig);
- } catch (IOException | InvalidRoutingDataException e) {
+ } catch (InvalidRoutingDataException e) {
throw new HelixException("Not able to connect on multi-realm mode.", e);
}
break;
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java b/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
index 840ec8f..d02f67e 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
@@ -19,8 +19,6 @@ package org.apache.helix.manager.zk;
* under the License.
*/
-import java.io.IOException;
-
import org.apache.helix.HelixException;
import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
@@ -109,7 +107,7 @@ public abstract class GenericZkHelixApiBuilder<B extends GenericZkHelixApiBuilde
try {
_zkAddress = resolveZkAddressWithShardingKey(_realmAwareZkConnectionConfig);
isZkAddressSet = true;
- } catch (IOException | InvalidRoutingDataException e) {
+ } catch (InvalidRoutingDataException e) {
LOG.warn(
"GenericZkHelixApiBuilder: ZkAddress is not set and failed to resolve ZkAddress with ZK path sharding key!",
e);
@@ -166,7 +164,7 @@ public abstract class GenericZkHelixApiBuilder<B extends GenericZkHelixApiBuilde
try {
return new FederatedZkClient(connectionConfig,
clientConfig.setZkSerializer(new ZNRecordSerializer()));
- } catch (IOException | InvalidRoutingDataException | IllegalStateException e) {
+ } catch (InvalidRoutingDataException | IllegalStateException e) {
throw new HelixException("GenericZkHelixApiBuilder: Failed to create FederatedZkClient!",
e);
}
@@ -197,17 +195,18 @@ public abstract class GenericZkHelixApiBuilder<B extends GenericZkHelixApiBuilde
* ZK address is not given in this Builder.
* @param connectionConfig
* @return
- * @throws IOException
* @throws InvalidRoutingDataException
*/
private String resolveZkAddressWithShardingKey(
RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig)
- throws IOException, InvalidRoutingDataException {
- boolean isMsdsEndpointSet =
- connectionConfig.getMsdsEndpoint() != null && !connectionConfig.getMsdsEndpoint().isEmpty();
- // TODO: Make RoutingDataReaderType configurable
- MetadataStoreRoutingData routingData = isMsdsEndpointSet ? RoutingDataManager
- .getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, connectionConfig.getMsdsEndpoint())
+ throws InvalidRoutingDataException {
+ boolean isRoutingDataSourceEndpointSet =
+ connectionConfig.getRoutingDataSourceEndpoint() != null && !connectionConfig
+ .getRoutingDataSourceEndpoint().isEmpty();
+ MetadataStoreRoutingData routingData = isRoutingDataSourceEndpointSet ? RoutingDataManager
+ .getMetadataStoreRoutingData(
+ RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
+ connectionConfig.getRoutingDataSourceEndpoint())
: RoutingDataManager.getMetadataStoreRoutingData();
return routingData.getMetadataStoreRealm(connectionConfig.getZkRealmShardingKey());
}
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
index 348f8d8..1759116 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
@@ -151,7 +151,7 @@ public class ZKHelixAdmin implements HelixAdmin {
try {
zkClient = new FederatedZkClient(
new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder().build(), clientConfig);
- } catch (IllegalStateException | IOException | InvalidRoutingDataException e) {
+ } catch (IllegalStateException | InvalidRoutingDataException e) {
throw new HelixException("Not able to connect on multi-realm mode.", e);
}
} else {
@@ -965,13 +965,15 @@ public class ZKHelixAdmin implements HelixAdmin {
|| _zkClient instanceof FederatedZkClient) {
// If on multi-zk mode, we retrieve cluster information from Metadata Store Directory Service.
Map<String, List<String>> realmToShardingKeys;
- String msdsEndpoint = _zkClient.getRealmAwareZkConnectionConfig().getMsdsEndpoint();
- if (msdsEndpoint == null || msdsEndpoint.isEmpty()) {
+ String routingDataSourceEndpoint =
+ _zkClient.getRealmAwareZkConnectionConfig().getRoutingDataSourceEndpoint();
+ if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
+ // If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
realmToShardingKeys = RoutingDataManager.getRawRoutingData();
} else {
- // TODO: Make RoutingDataReaderType configurable
- realmToShardingKeys =
- RoutingDataManager.getRawRoutingData(RoutingDataReaderType.HTTP, msdsEndpoint);
+ realmToShardingKeys = RoutingDataManager.getRawRoutingData(RoutingDataReaderType
+ .lookUp(_zkClient.getRealmAwareZkConnectionConfig().getRoutingDataSourceType()),
+ routingDataSourceEndpoint);
}
if (realmToShardingKeys == null || realmToShardingKeys.isEmpty()) {
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKUtil.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKUtil.java
index cda5f39..1015834 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKUtil.java
@@ -19,7 +19,6 @@ package org.apache.helix.manager.zk;
* under the License.
*/
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -622,7 +621,7 @@ public final class ZKUtil {
RealmAwareZkClient.RealmAwareZkClientConfig clientConfig =
new RealmAwareZkClient.RealmAwareZkClientConfig();
return new FederatedZkClient(connectionConfig, clientConfig);
- } catch (IllegalArgumentException | IOException | InvalidRoutingDataException e) {
+ } catch (IllegalArgumentException | InvalidRoutingDataException e) {
throw new HelixException("Not able to connect on realm-aware mode", e);
}
}
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 d2096b0..4e6a7b0 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
@@ -19,7 +19,6 @@ package org.apache.helix.manager.zk;
* under the License.
*/
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -40,6 +39,7 @@ import org.apache.helix.util.HelixUtil;
import org.apache.helix.zookeeper.api.client.HelixZkClient;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.helix.zookeeper.exception.ZkClientException;
import org.apache.helix.zookeeper.impl.client.FederatedZkClient;
import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
@@ -52,9 +52,8 @@ import org.apache.helix.zookeeper.zkclient.exception.ZkBadVersionException;
import org.apache.helix.zookeeper.zkclient.exception.ZkException;
import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
import org.apache.helix.zookeeper.zkclient.exception.ZkNodeExistsException;
-import org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer;
import org.apache.helix.zookeeper.zkclient.serialize.PathBasedZkSerializer;
-import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat;
@@ -1342,7 +1341,7 @@ public class ZkBaseDataAccessor<T> implements BaseDataAccessor<T> {
try {
return new FederatedZkClient(
new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder().build(), clientConfig);
- } catch (IllegalStateException | IOException | InvalidRoutingDataException e) {
+ } catch (IllegalStateException | InvalidRoutingDataException e) {
throw new HelixException("Not able to connect on multi-realm mode.", e);
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
index 8f0aa16..efdde81 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
@@ -94,7 +94,7 @@ public class ZkBucketDataAccessor implements BucketDataAccessor, AutoCloseable {
RealmAwareZkClient.RealmAwareZkClientConfig clientConfig =
new RealmAwareZkClient.RealmAwareZkClientConfig();
_zkClient = new FederatedZkClient(connectionConfig, clientConfig);
- } catch (IllegalArgumentException | IOException | InvalidRoutingDataException e) {
+ } catch (IllegalArgumentException | InvalidRoutingDataException e) {
throw new HelixException("Not able to connect on realm-aware mode", e);
}
} else {
diff --git a/helix-core/src/main/java/org/apache/helix/tools/ClusterSetup.java b/helix-core/src/main/java/org/apache/helix/tools/ClusterSetup.java
index 59180eb..667cab7 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ClusterSetup.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ClusterSetup.java
@@ -40,12 +40,10 @@ import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixConstants;
import org.apache.helix.HelixException;
-import org.apache.helix.PropertyKey.Builder;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
-import org.apache.helix.cloud.azure.AzureConstants;
-import org.apache.helix.cloud.constants.CloudProvider;
import org.apache.helix.PropertyKey;
import org.apache.helix.SystemPropertyKeys;
+import org.apache.helix.cloud.azure.AzureConstants;
+import org.apache.helix.cloud.constants.CloudProvider;
import org.apache.helix.manager.zk.GenericZkHelixApiBuilder;
import org.apache.helix.manager.zk.ZKHelixAdmin;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
@@ -162,7 +160,7 @@ public class ClusterSetup {
new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder().build(),
new RealmAwareZkClient.RealmAwareZkClientConfig()
.setZkSerializer(new ZNRecordSerializer()));
- } catch (IOException | InvalidRoutingDataException | IllegalStateException e) {
+ } catch (InvalidRoutingDataException | IllegalStateException e) {
throw new HelixException("Failed to create ConfigAccessor!", e);
}
} else {
diff --git a/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java b/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java
index 5146ce6..d72318a 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java
@@ -20,6 +20,7 @@ package org.apache.helix.integration.multizk;
*/
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -73,11 +74,14 @@ import org.apache.helix.tools.ClusterVerifiers.ZkHelixClusterVerifier;
import org.apache.helix.zookeeper.api.client.HelixZkClient;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
import org.apache.helix.zookeeper.api.client.ZkClientType;
+import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.exception.MultiZkException;
import org.apache.helix.zookeeper.impl.client.FederatedZkClient;
import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.ZkServer;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -109,16 +113,17 @@ public class TestMultiZkHelixJavaApis {
private HelixAdmin _zkHelixAdmin;
// Save System property configs from before this test and pass onto after the test
- private Map<String, String> _configStore = new HashMap<>();
+ private final Map<String, String> _configStore = new HashMap<>();
+
+ private static final String ZK_PREFIX = "localhost:";
+ private static final int ZK_START_PORT = 8777;
+ private String _msdsEndpoint;
@BeforeClass
public void beforeClass() throws Exception {
// Create 3 in-memory zookeepers and routing mapping
- final String zkPrefix = "localhost:";
- final int zkStartPort = 8777;
-
for (int i = 0; i < NUM_ZK; i++) {
- String zkAddress = zkPrefix + (zkStartPort + i);
+ String zkAddress = ZK_PREFIX + (ZK_START_PORT + i);
ZK_SERVER_MAP.put(zkAddress, TestHelper.startZkServer(zkAddress));
ZK_CLIENT_MAP.put(zkAddress, DedicatedZkClientFactory.getInstance()
.buildZkClient(new HelixZkClient.ZkConnectionConfig(zkAddress),
@@ -132,6 +137,8 @@ public class TestMultiZkHelixJavaApis {
final String msdsHostName = "localhost";
final int msdsPort = 11117;
final String msdsNamespace = "multiZkTest";
+ _msdsEndpoint =
+ "http://" + msdsHostName + ":" + msdsPort + "/admin/v2/namespaces/" + msdsNamespace;
_msds = new MockMetadataStoreDirectoryServer(msdsHostName, msdsPort, msdsNamespace,
_rawRoutingData);
_msds.startServer();
@@ -151,8 +158,7 @@ public class TestMultiZkHelixJavaApis {
// Turn on multiZk mode in System config
System.setProperty(SystemPropertyKeys.MULTI_ZK_ENABLED, "true");
// MSDS endpoint: http://localhost:11117/admin/v2/namespaces/multiZkTest
- System.setProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY,
- "http://" + msdsHostName + ":" + msdsPort + "/admin/v2/namespaces/" + msdsNamespace);
+ System.setProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY, _msdsEndpoint);
// Create a FederatedZkClient for admin work
_zkClient =
@@ -782,7 +788,8 @@ public class TestMultiZkHelixJavaApis {
new MockMetadataStoreDirectoryServer("localhost", 11118, "multiZkTest", secondRoutingData);
final RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig =
new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder()
- .setMsdsEndpoint(secondMsds.getEndpoint()).build();
+ .setRoutingDataSourceType(RoutingDataReaderType.HTTP.name())
+ .setRoutingDataSourceEndpoint(secondMsds.getEndpoint()).build();
secondMsds.startServer();
try {
@@ -996,4 +1003,69 @@ public class TestMultiZkHelixJavaApis {
}
return sb.toString();
}
+
+ /**
+ * Testing using ZK as the routing data source. We use BaseDataAccessor as the representative
+ * Helix API.
+ * Two modes are tested: ZK and HTTP-ZK fallback
+ */
+ @Test(dependsOnMethods = "testDifferentMsdsEndpointConfigs")
+ public void testZkRoutingDataSourceConfigs() {
+ // Set up routing data in ZK by connecting directly to ZK
+ BaseDataAccessor<ZNRecord> accessor =
+ new ZkBaseDataAccessor.Builder<ZNRecord>().setZkAddress(ZK_PREFIX + ZK_START_PORT).build();
+
+ // Create ZK realm routing data ZNRecord
+ _rawRoutingData.forEach((realm, keys) -> {
+ ZNRecord znRecord = new ZNRecord(realm);
+ znRecord.setListField(MetadataStoreRoutingConstants.ZNRECORD_LIST_FIELD_KEY,
+ new ArrayList<>(keys));
+ accessor.set(MetadataStoreRoutingConstants.ROUTING_DATA_PATH + "/" + realm, znRecord,
+ AccessOption.PERSISTENT);
+ });
+
+ // Create connection configs with the source type set to each type
+ final RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder connectionConfigBuilder =
+ new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder();
+ final RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfigZk =
+ connectionConfigBuilder.setRoutingDataSourceType(RoutingDataReaderType.ZK.name())
+ .setRoutingDataSourceEndpoint(ZK_PREFIX + ZK_START_PORT).build();
+ final RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfigHttpZkFallback =
+ connectionConfigBuilder
+ .setRoutingDataSourceType(RoutingDataReaderType.HTTP_ZK_FALLBACK.name())
+ .setRoutingDataSourceEndpoint(_msdsEndpoint + "," + ZK_PREFIX + ZK_START_PORT).build();
+ final RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfigHttp =
+ connectionConfigBuilder.setRoutingDataSourceType(RoutingDataReaderType.HTTP.name())
+ .setRoutingDataSourceEndpoint(_msdsEndpoint).build();
+
+ // Reset cached routing data
+ RoutingDataManager.reset();
+ // Shutdown MSDS to ensure that these accessors are able to pull routing data from ZK
+ _msds.stopServer();
+
+ // Create a BaseDataAccessor instance with the connection config
+ BaseDataAccessor<ZNRecord> zkBasedAccessor = new ZkBaseDataAccessor.Builder<ZNRecord>()
+ .setRealmAwareZkConnectionConfig(connectionConfigZk).build();
+ BaseDataAccessor<ZNRecord> httpZkFallbackBasedAccessor =
+ new ZkBaseDataAccessor.Builder<ZNRecord>()
+ .setRealmAwareZkConnectionConfig(connectionConfigHttpZkFallback).build();
+ try {
+ BaseDataAccessor<ZNRecord> httpBasedAccessor = new ZkBaseDataAccessor.Builder<ZNRecord>()
+ .setRealmAwareZkConnectionConfig(connectionConfigHttp).build();
+ Assert.fail("Must fail with a MultiZkException because HTTP connection will be refused.");
+ } catch (MultiZkException e) {
+ // Okay
+ }
+
+ // Check that all clusters appear as existing to this accessor
+ CLUSTER_LIST.forEach(cluster -> {
+ Assert.assertTrue(zkBasedAccessor.exists("/" + cluster, AccessOption.PERSISTENT));
+ Assert.assertTrue(httpZkFallbackBasedAccessor.exists("/" + cluster, AccessOption.PERSISTENT));
+ });
+
+ // Close all connections
+ accessor.close();
+ zkBasedAccessor.close();
+ httpZkFallbackBasedAccessor.close();
+ }
}
diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java b/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
index f5d8915..176180f 100644
--- a/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
+++ b/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
@@ -20,7 +20,6 @@ package org.apache.helix.rest.server;
* under the License.
*/
-import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -134,13 +133,13 @@ public class ServerContext implements IZkDataListener, IZkChildListener, IZkStat
new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder();
// If MSDS endpoint is set for this namespace, use that instead.
if (_msdsEndpoint != null && !_msdsEndpoint.isEmpty()) {
- connectionConfigBuilder.setMsdsEndpoint(_msdsEndpoint);
+ connectionConfigBuilder.setRoutingDataSourceEndpoint(_msdsEndpoint);
}
_zkClient = new FederatedZkClient(connectionConfigBuilder.build(),
new RealmAwareZkClient.RealmAwareZkClientConfig()
.setZkSerializer(new ZNRecordSerializer()));
LOG.info("ServerContext: FederatedZkClient created successfully!");
- } catch (IOException | InvalidRoutingDataException | IllegalStateException e) {
+ } catch (InvalidRoutingDataException | IllegalStateException e) {
throw new HelixException("Failed to create FederatedZkClient!", e);
}
} else {
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 b1ecc38..e8c91f1 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
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
+import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
@@ -56,8 +57,7 @@ public interface RealmAwareZkClient {
* MULTI_REALM: CRUD and change subscription are supported. Operations involving EPHEMERAL CreateMode will throw an UnsupportedOperationException.
*/
enum RealmMode {
- SINGLE_REALM,
- MULTI_REALM
+ SINGLE_REALM, MULTI_REALM
}
int DEFAULT_OPERATION_TIMEOUT = Integer.MAX_VALUE;
@@ -65,6 +65,7 @@ public interface RealmAwareZkClient {
int DEFAULT_SESSION_TIMEOUT = 30 * 1000;
// listener subscription
+
/**
* Subscribe the path and the listener will handle child events of the path.
* Add exists watch to path if the path does not exist in ZooKeeper server.
@@ -86,7 +87,8 @@ public interface RealmAwareZkClient {
* @return ChildrentSubsribeResult. If the path does not exists, the isInstalled field
* is false. Otherwise, it is true and list of children are returned.
*/
- ChildrenSubscribeResult subscribeChildChanges(String path, IZkChildListener listener, boolean skipWatchingNonExistNode);
+ ChildrenSubscribeResult subscribeChildChanges(String path, IZkChildListener listener,
+ boolean skipWatchingNonExistNode);
void unsubscribeChildChanges(String path, IZkChildListener listener);
@@ -108,7 +110,8 @@ public interface RealmAwareZkClient {
* @param skipWatchingNonExistNode True means not installing any watch if path does not exist.
* return True if installation of watch succeed. Otherwise, return false.
*/
- boolean subscribeDataChanges(String path, IZkDataListener listener, boolean skipWatchingNonExistNode);
+ boolean subscribeDataChanges(String path, IZkDataListener listener,
+ boolean skipWatchingNonExistNode);
void unsubscribeDataChanges(String path, IZkDataListener listener);
@@ -378,12 +381,14 @@ public interface RealmAwareZkClient {
* NOTE: this field will be ignored if RealmMode is MULTI_REALM!
*/
private String _zkRealmShardingKey;
- private String _msdsEndpoint;
+ private String _routingDataSourceType;
+ private String _routingDataSourceEndpoint;
private int _sessionTimeout = DEFAULT_SESSION_TIMEOUT;
private RealmAwareZkConnectionConfig(Builder builder) {
_zkRealmShardingKey = builder._zkRealmShardingKey;
- _msdsEndpoint = builder._msdsEndpoint;
+ _routingDataSourceType = builder._routingDataSourceType;
+ _routingDataSourceEndpoint = builder._routingDataSourceEndpoint;
_sessionTimeout = builder._sessionTimeout;
}
@@ -424,14 +429,19 @@ public interface RealmAwareZkClient {
return _sessionTimeout;
}
- public String getMsdsEndpoint() {
- return _msdsEndpoint;
+ public String getRoutingDataSourceType() {
+ return _routingDataSourceType;
+ }
+
+ public String getRoutingDataSourceEndpoint() {
+ return _routingDataSourceEndpoint;
}
public static class Builder {
private RealmMode _realmMode;
private String _zkRealmShardingKey;
- private String _msdsEndpoint;
+ private String _routingDataSourceType;
+ private String _routingDataSourceEndpoint;
private int _sessionTimeout = DEFAULT_SESSION_TIMEOUT;
public Builder() {
@@ -453,8 +463,13 @@ public interface RealmAwareZkClient {
return this;
}
- public Builder setMsdsEndpoint(String msdsEndpoint) {
- _msdsEndpoint = msdsEndpoint;
+ public Builder setRoutingDataSourceType(String routingDataSourceType) {
+ _routingDataSourceType = routingDataSourceType;
+ return this;
+ }
+
+ public Builder setRoutingDataSourceEndpoint(String routingDataSourceEndpoint) {
+ _routingDataSourceEndpoint = routingDataSourceEndpoint;
return this;
}
@@ -482,6 +497,13 @@ public interface RealmAwareZkClient {
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
+ throw new IllegalArgumentException(
+ "RealmAwareZkConnectionConfig.Builder: routing data source type and endpoint are not configured properly! Type: "
+ + _routingDataSourceType + " Endpoint: " + _routingDataSourceEndpoint);
+ }
}
}
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java
index aedef36..33b6c70 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java
@@ -19,6 +19,10 @@ package org.apache.helix.zookeeper.constant;
* under the License.
*/
+import org.apache.commons.lang3.EnumUtils;
+import org.apache.helix.zookeeper.exception.MultiZkException;
+
+
/**
* RoutingDataReaderType is an enum that designates the reader type and the class name that can be
* used to create an instance of RoutingDataReader by reflection.
@@ -37,4 +41,14 @@ public enum RoutingDataReaderType {
public String getClassName() {
return this.className;
}
+
+ public static RoutingDataReaderType lookUp(String enumString) {
+ RoutingDataReaderType type =
+ EnumUtils.getEnumIgnoreCase(RoutingDataReaderType.class, enumString);
+ if (type == null) {
+ throw new MultiZkException(
+ "RoutingDataReaderType::lookUp: Unable to find the enum! String given: " + enumString);
+ }
+ return type;
+ }
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
index 501670c..dbe64f3 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
@@ -19,7 +19,6 @@ package org.apache.helix.zookeeper.impl.client;
* under the License.
*/
-import java.io.IOException;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.TimeUnit;
@@ -69,12 +68,10 @@ public class DedicatedZkClient implements RealmAwareZkClient {
* such as CRUD, change callback, and ephemeral operations for a single ZkRealmShardingKey.
* @param connectionConfig
* @param clientConfig
- * @throws IOException
* @throws InvalidRoutingDataException
*/
public DedicatedZkClient(RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig,
- RealmAwareZkClient.RealmAwareZkClientConfig clientConfig)
- throws IOException, InvalidRoutingDataException {
+ RealmAwareZkClient.RealmAwareZkClientConfig clientConfig) throws InvalidRoutingDataException {
if (connectionConfig == null) {
throw new IllegalArgumentException("RealmAwareZkConnectionConfig cannot be null!");
}
@@ -84,14 +81,15 @@ public class DedicatedZkClient implements RealmAwareZkClient {
_connectionConfig = connectionConfig;
_clientConfig = clientConfig;
- // Get the routing data from a static Singleton HttpRoutingDataReader
- String msdsEndpoint = connectionConfig.getMsdsEndpoint();
- if (msdsEndpoint == null || msdsEndpoint.isEmpty()) {
+ // Get MetadataStoreRoutingData
+ String routingDataSourceEndpoint = connectionConfig.getRoutingDataSourceEndpoint();
+ if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
+ // If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
_metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData();
} else {
- // TODO: Make RoutingDataReaderType configurable
- _metadataStoreRoutingData =
- RoutingDataManager.getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, msdsEndpoint);
+ _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData(
+ RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
+ routingDataSourceEndpoint);
}
_zkRealmShardingKey = connectionConfig.getZkRealmShardingKey();
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
index 22b9fe5..01e7fc7 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
@@ -19,7 +19,6 @@ package org.apache.helix.zookeeper.impl.client;
* under the License.
*/
-import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -87,8 +86,7 @@ public class FederatedZkClient implements RealmAwareZkClient {
// TODO: support capacity of ZkClient number in one FederatedZkClient and do garbage collection.
public FederatedZkClient(RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig,
- RealmAwareZkClient.RealmAwareZkClientConfig clientConfig)
- throws IOException, InvalidRoutingDataException {
+ RealmAwareZkClient.RealmAwareZkClientConfig clientConfig) throws InvalidRoutingDataException {
if (connectionConfig == null) {
throw new IllegalArgumentException("RealmAwareZkConnectionConfig cannot be null!");
}
@@ -96,14 +94,15 @@ public class FederatedZkClient implements RealmAwareZkClient {
throw new IllegalArgumentException("RealmAwareZkClientConfig cannot be null!");
}
- // Attempt to get MetadataStoreRoutingData
- String msdsEndpoint = connectionConfig.getMsdsEndpoint();
- if (msdsEndpoint == null || msdsEndpoint.isEmpty()) {
+ // Get MetadataStoreRoutingData
+ String routingDataSourceEndpoint = connectionConfig.getRoutingDataSourceEndpoint();
+ if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
+ // If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
_metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData();
} else {
- // TODO: Make RoutingDataReaderType configurable
- _metadataStoreRoutingData =
- RoutingDataManager.getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, msdsEndpoint);
+ _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData(
+ RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
+ routingDataSourceEndpoint);
}
_isClosed = false;
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 341731e..7c3a562 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
@@ -19,7 +19,6 @@ package org.apache.helix.zookeeper.impl.client;
* under the License.
*/
-import java.io.IOException;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.TimeUnit;
@@ -65,8 +64,7 @@ public class SharedZkClient implements RealmAwareZkClient {
private final RealmAwareZkClient.RealmAwareZkClientConfig _clientConfig;
public SharedZkClient(RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig,
- RealmAwareZkClient.RealmAwareZkClientConfig clientConfig)
- throws IOException, InvalidRoutingDataException {
+ RealmAwareZkClient.RealmAwareZkClientConfig clientConfig) throws InvalidRoutingDataException {
if (connectionConfig == null) {
throw new IllegalArgumentException("RealmAwareZkConnectionConfig cannot be null!");
}
@@ -76,14 +74,15 @@ public class SharedZkClient implements RealmAwareZkClient {
_connectionConfig = connectionConfig;
_clientConfig = clientConfig;
- // Get the routing data from a static Singleton HttpRoutingDataReader
- String msdsEndpoint = connectionConfig.getMsdsEndpoint();
- if (msdsEndpoint == null || msdsEndpoint.isEmpty()) {
+ // Get MetadataStoreRoutingData
+ String routingDataSourceEndpoint = connectionConfig.getRoutingDataSourceEndpoint();
+ if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
+ // If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
_metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData();
} else {
- // TODO: Make RoutingDataReaderType configurable
- _metadataStoreRoutingData =
- RoutingDataManager.getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, msdsEndpoint);
+ _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData(
+ RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
+ routingDataSourceEndpoint);
}
_zkRealmShardingKey = connectionConfig.getZkRealmShardingKey();
[helix] 08/12: Make RoutingDataManager a pure Singleton with
double-checked locking
Posted by hu...@apache.org.
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
commit f001cde49f5ede583154855865ea0450496d48d0
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Fri Jul 24 15:49:07 2020 -0700
Make RoutingDataManager a pure Singleton with double-checked locking
RoutingDataManager was a stateful static class, but since it contains caches, it would be better to make it a Singleton. This commit makes it a singleton and updates the code accordingly.
---
.../helix/manager/zk/GenericZkHelixApiBuilder.java | 11 +++---
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 5 ++-
.../multizk/TestMultiZkHelixJavaApis.java | 2 +-
.../apache/helix/rest/server/ServerContext.java | 2 +-
.../zookeeper/impl/client/DedicatedZkClient.java | 4 +-
.../zookeeper/impl/client/FederatedZkClient.java | 4 +-
.../zookeeper/impl/client/SharedZkClient.java | 4 +-
.../zookeeper/routing/RoutingDataManager.java | 45 +++++++++++++++-------
.../zookeeper/util/TestRoutingDataManager.java | 10 ++---
9 files changed, 53 insertions(+), 34 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java b/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
index d02f67e..1a4a69b 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
@@ -203,11 +203,12 @@ public abstract class GenericZkHelixApiBuilder<B extends GenericZkHelixApiBuilde
boolean isRoutingDataSourceEndpointSet =
connectionConfig.getRoutingDataSourceEndpoint() != null && !connectionConfig
.getRoutingDataSourceEndpoint().isEmpty();
- MetadataStoreRoutingData routingData = isRoutingDataSourceEndpointSet ? RoutingDataManager
- .getMetadataStoreRoutingData(
- RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
- connectionConfig.getRoutingDataSourceEndpoint())
- : RoutingDataManager.getMetadataStoreRoutingData();
+ MetadataStoreRoutingData routingData =
+ isRoutingDataSourceEndpointSet ? RoutingDataManager.getInstance()
+ .getMetadataStoreRoutingData(
+ RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
+ connectionConfig.getRoutingDataSourceEndpoint())
+ : RoutingDataManager.getInstance().getMetadataStoreRoutingData();
return routingData.getMetadataStoreRealm(connectionConfig.getZkRealmShardingKey());
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
index 1759116..1b3eb8f 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
@@ -969,9 +969,10 @@ public class ZKHelixAdmin implements HelixAdmin {
_zkClient.getRealmAwareZkConnectionConfig().getRoutingDataSourceEndpoint();
if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
// If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
- realmToShardingKeys = RoutingDataManager.getRawRoutingData();
+ realmToShardingKeys = RoutingDataManager.getInstance().getRawRoutingData();
} else {
- realmToShardingKeys = RoutingDataManager.getRawRoutingData(RoutingDataReaderType
+ realmToShardingKeys = RoutingDataManager.getInstance().getRawRoutingData(
+ RoutingDataReaderType
.lookUp(_zkClient.getRealmAwareZkConnectionConfig().getRoutingDataSourceType()),
routingDataSourceEndpoint);
}
diff --git a/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java b/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java
index d72318a..f48a96f 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java
@@ -1039,7 +1039,7 @@ public class TestMultiZkHelixJavaApis {
.setRoutingDataSourceEndpoint(_msdsEndpoint).build();
// Reset cached routing data
- RoutingDataManager.reset();
+ RoutingDataManager.getInstance().reset();
// Shutdown MSDS to ensure that these accessors are able to pull routing data from ZK
_msds.stopServer();
diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java b/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
index 176180f..dc19da7 100644
--- a/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
+++ b/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
@@ -323,7 +323,7 @@ public class ServerContext implements IZkDataListener, IZkChildListener, IZkStat
_zkAddr);
try {
// Reset RoutingDataManager's cache
- RoutingDataManager.reset();
+ RoutingDataManager.getInstance().reset();
// All Helix APIs will be closed implicitly because ZkClient is closed
if (_zkClient != null && !_zkClient.isClosed()) {
_zkClient.close();
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
index dbe64f3..4527fe8 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
@@ -85,9 +85,9 @@ public class DedicatedZkClient implements RealmAwareZkClient {
String routingDataSourceEndpoint = connectionConfig.getRoutingDataSourceEndpoint();
if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
// If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
- _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData();
+ _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData();
} else {
- _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData(
+ _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData(
RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
routingDataSourceEndpoint);
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
index 01e7fc7..11d0291 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
@@ -98,9 +98,9 @@ public class FederatedZkClient implements RealmAwareZkClient {
String routingDataSourceEndpoint = connectionConfig.getRoutingDataSourceEndpoint();
if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
// If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
- _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData();
+ _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData();
} else {
- _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData(
+ _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData(
RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
routingDataSourceEndpoint);
}
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 7c3a562..14a4794 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
@@ -78,9 +78,9 @@ public class SharedZkClient implements RealmAwareZkClient {
String routingDataSourceEndpoint = connectionConfig.getRoutingDataSourceEndpoint();
if (routingDataSourceEndpoint == null || routingDataSourceEndpoint.isEmpty()) {
// If endpoint is not given explicitly, use HTTP and the endpoint set in System Properties
- _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData();
+ _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData();
} else {
- _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData(
+ _metadataStoreRoutingData = RoutingDataManager.getInstance().getMetadataStoreRoutingData(
RoutingDataReaderType.lookUp(connectionConfig.getRoutingDataSourceType()),
routingDataSourceEndpoint);
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
index e489088..bfc12aa 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
@@ -33,28 +33,47 @@ import org.apache.helix.zookeeper.exception.MultiZkException;
/**
- * RoutingDataManager is a static Singleton that
+ * 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 static final String DEFAULT_MSDS_ENDPOINT =
+ private final String DEFAULT_MSDS_ENDPOINT =
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 static final Map<String, Map<String, List<String>>> _rawRoutingDataMap =
+ private final Map<String, Map<String, List<String>>> _rawRoutingDataMap =
new ConcurrentHashMap<>();
// The following map stands for (RoutingDataReaderType_endpoint ID, MetadataStoreRoutingData)
- private static final Map<String, MetadataStoreRoutingData> _metadataStoreRoutingDataMap =
+ private final Map<String, MetadataStoreRoutingData> _metadataStoreRoutingDataMap =
new ConcurrentHashMap<>();
+ // 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;
}
/**
@@ -62,7 +81,7 @@ public class RoutingDataManager {
* config.
* @return
*/
- public static Map<String, List<String>> getRawRoutingData() {
+ public Map<String, List<String>> getRawRoutingData() {
if (DEFAULT_MSDS_ENDPOINT == null || DEFAULT_MSDS_ENDPOINT.isEmpty()) {
throw new IllegalStateException(
"HttpRoutingDataReader was unable to find a valid MSDS endpoint String in System "
@@ -79,8 +98,8 @@ public class RoutingDataManager {
* @param routingDataReaderType
* @param endpoint
*/
- public static Map<String, List<String>> getRawRoutingData(
- RoutingDataReaderType routingDataReaderType, String 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) {
@@ -101,8 +120,7 @@ public class RoutingDataManager {
* MSDS configured in the JVM config.
* @return MetadataStoreRoutingData
*/
- public static MetadataStoreRoutingData getMetadataStoreRoutingData()
- throws InvalidRoutingDataException {
+ public MetadataStoreRoutingData getMetadataStoreRoutingData() throws InvalidRoutingDataException {
if (DEFAULT_MSDS_ENDPOINT == null || DEFAULT_MSDS_ENDPOINT.isEmpty()) {
throw new IllegalStateException(
"HttpRoutingDataReader was unable to find a valid MSDS endpoint String in System "
@@ -119,7 +137,7 @@ public class RoutingDataManager {
* @throws IOException
* @throws InvalidRoutingDataException
*/
- public static MetadataStoreRoutingData getMetadataStoreRoutingData(
+ public MetadataStoreRoutingData getMetadataStoreRoutingData(
RoutingDataReaderType routingDataReaderType, String endpoint)
throws InvalidRoutingDataException {
String routingDataCacheKey = getRoutingDataCacheKey(routingDataReaderType, endpoint);
@@ -141,7 +159,7 @@ public class RoutingDataManager {
/**
* Clears the statically-cached routing data and private fields.
*/
- public synchronized static void reset() {
+ public synchronized void reset() {
_rawRoutingDataMap.clear();
_metadataStoreRoutingDataMap.clear();
}
@@ -151,8 +169,7 @@ public class RoutingDataManager {
* @param routingDataReaderType
* @return
*/
- private static RoutingDataReader resolveRoutingDataReader(
- RoutingDataReaderType routingDataReaderType) {
+ private RoutingDataReader resolveRoutingDataReader(RoutingDataReaderType routingDataReaderType) {
// Instantiate an instance of routing data reader using the type
try {
return (RoutingDataReader) Class.forName(routingDataReaderType.getClassName()).newInstance();
@@ -169,7 +186,7 @@ public class RoutingDataManager {
* @param endpoint
* @return
*/
- private static String getRoutingDataCacheKey(RoutingDataReaderType routingDataReaderType,
+ private String getRoutingDataCacheKey(RoutingDataReaderType routingDataReaderType,
String endpoint) {
if (routingDataReaderType == null) {
throw new MultiZkException("RoutingDataManager: RoutingDataReaderType cannot be null!");
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
index 2fcba96..57b59ee 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
@@ -68,14 +68,14 @@ public class TestRoutingDataManager extends ZkTestBase {
@Test
public void testGetRawRoutingData() {
- Map<String, List<String>> rawRoutingData = RoutingDataManager.getRawRoutingData();
+ Map<String, List<String>> rawRoutingData = RoutingDataManager.getInstance().getRawRoutingData();
TestConstants.FAKE_ROUTING_DATA.forEach((realm, keys) -> Assert
.assertEquals(new HashSet(rawRoutingData.get(realm)), new HashSet(keys)));
}
@Test(dependsOnMethods = "testGetRawRoutingData")
- public void testGetMetadataStoreRoutingData() throws IOException, InvalidRoutingDataException {
- MetadataStoreRoutingData data = RoutingDataManager.getMetadataStoreRoutingData();
+ public void testGetMetadataStoreRoutingData() throws InvalidRoutingDataException {
+ MetadataStoreRoutingData data = RoutingDataManager.getInstance().getMetadataStoreRoutingData();
Map<String, String> allMappings = data.getAllMappingUnderPath("/");
Map<String, Set<String>> groupedMappings = allMappings.entrySet().stream().collect(Collectors
.groupingBy(Map.Entry::getValue,
@@ -103,7 +103,7 @@ public class TestRoutingDataManager extends ZkTestBase {
// HttpRoutingDataReader should still return old data because it's static
// Make sure the results don't contain the new realm
- Map<String, List<String>> rawRoutingData = RoutingDataManager.getRawRoutingData();
+ Map<String, List<String>> rawRoutingData = RoutingDataManager.getInstance().getRawRoutingData();
Assert.assertFalse(rawRoutingData.containsKey(newRealm));
// Remove newRealm and check for equality
@@ -112,7 +112,7 @@ public class TestRoutingDataManager extends ZkTestBase {
TestConstants.FAKE_ROUTING_DATA.forEach((realm, keys) -> Assert
.assertEquals(new HashSet(rawRoutingData.get(realm)), new HashSet(keys)));
- MetadataStoreRoutingData data = RoutingDataManager.getMetadataStoreRoutingData();
+ MetadataStoreRoutingData data = RoutingDataManager.getInstance().getMetadataStoreRoutingData();
Map<String, String> allMappings = data.getAllMappingUnderPath("/");
Map<String, Set<String>> groupedMappings = allMappings.entrySet().stream().collect(Collectors
.groupingBy(Map.Entry::getValue,
[helix] 04/12: Change interface for RoutingDataReader
Posted by hu...@apache.org.
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
commit 3bfc766e863e4a132f61786e09dcf5aad9cb3ea2
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Wed Jul 22 11:08:47 2020 -0700
Change interface for RoutingDataReader
This commit changes the interface for RoutingDataReader in order to accommodate various types of routing data source.
---
.../helix/zookeeper/routing/HttpRoutingDataReader.java | 17 -----------------
.../routing/HttpZkFallbackRoutingDataReader.java | 4 ++++
.../helix/zookeeper/routing/RoutingDataManager.java | 13 ++++++-------
.../helix/zookeeper/routing/RoutingDataReader.java | 12 ++++++++++--
.../helix/zookeeper/routing/ZkRoutingDataReader.java | 4 ++++
.../routing/TestHttpZkFallbackRoutingDataReader.java | 4 ++++
.../zookeeper/routing/TestZkRoutingDataReader.java | 4 ++++
.../helix/zookeeper/util/TestRoutingDataManager.java | 15 ++++++++-------
8 files changed, 40 insertions(+), 33 deletions(-)
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpRoutingDataReader.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpRoutingDataReader.java
index 4ef9881..34f5259 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpRoutingDataReader.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpRoutingDataReader.java
@@ -27,9 +27,6 @@ import java.util.stream.Collectors;
import com.fasterxml.jackson.databind.ObjectMapper;
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.exception.MultiZkException;
import org.apache.http.HttpEntity;
import org.apache.http.client.config.RequestConfig;
@@ -49,20 +46,6 @@ public class HttpRoutingDataReader implements RoutingDataReader {
private static final int DEFAULT_HTTP_TIMEOUT_IN_MS = 5000;
/**
- * Returns an object form of metadata store routing data.
- * @param endpoint
- * @return
- */
- @Override
- public MetadataStoreRoutingData getMetadataStoreRoutingData(String endpoint) {
- try {
- return new TrieRoutingData(getRawRoutingData(endpoint));
- } catch (InvalidRoutingDataException e) {
- throw new MultiZkException(e);
- }
- }
-
- /**
* Returns a map form of metadata store routing data.
* The map fields stand for metadata store realm address (key), and a corresponding list of ZK
* path sharding keys (key).
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpZkFallbackRoutingDataReader.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpZkFallbackRoutingDataReader.java
new file mode 100644
index 0000000..1b4713c
--- /dev/null
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpZkFallbackRoutingDataReader.java
@@ -0,0 +1,4 @@
+package org.apache.helix.zookeeper.routing;
+
+public class HttpZkFallbackRoutingDataReader {
+}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
index aa3986d..e489088 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
@@ -153,17 +153,13 @@ public class RoutingDataManager {
*/
private static RoutingDataReader resolveRoutingDataReader(
RoutingDataReaderType routingDataReaderType) {
- // RoutingDataReaderType.HTTP by default if not found
- routingDataReaderType =
- routingDataReaderType == null ? RoutingDataReaderType.HTTP : 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(
- "RoutingDataManager: failed to instantiate RoutingDataReader! ReaderType className: "
- + routingDataReaderType.getClassName(), e);
+ "RoutingDataManager: failed to instantiate RoutingDataReader! RoutingDataReaderType: "
+ + routingDataReaderType, e);
}
}
@@ -175,6 +171,9 @@ public class RoutingDataManager {
*/
private static String getRoutingDataCacheKey(RoutingDataReaderType routingDataReaderType,
String endpoint) {
+ if (routingDataReaderType == null) {
+ throw new MultiZkException("RoutingDataManager: RoutingDataReaderType cannot be null!");
+ }
return routingDataReaderType.name() + "_" + endpoint;
}
-}
+}
\ No newline at end of file
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataReader.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataReader.java
index 9e6c258..a171855 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataReader.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataReader.java
@@ -23,16 +23,24 @@ import java.util.List;
import java.util.Map;
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.exception.MultiZkException;
public interface RoutingDataReader {
-
/**
* Returns an object form of metadata store routing data.
* @param endpoint
* @return
*/
- MetadataStoreRoutingData getMetadataStoreRoutingData(String endpoint);
+ default MetadataStoreRoutingData getMetadataStoreRoutingData(String endpoint) {
+ try {
+ return new TrieRoutingData(getRawRoutingData(endpoint));
+ } catch (InvalidRoutingDataException e) {
+ throw new MultiZkException(e);
+ }
+ }
/**
* Returns a map form of metadata store routing data.
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/ZkRoutingDataReader.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/ZkRoutingDataReader.java
new file mode 100644
index 0000000..3db3b55
--- /dev/null
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/ZkRoutingDataReader.java
@@ -0,0 +1,4 @@
+package org.apache.helix.zookeeper.routing;
+
+public class ZkRoutingDataReader {
+}
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestHttpZkFallbackRoutingDataReader.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestHttpZkFallbackRoutingDataReader.java
new file mode 100644
index 0000000..85896af
--- /dev/null
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestHttpZkFallbackRoutingDataReader.java
@@ -0,0 +1,4 @@
+package org.apache.helix.zookeeper.routing;
+
+public class TestHttpZkFallbackRoutingDataReader {
+}
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestZkRoutingDataReader.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestZkRoutingDataReader.java
new file mode 100644
index 0000000..68f8df4
--- /dev/null
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestZkRoutingDataReader.java
@@ -0,0 +1,4 @@
+package org.apache.helix.zookeeper.routing;
+
+public class TestZkRoutingDataReader {
+}
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
index e7d0c43..2fcba96 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
@@ -44,20 +44,21 @@ import org.testng.annotations.Test;
public class TestRoutingDataManager extends ZkTestBase {
private MockMetadataStoreDirectoryServer _msdsServer;
- private final String _host = "localhost";
- private final int _port = 1991;
- private final String _namespace = "TestRoutingDataManager";
+ private static final String HOST = "localhost";
+ private static final int PORT = 1991;
+ private static final String NAMESPACE = "TestRoutingDataManager";
+ private static final String MSDS_ENDPOINT =
+ "http://" + HOST + ":" + PORT + "/admin/v2/namespaces/" + NAMESPACE;
@BeforeClass
public void beforeClass() throws IOException {
// Start MockMSDS
- _msdsServer = new MockMetadataStoreDirectoryServer(_host, _port, _namespace,
+ _msdsServer = new MockMetadataStoreDirectoryServer(HOST, PORT, NAMESPACE,
TestConstants.FAKE_ROUTING_DATA);
_msdsServer.startServer();
// Register the endpoint as a System property
- String msdsEndpoint = "http://" + _host + ":" + _port + "/admin/v2/namespaces/" + _namespace;
- System.setProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY, msdsEndpoint);
+ System.setProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY, MSDS_ENDPOINT);
}
@AfterClass
@@ -97,7 +98,7 @@ public class TestRoutingDataManager extends ZkTestBase {
// Kill MSDS and restart with a new mapping
_msdsServer.stopServer();
- _msdsServer = new MockMetadataStoreDirectoryServer(_host, _port, _namespace, newRoutingData);
+ _msdsServer = new MockMetadataStoreDirectoryServer(HOST, PORT, NAMESPACE, newRoutingData);
_msdsServer.startServer();
// HttpRoutingDataReader should still return old data because it's static
[helix] 02/12: Fix string
Posted by hu...@apache.org.
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
commit a686d58e426eeebf0e6b0c790c8de71c7c44a32e
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Mon Jul 20 23:55:16 2020 -0700
Fix string
---
.../apache/helix/zookeeper/constant/RoutingDataReaderType.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java
index 85b8194..aedef36 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java
@@ -19,10 +19,14 @@ package org.apache.helix.zookeeper.constant;
* under the License.
*/
+/**
+ * RoutingDataReaderType is an enum that designates the reader type and the class name that can be
+ * used to create an instance of RoutingDataReader by reflection.
+ */
public enum RoutingDataReaderType {
HTTP("org.apache.helix.zookeeper.routing.HttpRoutingDataReader"),
- ZK("ZkRoutingDataReader"),
- HTTP_ZK_FALLBACK("HttpZkFallbackRoutingDataReader");
+ ZK("org.apache.helix.zookeeper.routing.ZkRoutingDataReader"),
+ HTTP_ZK_FALLBACK("org.apache.helix.zookeeper.routing.HttpZkFallbackRoutingDataReader");
private final String className;
[helix] 12/12: Change UPDATE_ROUTING_DATA_ON_CACHE_MISS and remove
unused imports
Posted by hu...@apache.org.
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
commit 28dc1cfbee22797567a056050e9b9f9b5d680e88
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Thu Aug 13 16:51:10 2020 -0700
Change UPDATE_ROUTING_DATA_ON_CACHE_MISS and remove unused imports
This commit updates the string constant values for UPDATE_ROUTING_DATA_ON_CACHE_MISS so that it's in line with other constants and removes unused imports and unnecessary IOExceptions thrown.
---
.../tools/ClusterVerifiers/ZkHelixClusterVerifier.java | 5 ++---
.../helix/zookeeper/constant/RoutingSystemPropertyKeys.java | 2 +-
.../helix/zookeeper/impl/client/DedicatedZkClient.java | 2 --
.../apache/helix/zookeeper/impl/client/SharedZkClient.java | 2 --
.../zookeeper/impl/factory/DedicatedZkClientFactory.java | 4 +---
.../helix/zookeeper/impl/factory/SharedZkClientFactory.java | 3 +--
.../apache/helix/zookeeper/routing/RoutingDataManager.java | 13 ++++++++++---
7 files changed, 15 insertions(+), 16 deletions(-)
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 98bb835..fe24b3b 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,7 +19,6 @@ 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;
@@ -120,7 +119,7 @@ public abstract class ZkHelixClusterVerifier
new RealmAwareZkClient.RealmAwareZkClientConfig();
_zkClient = DedicatedZkClientFactory.getInstance()
.buildZkClient(connectionConfigBuilder.build(), clientConfig);
- } catch (IOException | InvalidRoutingDataException | IllegalStateException e) {
+ } catch (InvalidRoutingDataException | IllegalStateException e) {
// Note: IllegalStateException is for HttpRoutingDataReader if MSDS endpoint cannot be
// found
throw new HelixException("ZkHelixClusterVerifier: failed to create ZkClient!", e);
@@ -375,7 +374,7 @@ public abstract class ZkHelixClusterVerifier
// First, try to create a RealmAwareZkClient that's a DedicatedZkClient
return DedicatedZkClientFactory.getInstance()
.buildZkClient(connectionConfig, clientConfig);
- } catch (IOException | InvalidRoutingDataException | IllegalStateException e) {
+ } catch (InvalidRoutingDataException | IllegalStateException e) {
throw new HelixException("ZkHelixClusterVerifier: failed to create ZkClient!", e);
}
} else {
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
index e22ad08..12fd023 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
@@ -29,7 +29,7 @@ public class RoutingSystemPropertyKeys {
* re-read the routing data from the routing data source upon ZK path sharding key cache miss.
*/
public static final String UPDATE_ROUTING_DATA_ON_CACHE_MISS =
- "update.routing.data.on.cache.miss.enabled";
+ "routing.data.update.on.cache.miss.enabled";
/**
* The interval to use between routing data updates from the routing data source.
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
index cc37853..74ff0ad 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
@@ -27,8 +27,6 @@ import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.api.client.ChildrenSubscribeResult;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
-import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
-import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
import org.apache.helix.zookeeper.zkclient.IZkConnection;
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 bcc4b12..7f529d2 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
@@ -28,9 +28,7 @@ import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.api.client.ChildrenSubscribeResult;
import org.apache.helix.zookeeper.api.client.HelixZkClient;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
-import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
-import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/DedicatedZkClientFactory.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/DedicatedZkClientFactory.java
index 2417584..bbccd22 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/DedicatedZkClientFactory.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/DedicatedZkClientFactory.java
@@ -19,8 +19,6 @@ package org.apache.helix.zookeeper.impl.factory;
* under the License.
*/
-import java.io.IOException;
-
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.api.client.HelixZkClient;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
@@ -40,7 +38,7 @@ public class DedicatedZkClientFactory extends HelixZkClientFactory {
public RealmAwareZkClient buildZkClient(
RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig,
RealmAwareZkClient.RealmAwareZkClientConfig clientConfig)
- throws IOException, InvalidRoutingDataException {
+ throws InvalidRoutingDataException {
return new DedicatedZkClient(connectionConfig, clientConfig);
}
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 8cb8f33..e3cbf5d 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
@@ -19,7 +19,6 @@ package org.apache.helix.zookeeper.impl.factory;
* under the License.
*/
-import java.io.IOException;
import java.util.HashMap;
import java.util.List;
@@ -60,7 +59,7 @@ public class SharedZkClientFactory extends HelixZkClientFactory {
public RealmAwareZkClient buildZkClient(
RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig,
RealmAwareZkClient.RealmAwareZkClientConfig clientConfig)
- throws IOException, InvalidRoutingDataException {
+ throws InvalidRoutingDataException {
// Note, the logic sharing connectionManager logic is inside SharedZkClient, similar to innerSharedZkClient.
return new SharedZkClient(connectionConfig, clientConfig);
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
index 853bd5c..2afd62c 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
@@ -30,6 +30,9 @@ 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;
+import org.apache.helix.zookeeper.impl.client.SharedZkClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
@@ -39,6 +42,8 @@ import org.apache.helix.zookeeper.exception.MultiZkException;
* 3. provides public methods for reading routing data from various sources (configurable)
*/
public class RoutingDataManager {
+ private static Logger LOG = LoggerFactory.getLogger(RoutingDataManager.class);
+
/** HTTP call to MSDS is used to fetch routing data by default */
private String _defaultMsdsEndpoint =
System.getProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY);
@@ -188,9 +193,11 @@ public class RoutingDataManager {
try {
return (RoutingDataReader) Class.forName(routingDataReaderType.getClassName()).newInstance();
} catch (ClassNotFoundException | IllegalAccessException | InstantiationException e) {
- throw new MultiZkException(
- "RoutingDataManager: failed to instantiate RoutingDataReader! RoutingDataReaderType: "
- + routingDataReaderType, e);
+ String errMsg =
+ "RoutingDataManager::resolveRoutingDataReader: failed to instantiate RoutingDataReader! "
+ + "RoutingDataReaderType: " + routingDataReaderType;
+ LOG.error(errMsg, e);
+ throw new MultiZkException(errMsg, e);
}
}
[helix] 10/12: Implement routing data update upon cache miss for
FederatedZkClient
Posted by hu...@apache.org.
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
commit 59cd6740a71764f3aad336cabb55460365b914dd
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Sat Jul 25 23:02:05 2020 -0700
Implement routing data update upon cache miss for FederatedZkClient
FederatedZkClient is the only implementation of RealmAwareZkClient that is capable of accesing multiple ZKs. There are potential use cases where it would be beneficial for FederatedZkClient and Helix Java APIs on multi-ZK mode to automatically trigger a routing data update when the ZK path sharding key is not found in the cached routing data.
This commit implements the feature and allows users to turn it on by setting a field in System Properties.
---
.../multizk/TestMultiZkHelixJavaApis.java | 4 +-
.../zookeeper/impl/client/FederatedZkClient.java | 71 +++++-
.../zookeeper/routing/RoutingDataManager.java | 12 +-
.../impl/client/RealmAwareZkClientTestBase.java | 8 +-
.../impl/client/TestFederatedZkClient.java | 255 ++++++++++++++++++++-
.../zookeeper/util/TestRoutingDataManager.java | 3 +
6 files changed, 325 insertions(+), 28 deletions(-)
diff --git a/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java b/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java
index f48a96f..497de17 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/multizk/TestMultiZkHelixJavaApis.java
@@ -970,8 +970,8 @@ public class TestMultiZkHelixJavaApis {
if (shouldSucceed) {
Assert.fail("ZK Realm should be found for /" + cluster);
} else {
- Assert.assertTrue(
- e.getMessage().startsWith("Cannot find ZK realm for the path: /" + cluster));
+ Assert.assertTrue(e.getMessage()
+ .startsWith("No sharding key found within the provided path. Path: /" + cluster));
}
} catch (IllegalArgumentException e) {
if (shouldSucceed) {
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
index 679d4e7..4354537 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
@@ -30,7 +30,8 @@ import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.api.client.ChildrenSubscribeResult;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
-import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
+import org.apache.helix.zookeeper.constant.RoutingSystemPropertyKeys;
+import org.apache.helix.zookeeper.exception.MultiZkException;
import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
@@ -74,7 +75,7 @@ public class FederatedZkClient implements RealmAwareZkClient {
private static final String DEDICATED_ZK_CLIENT_FACTORY =
DedicatedZkClientFactory.class.getSimpleName();
- private final MetadataStoreRoutingData _metadataStoreRoutingData;
+ private volatile MetadataStoreRoutingData _metadataStoreRoutingData;
private final RealmAwareZkClient.RealmAwareZkConnectionConfig _connectionConfig;
private final RealmAwareZkClient.RealmAwareZkClientConfig _clientConfig;
@@ -83,6 +84,8 @@ public class FederatedZkClient implements RealmAwareZkClient {
private volatile boolean _isClosed;
private PathBasedZkSerializer _pathBasedZkSerializer;
+ private final boolean _routingDataUpdateOnCacheMissEnabled = Boolean.parseBoolean(
+ System.getProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS));
// TODO: support capacity of ZkClient number in one FederatedZkClient and do garbage collection.
public FederatedZkClient(RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig,
@@ -509,7 +512,6 @@ public class FederatedZkClient implements RealmAwareZkClient {
checkClosedState();
String zkRealm = getZkRealm(path);
-
// Use this zkClient reference to protect the returning zkClient from being null because of
// race condition. Once we get the reference, even _zkRealmToZkClientMap is cleared by closed(),
// this zkClient is not null which guarantees the returned value not null.
@@ -541,17 +543,66 @@ public class FederatedZkClient implements RealmAwareZkClient {
}
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) {
+ // 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();
+ _metadataStoreRoutingData =
+ RealmAwareZkClient.getMetadataStoreRoutingData(_connectionConfig);
+ // No try-catch for the following call because if this throws a
+ // NoSuchElementException, it means the ZK path sharding key doesn't exist even
+ // after a full cache refresh
+ zkRealm = _metadataStoreRoutingData.getMetadataStoreRealm(path);
+ }
+ }
+ }
+ }
+ }
}
-
return zkRealm;
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
index bfc12aa..6df9616 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
@@ -40,7 +40,7 @@ import org.apache.helix.zookeeper.exception.MultiZkException;
*/
public class RoutingDataManager {
/** HTTP call to MSDS is used to fetch routing data by default */
- private final String DEFAULT_MSDS_ENDPOINT =
+ private String _defaultMsdsEndpoint =
System.getProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY);
/** Double-checked locking requires that the following fields be final (volatile) */
@@ -82,12 +82,12 @@ public class RoutingDataManager {
* @return
*/
public Map<String, List<String>> getRawRoutingData() {
- if (DEFAULT_MSDS_ENDPOINT == null || DEFAULT_MSDS_ENDPOINT.isEmpty()) {
+ 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, DEFAULT_MSDS_ENDPOINT);
+ return getRawRoutingData(RoutingDataReaderType.HTTP, _defaultMsdsEndpoint);
}
/**
@@ -121,12 +121,12 @@ public class RoutingDataManager {
* @return MetadataStoreRoutingData
*/
public MetadataStoreRoutingData getMetadataStoreRoutingData() throws InvalidRoutingDataException {
- if (DEFAULT_MSDS_ENDPOINT == null || DEFAULT_MSDS_ENDPOINT.isEmpty()) {
+ 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, DEFAULT_MSDS_ENDPOINT);
+ return getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, _defaultMsdsEndpoint);
}
/**
@@ -162,6 +162,8 @@ public class RoutingDataManager {
public synchronized void reset() {
_rawRoutingDataMap.clear();
_metadataStoreRoutingDataMap.clear();
+ _defaultMsdsEndpoint =
+ System.getProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY);
}
/**
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 acb2299..1ff56bd 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
@@ -36,10 +36,10 @@ public abstract class RealmAwareZkClientTestBase extends ZkTestBase {
protected static final String TEST_INVALID_PATH = ZK_SHARDING_KEY_PREFIX + "_invalid" + "/a/b/c";
// Create a MockMSDS for testing
- private static MockMetadataStoreDirectoryServer _msdsServer;
- private static final String MSDS_HOSTNAME = "localhost";
- private static final int MSDS_PORT = 1111;
- private static final String MSDS_NAMESPACE = "test";
+ protected static MockMetadataStoreDirectoryServer _msdsServer;
+ protected static final String MSDS_HOSTNAME = "localhost";
+ protected static final int MSDS_PORT = 19910;
+ protected static final String MSDS_NAMESPACE = "test";
@BeforeClass
public void beforeClass() throws IOException, InvalidRoutingDataException {
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java
index 0472df7..93e5892 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java
@@ -20,15 +20,26 @@ package org.apache.helix.zookeeper.impl.client;
*/
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.TimeUnit;
+import org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
+import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
+import org.apache.helix.msdcommon.mock.MockMetadataStoreDirectoryServer;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
+import org.apache.helix.zookeeper.constant.RoutingSystemPropertyKeys;
+import org.apache.helix.zookeeper.constant.TestConstants;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.IZkStateListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Op;
@@ -179,8 +190,8 @@ public class TestFederatedZkClient extends RealmAwareZkClientTestBase {
_realmAwareZkClient.createPersistent(TEST_INVALID_PATH, true);
Assert.fail("Create() should not succeed on an invalid path!");
} catch (NoSuchElementException ex) {
- Assert
- .assertEquals(ex.getMessage(), "Cannot find ZK realm for the path: " + TEST_INVALID_PATH);
+ Assert.assertEquals(ex.getMessage(),
+ "No sharding key found within the provided path. Path: " + TEST_INVALID_PATH);
}
}
@@ -195,8 +206,8 @@ public class TestFederatedZkClient extends RealmAwareZkClientTestBase {
_realmAwareZkClient.exists(TEST_INVALID_PATH);
Assert.fail("Exists() should not succeed on an invalid path!");
} catch (NoSuchElementException ex) {
- Assert
- .assertEquals(ex.getMessage(), "Cannot find ZK realm for the path: " + TEST_INVALID_PATH);
+ Assert.assertEquals(ex.getMessage(),
+ "No sharding key found within the provided path. Path: " + TEST_INVALID_PATH);
}
}
@@ -209,8 +220,8 @@ public class TestFederatedZkClient extends RealmAwareZkClientTestBase {
_realmAwareZkClient.delete(TEST_INVALID_PATH);
Assert.fail("Exists() should not succeed on an invalid path!");
} catch (NoSuchElementException ex) {
- Assert
- .assertEquals(ex.getMessage(), "Cannot find ZK realm for the path: " + TEST_INVALID_PATH);
+ Assert.assertEquals(ex.getMessage(),
+ "No sharding key found within the provided path. Path: " + TEST_INVALID_PATH);
}
Assert.assertTrue(_realmAwareZkClient.delete(TEST_REALM_ONE_VALID_PATH));
@@ -258,12 +269,242 @@ public class TestFederatedZkClient extends RealmAwareZkClientTestBase {
Assert.assertFalse(_realmAwareZkClient.exists(TEST_REALM_TWO_VALID_PATH));
}
+ /**
+ * This tests the routing data update feature only enabled when
+ * RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS is set to true.
+ * Routing data source is MSDS.
+ */
+ @Test(dependsOnMethods = "testMultiRealmCRUD")
+ public void testUpdateRoutingDataOnCacheMissMSDS()
+ throws IOException, InvalidRoutingDataException {
+ // Enable routing data update upon cache miss
+ System.setProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS, "true");
+
+ RoutingDataManager.getInstance().getMetadataStoreRoutingData();
+ _msdsServer.stopServer();
+ /*
+ * Test is 2-tiered because cache update is 2-tiered
+ * Case 1:
+ * - RoutingDataManager (in-memory) does not have the key
+ * - MSDS has the key
+ * This simulates a case where FederatedZkClient must do a I/O based update.
+ */
+ // Start MSDS with a new key
+ String newShardingKey = "/sharding-key-9";
+ String zkRealm = "localhost:2127";
+ Map<String, Collection<String>> rawRoutingData = new HashMap<>();
+ rawRoutingData.put(zkRealm, new ArrayList<>());
+ rawRoutingData.get(zkRealm).add(newShardingKey); // Add a new key
+ _msdsServer = new MockMetadataStoreDirectoryServer(MSDS_HOSTNAME, MSDS_PORT, MSDS_NAMESPACE,
+ rawRoutingData);
+ _msdsServer.startServer();
+
+ // Verify that RoutingDataManager does not have the key
+ MetadataStoreRoutingData routingData =
+ RoutingDataManager.getInstance().getMetadataStoreRoutingData();
+ try {
+ routingData.getMetadataStoreRealm(newShardingKey);
+ Assert.fail("Must throw NoSuchElementException!");
+ } catch (NoSuchElementException e) {
+ // Expected
+ }
+
+ // Create a new FederatedZkClient
+ FederatedZkClient federatedZkClient = new FederatedZkClient(
+ new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder()
+ .setRoutingDataSourceType(RoutingDataReaderType.HTTP.name())
+ .setRoutingDataSourceEndpoint(
+ "http://" + MSDS_HOSTNAME + ":" + MSDS_PORT + "/admin/v2/namespaces/"
+ + MSDS_NAMESPACE).build(), new RealmAwareZkClient.RealmAwareZkClientConfig());
+
+ // exists() must succeed and RoutingDataManager should now have the key (cache update must have
+ // happened)
+ // False expected for the following call because the znode does not exist and we are checking
+ // whether the call succeeds or not
+ Assert.assertFalse(federatedZkClient.exists(newShardingKey));
+ Assert.assertEquals(zkRealm, RoutingDataManager.getInstance().getMetadataStoreRoutingData()
+ .getMetadataStoreRealm(newShardingKey));
+
+ /*
+ * Case 2:
+ * - RoutingDataManager has the key
+ * - MSDS does not have the key
+ * - continue using the same ZkClient because we want an existing federated client that does
+ * not have the key
+ */
+ _msdsServer.stopServer();
+ // Create an MSDS with the key and reset MSDS so it doesn't contain the key
+ String newShardingKey2 = "/sharding-key-10";
+ rawRoutingData.get(zkRealm).add(newShardingKey2);
+ _msdsServer = new MockMetadataStoreDirectoryServer(MSDS_HOSTNAME, MSDS_PORT, MSDS_NAMESPACE,
+ rawRoutingData);
+ _msdsServer.startServer();
+
+ // Make sure RoutingDataManager has the key
+ RoutingDataManager.getInstance().reset();
+ Assert.assertEquals(zkRealm, RoutingDataManager.getInstance().getMetadataStoreRoutingData()
+ .getMetadataStoreRealm(newShardingKey2));
+
+ // Reset MSDS so it doesn't contain the key
+ _msdsServer.stopServer();
+ _msdsServer = new MockMetadataStoreDirectoryServer(MSDS_HOSTNAME, MSDS_PORT, MSDS_NAMESPACE,
+ TestConstants.FAKE_ROUTING_DATA); // FAKE_ROUTING_DATA doesn't contain the key
+ _msdsServer.startServer();
+
+ // exists() must succeed and RoutingDataManager should still have the key
+ // This means that we do not do a hard update (I/O based update) because in-memory cache already
+ // has the key
+ // False expected for the following call because the znode does not exist and we are checking
+ // whether the call succeeds or not
+ Assert.assertFalse(federatedZkClient.exists(newShardingKey2));
+ Assert.assertEquals(zkRealm, RoutingDataManager.getInstance().getMetadataStoreRoutingData()
+ .getMetadataStoreRealm(newShardingKey2));
+ // Also check that MSDS does not have the new sharding key through resetting RoutingDataManager
+ // and re-reading from MSDS
+ RoutingDataManager.getInstance().reset();
+ try {
+ RoutingDataManager.getInstance().getMetadataStoreRoutingData()
+ .getMetadataStoreRealm(newShardingKey2);
+ Assert.fail("NoSuchElementException expected!");
+ } catch (NoSuchElementException e) {
+ // Expected because MSDS does not contain the key
+ }
+
+ // Clean up federatedZkClient
+ federatedZkClient.close();
+ // Shut down MSDS
+ _msdsServer.stopServer();
+ // Disable System property
+ System.setProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS, "false");
+ }
+
+ /**
+ * This tests the routing data update feature only enabled when
+ * RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS is set to true.
+ * Routing data source is ZK.
+ */
+ @Test(dependsOnMethods = "testUpdateRoutingDataOnCacheMissMSDS")
+ public void testUpdateRoutingDataOnCacheMissZK() throws IOException, InvalidRoutingDataException {
+ // Set up routing data in ZK with empty sharding key list
+ String zkRealm = "localhost:2127";
+ String newShardingKey = "/sharding-key-9";
+ String newShardingKey2 = "/sharding-key-10";
+ ZkClient zkClient =
+ new ZkClient.Builder().setZkServer(zkRealm).setZkSerializer(new ZNRecordSerializer())
+ .build();
+ zkClient.create(MetadataStoreRoutingConstants.ROUTING_DATA_PATH, null, CreateMode.PERSISTENT);
+ ZNRecord zkRealmRecord = new ZNRecord(zkRealm);
+ List<String> keyList =
+ new ArrayList<>(TestConstants.TEST_KEY_LIST_1); // Need a non-empty keyList
+ zkRealmRecord.setListField(MetadataStoreRoutingConstants.ZNRECORD_LIST_FIELD_KEY, keyList);
+ zkClient.create(MetadataStoreRoutingConstants.ROUTING_DATA_PATH + "/" + zkRealm, zkRealmRecord,
+ CreateMode.PERSISTENT);
+
+ // Enable routing data update upon cache miss
+ System.setProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS, "true");
+
+ RoutingDataManager.getInstance().reset();
+ RoutingDataManager.getInstance().getMetadataStoreRoutingData(RoutingDataReaderType.ZK, zkRealm);
+ /*
+ * Test is 2-tiered because cache update is 2-tiered
+ * Case 1:
+ * - RoutingDataManager does not have the key
+ * - ZK has the key
+ * This simulates a case where FederatedZkClient must do a I/O based update (must read from ZK).
+ */
+ // Add the key to ZK
+ zkRealmRecord.getListField(MetadataStoreRoutingConstants.ZNRECORD_LIST_FIELD_KEY)
+ .add(newShardingKey);
+ zkClient
+ .writeData(MetadataStoreRoutingConstants.ROUTING_DATA_PATH + "/" + zkRealm, zkRealmRecord);
+
+ // Verify that RoutingDataManager does not have the key
+ MetadataStoreRoutingData routingData = RoutingDataManager.getInstance()
+ .getMetadataStoreRoutingData(RoutingDataReaderType.ZK, zkRealm);
+ try {
+ routingData.getMetadataStoreRealm(newShardingKey);
+ Assert.fail("Must throw NoSuchElementException!");
+ } catch (NoSuchElementException e) {
+ // Expected
+ }
+
+ // Create a new FederatedZkClient
+ FederatedZkClient federatedZkClient = new FederatedZkClient(
+ new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder()
+ .setRoutingDataSourceType(RoutingDataReaderType.ZK.name())
+ .setRoutingDataSourceEndpoint(zkRealm).build(),
+ new RealmAwareZkClient.RealmAwareZkClientConfig());
+
+ // exists() must succeed and RoutingDataManager should now have the key (cache update must
+ // have happened)
+ // False expected for the following call because the znode does not exist and we are checking
+ // whether the call succeeds or not
+ Assert.assertFalse(federatedZkClient.exists(newShardingKey));
+ Assert.assertEquals(zkRealm, RoutingDataManager.getInstance()
+ .getMetadataStoreRoutingData(RoutingDataReaderType.ZK, zkRealm)
+ .getMetadataStoreRealm(newShardingKey));
+
+ /*
+ * Case 2:
+ * - RoutingDataManager has the key
+ * - ZK does not have the key
+ * - continue using the same ZkClient because we want an existing federated client that does
+ * not have the key
+ */
+ // Add newShardingKey2 to ZK's routing data (in order to give RoutingDataManager the key)
+ zkRealmRecord.getListField(MetadataStoreRoutingConstants.ZNRECORD_LIST_FIELD_KEY)
+ .add(newShardingKey2);
+ zkClient
+ .writeData(MetadataStoreRoutingConstants.ROUTING_DATA_PATH + "/" + zkRealm, zkRealmRecord);
+
+ // Update RoutingDataManager so it has the key
+ RoutingDataManager.getInstance().reset();
+ Assert.assertEquals(zkRealm, RoutingDataManager.getInstance()
+ .getMetadataStoreRoutingData(RoutingDataReaderType.ZK, zkRealm)
+ .getMetadataStoreRealm(newShardingKey2));
+
+ // Remove newShardingKey2 from ZK
+ zkRealmRecord.getListField(MetadataStoreRoutingConstants.ZNRECORD_LIST_FIELD_KEY)
+ .remove(newShardingKey2);
+ zkClient
+ .writeData(MetadataStoreRoutingConstants.ROUTING_DATA_PATH + "/" + zkRealm, zkRealmRecord);
+
+ // exists() must succeed and RoutingDataManager should still have the key
+ // This means that we do not do a hard update (I/O based update) because in-memory cache already
+ // has the key
+ // False expected for the following call because the znode does not exist and we are checking
+ // whether the call succeeds or not
+ Assert.assertFalse(federatedZkClient.exists(newShardingKey2));
+ Assert.assertEquals(zkRealm, RoutingDataManager.getInstance()
+ .getMetadataStoreRoutingData(RoutingDataReaderType.ZK, zkRealm)
+ .getMetadataStoreRealm(newShardingKey2));
+ // Also check that ZK does not have the new sharding key through resetting RoutingDataManager
+ // and re-reading from ZK
+ RoutingDataManager.getInstance().reset();
+ try {
+ RoutingDataManager.getInstance()
+ .getMetadataStoreRoutingData(RoutingDataReaderType.ZK, zkRealm)
+ .getMetadataStoreRealm(newShardingKey2);
+ Assert.fail("NoSuchElementException expected!");
+ } catch (NoSuchElementException e) {
+ // Expected because ZK does not contain the key
+ }
+
+ // Clean up federatedZkClient
+ federatedZkClient.close();
+ // Clean up ZK writes and ZkClient
+ zkClient.deleteRecursively(MetadataStoreRoutingConstants.ROUTING_DATA_PATH);
+ zkClient.close();
+ // Disable System property
+ System.setProperty(RoutingSystemPropertyKeys.UPDATE_ROUTING_DATA_ON_CACHE_MISS, "false");
+ }
+
/*
* Tests that close() works.
* TODO: test that all raw zkClients are closed after FederatedZkClient close() is called. This
* could help avoid ZkClient leakage.
*/
- @Test(dependsOnMethods = "testMultiRealmCRUD")
+ @Test(dependsOnMethods = "testUpdateRoutingDataOnCacheMissZK")
public void testClose() {
Assert.assertFalse(_realmAwareZkClient.isClosed());
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
index 57b59ee..19ced01 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
@@ -59,6 +59,9 @@ public class TestRoutingDataManager extends ZkTestBase {
// Register the endpoint as a System property
System.setProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY, MSDS_ENDPOINT);
+
+ // Reset RoutingDataManager
+ RoutingDataManager.getInstance().reset();
}
@AfterClass
[helix] 06/12: Implement ZkRoutingDataReader
Posted by hu...@apache.org.
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
commit d3477bed870ba7778d55ed49ea14e30a09c88cdc
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Wed Jul 22 11:10:42 2020 -0700
Implement ZkRoutingDataReader
In order to allow certain users to use ZK as the sole routing data source, we add ZkRoutingDataReader that transiently creates a ZK connection to read the routing data from the routing ZK.
---
.../zookeeper/routing/ZkRoutingDataReader.java | 81 +++++++++++++++++++++-
.../helix/zookeeper/constant/TestConstants.java | 16 +++--
.../zookeeper/routing/TestZkRoutingDataReader.java | 75 +++++++++++++++++++-
3 files changed, 164 insertions(+), 8 deletions(-)
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/ZkRoutingDataReader.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/ZkRoutingDataReader.java
index 3db3b55..48c8604 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/ZkRoutingDataReader.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/ZkRoutingDataReader.java
@@ -1,4 +1,83 @@
package org.apache.helix.zookeeper.routing;
-public class ZkRoutingDataReader {
+/*
+ * 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.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.exception.MultiZkException;
+import org.apache.helix.zookeeper.impl.client.ZkClient;
+import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
+
+
+/**
+ * Zk-based RoutingDataReader that establishes a ZK connection to the routing ZK to fetch routing
+ * data.
+ * The reading of routing data by nature should only be performed in cases of a Helix client
+ * initialization or routing data reset. That means we do not have to maintain an active ZK
+ * connection. To minimize the number of client-side ZK connections, ZkRoutingDataReader establishes
+ * a ZK session temporarily only to read from ZK afresh and closes sessions upon read completion.
+ */
+public class ZkRoutingDataReader implements RoutingDataReader {
+
+ /**
+ * Returns a map form of metadata store routing data.
+ * The map fields stand for metadata store realm address (key), and a corresponding list of ZK
+ * path sharding keys (key).
+ * @param endpoint
+ * @return
+ */
+ @Override
+ public Map<String, List<String>> getRawRoutingData(String endpoint) {
+ ZkClient zkClient =
+ new ZkClient.Builder().setZkServer(endpoint).setZkSerializer(new ZNRecordSerializer())
+ .build();
+
+ Map<String, List<String>> routingData = new HashMap<>();
+ List<String> allRealmAddresses;
+ try {
+ allRealmAddresses = zkClient.getChildren(MetadataStoreRoutingConstants.ROUTING_DATA_PATH);
+ } catch (ZkNoNodeException e) {
+ throw new MultiZkException(
+ "Routing data directory ZNode " + MetadataStoreRoutingConstants.ROUTING_DATA_PATH
+ + " does not exist. Routing ZooKeeper address: " + endpoint);
+ }
+ if (allRealmAddresses != null) {
+ for (String realmAddress : allRealmAddresses) {
+ ZNRecord record = zkClient
+ .readData(MetadataStoreRoutingConstants.ROUTING_DATA_PATH + "/" + realmAddress, true);
+ if (record != null) {
+ List<String> shardingKeys =
+ record.getListField(MetadataStoreRoutingConstants.ZNRECORD_LIST_FIELD_KEY);
+ routingData
+ .put(realmAddress, shardingKeys != null ? shardingKeys : Collections.emptyList());
+ }
+ }
+ }
+
+ zkClient.close();
+ return routingData;
+ }
}
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/constant/TestConstants.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/constant/TestConstants.java
index a217245..d6c6713 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/constant/TestConstants.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/constant/TestConstants.java
@@ -20,6 +20,7 @@ package org.apache.helix.zookeeper.constant;
*/
import java.util.Collection;
+import java.util.List;
import java.util.Map;
import com.google.common.collect.ImmutableList;
@@ -34,12 +35,15 @@ public class TestConstants {
public static final String ZK_PREFIX = "localhost:";
public static final int ZK_START_PORT = 2127;
+ public static final List<String> TEST_KEY_LIST_1 =
+ ImmutableList.of("/sharding-key-0", "/sharding-key-1", "/sharding-key-2");
+ public static final List<String> TEST_KEY_LIST_2 =
+ ImmutableList.of("/sharding-key-3", "/sharding-key-4", "/sharding-key-5");
+ public static final List<String> TEST_KEY_LIST_3 =
+ ImmutableList.of("/sharding-key-6", "/sharding-key-7", "/sharding-key-8");
+
// Based on the ZK hostname constants, construct a set of fake routing data mappings
public static final Map<String, Collection<String>> FAKE_ROUTING_DATA = ImmutableMap
- .of(ZK_PREFIX + ZK_START_PORT,
- ImmutableList.of("/sharding-key-0", "/sharding-key-1", "/sharding-key-2"),
- ZK_PREFIX + (ZK_START_PORT + 1),
- ImmutableList.of("/sharding-key-3", "/sharding-key-4", "/sharding-key-5"),
- ZK_PREFIX + (ZK_START_PORT + 2),
- ImmutableList.of("/sharding-key-6", "/sharding-key-7", "/sharding-key-8"));
+ .of(ZK_PREFIX + ZK_START_PORT, TEST_KEY_LIST_1, ZK_PREFIX + (ZK_START_PORT + 1),
+ TEST_KEY_LIST_2, ZK_PREFIX + (ZK_START_PORT + 2), TEST_KEY_LIST_3);
}
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestZkRoutingDataReader.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestZkRoutingDataReader.java
index 68f8df4..cc1936b 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestZkRoutingDataReader.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestZkRoutingDataReader.java
@@ -1,4 +1,77 @@
package org.apache.helix.zookeeper.routing;
-public class TestZkRoutingDataReader {
+/*
+ * 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.util.List;
+import java.util.Map;
+
+import org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
+import org.apache.helix.zookeeper.constant.TestConstants;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.ZkTestBase;
+import org.apache.helix.zookeeper.zkclient.ZkClient;
+import org.apache.helix.zookeeper.zkclient.ZkServer;
+import org.apache.zookeeper.CreateMode;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class TestZkRoutingDataReader extends ZkTestBase {
+ private static final String ROUTING_ZK_ADDR = "localhost:2358";
+ private ZkServer _routingZk;
+
+ @BeforeClass
+ public void beforeClass() {
+ // Start a separate ZK for isolation
+ _routingZk = startZkServer(ROUTING_ZK_ADDR);
+
+ // Create ZK realm routing data ZNRecord
+ ZNRecord znRecord = new ZNRecord(ROUTING_ZK_ADDR);
+ znRecord.setListField(MetadataStoreRoutingConstants.ZNRECORD_LIST_FIELD_KEY,
+ TestConstants.TEST_KEY_LIST_1);
+
+ // Write raw routing data to ZK
+ ZkClient zkClient = _routingZk.getZkClient();
+ zkClient.setZkSerializer(new ZNRecordSerializer());
+ zkClient.create(MetadataStoreRoutingConstants.ROUTING_DATA_PATH, null, CreateMode.PERSISTENT);
+ zkClient
+ .create(MetadataStoreRoutingConstants.ROUTING_DATA_PATH + "/" + ROUTING_ZK_ADDR, znRecord,
+ CreateMode.PERSISTENT);
+ }
+
+ @AfterClass
+ public void afterClass() {
+ _routingZk.shutdown();
+ }
+
+ @Test
+ public void testGetRawRoutingData() {
+ Map<String, List<String>> rawRoutingData =
+ new ZkRoutingDataReader().getRawRoutingData(ROUTING_ZK_ADDR);
+
+ // Check that the returned content matches up with what we expect (1 realm, 3 keys)
+ Assert.assertEquals(rawRoutingData.size(), 1);
+ Assert.assertTrue(rawRoutingData.containsKey(ROUTING_ZK_ADDR));
+ Assert.assertEquals(rawRoutingData.get(ROUTING_ZK_ADDR), TestConstants.TEST_KEY_LIST_1);
+ }
}
[helix] 01/12: Implement RoutingDataManager to replace
HttpRoutingDataReader
Posted by hu...@apache.org.
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
commit 5e85638aa8b2ef8c9f8f17da88b8a3cfdaa37534
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Mon Jul 20 23:46:15 2020 -0700
Implement RoutingDataManager to replace HttpRoutingDataReader
We want to make the routing data source configurable. As such, using HttpRoutingDataReader as the static Singleton that fetches and caches the routing data is no longer appropriate. This change adds an implementation of RoutingDataManager and the new RoutingDataReader interface, with HttpRoutingDataReader as one of its implementations. This is the first step towards making routing data source configurable - other readers will be added in the future commits.
---
.../helix/manager/zk/GenericZkHelixApiBuilder.java | 10 +-
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 9 +-
.../apache/helix/rest/server/ServerContext.java | 6 +-
.../zookeeper/api/client/RealmAwareZkClient.java | 4 -
.../zookeeper/constant/RoutingDataReaderType.java | 36 ++++
.../constant/RoutingSystemPropertyKeys.java | 31 ++++
.../zookeeper/exception/MultiZkException.java | 37 ++++
.../zookeeper/impl/client/DedicatedZkClient.java | 9 +-
.../zookeeper/impl/client/FederatedZkClient.java | 9 +-
.../zookeeper/impl/client/SharedZkClient.java | 9 +-
.../zookeeper/routing/HttpRoutingDataReader.java | 132 ++++++++++++++
.../zookeeper/routing/RoutingDataManager.java | 181 +++++++++++++++++++
.../helix/zookeeper/routing/RoutingDataReader.java | 45 +++++
.../zookeeper/util/HttpRoutingDataReader.java | 197 ---------------------
...DataReader.java => TestRoutingDataManager.java} | 13 +-
15 files changed, 502 insertions(+), 226 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java b/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
index c7480c3..840ec8f 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/GenericZkHelixApiBuilder.java
@@ -26,10 +26,11 @@ import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
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.constant.RoutingDataReaderType;
import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.helix.zookeeper.impl.client.FederatedZkClient;
import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
-import org.apache.helix.zookeeper.util.HttpRoutingDataReader;
+import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -204,9 +205,10 @@ public abstract class GenericZkHelixApiBuilder<B extends GenericZkHelixApiBuilde
throws IOException, InvalidRoutingDataException {
boolean isMsdsEndpointSet =
connectionConfig.getMsdsEndpoint() != null && !connectionConfig.getMsdsEndpoint().isEmpty();
- MetadataStoreRoutingData routingData = isMsdsEndpointSet ? HttpRoutingDataReader
- .getMetadataStoreRoutingData(connectionConfig.getMsdsEndpoint())
- : HttpRoutingDataReader.getMetadataStoreRoutingData();
+ // TODO: Make RoutingDataReaderType configurable
+ MetadataStoreRoutingData routingData = isMsdsEndpointSet ? RoutingDataManager
+ .getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, connectionConfig.getMsdsEndpoint())
+ : RoutingDataManager.getMetadataStoreRoutingData();
return routingData.getMetadataStoreRealm(connectionConfig.getZkRealmShardingKey());
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
index 0e123b2..bce8105 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
@@ -83,11 +83,12 @@ import org.apache.helix.util.HelixUtil;
import org.apache.helix.util.RebalanceUtil;
import org.apache.helix.zookeeper.api.client.HelixZkClient;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.exception.ZkClientException;
import org.apache.helix.zookeeper.impl.client.FederatedZkClient;
import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
-import org.apache.helix.zookeeper.util.HttpRoutingDataReader;
+import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.exception.ZkException;
import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
@@ -967,9 +968,11 @@ public class ZKHelixAdmin implements HelixAdmin {
String msdsEndpoint = _zkClient.getRealmAwareZkConnectionConfig().getMsdsEndpoint();
try {
if (msdsEndpoint == null || msdsEndpoint.isEmpty()) {
- realmToShardingKeys = HttpRoutingDataReader.getRawRoutingData();
+ realmToShardingKeys = RoutingDataManager.getRawRoutingData();
} else {
- realmToShardingKeys = HttpRoutingDataReader.getRawRoutingData(msdsEndpoint);
+ // TODO: Make RoutingDataReaderType configurable
+ realmToShardingKeys =
+ RoutingDataManager.getRawRoutingData(RoutingDataReaderType.HTTP, msdsEndpoint);
}
} catch (IOException e) {
throw new HelixException(
diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java b/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
index a9952da..f5d8915 100644
--- a/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
+++ b/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
@@ -48,7 +48,7 @@ import org.apache.helix.zookeeper.impl.client.FederatedZkClient;
import org.apache.helix.zookeeper.impl.client.ZkClient;
import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
-import org.apache.helix.zookeeper.util.HttpRoutingDataReader;
+import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
import org.apache.helix.zookeeper.zkclient.IZkStateListener;
@@ -323,8 +323,8 @@ public class ServerContext implements IZkDataListener, IZkChildListener, IZkStat
LOG.info("ServerContext: Resetting ZK resources due to routing data change! Routing ZK: {}",
_zkAddr);
try {
- // Reset HttpRoutingDataReader's cache
- HttpRoutingDataReader.reset();
+ // Reset RoutingDataManager's cache
+ RoutingDataManager.reset();
// All Helix APIs will be closed implicitly because ZkClient is closed
if (_zkClient != null && !_zkClient.isClosed()) {
_zkClient.close();
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 288cdfa..b1ecc38 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
@@ -19,14 +19,10 @@ package org.apache.helix.zookeeper.api.client;
* under the License.
*/
-import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
-import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
-import org.apache.helix.zookeeper.exception.ZkClientException;
-import org.apache.helix.zookeeper.util.HttpRoutingDataReader;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java
new file mode 100644
index 0000000..85b8194
--- /dev/null
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingDataReaderType.java
@@ -0,0 +1,36 @@
+package org.apache.helix.zookeeper.constant;
+
+/*
+ * 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.
+ */
+
+public enum RoutingDataReaderType {
+ HTTP("org.apache.helix.zookeeper.routing.HttpRoutingDataReader"),
+ ZK("ZkRoutingDataReader"),
+ HTTP_ZK_FALLBACK("HttpZkFallbackRoutingDataReader");
+
+ private final String className;
+
+ RoutingDataReaderType(String className) {
+ this.className = className;
+ }
+
+ public String getClassName() {
+ return this.className;
+ }
+}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
new file mode 100644
index 0000000..23b8ebc
--- /dev/null
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/RoutingSystemPropertyKeys.java
@@ -0,0 +1,31 @@
+package org.apache.helix.zookeeper.constant;
+
+/*
+ * 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.
+ */
+
+/**
+ * This class contains various routing-related system property keys for multi-zk clients.
+ */
+public class RoutingSystemPropertyKeys {
+
+ /**
+ * This static constant is used to refer to which implementation of RoutingDataReader to use.
+ */
+ public static final String ROUTING_DATA_READER_TYPE = "routing.data.reader.type";
+}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/exception/MultiZkException.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/exception/MultiZkException.java
new file mode 100644
index 0000000..22d6669
--- /dev/null
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/exception/MultiZkException.java
@@ -0,0 +1,37 @@
+package org.apache.helix.zookeeper.exception;
+/*
+ * 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.
+ */
+
+/**
+ * Exception for multi ZK-related failures.
+ */
+public class MultiZkException extends RuntimeException {
+
+ public MultiZkException(String message) {
+ super(message);
+ }
+
+ public MultiZkException(Throwable cause) {
+ super(cause);
+ }
+
+ public MultiZkException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
index 3a50c87..501670c 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
@@ -28,7 +28,8 @@ import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.api.client.ChildrenSubscribeResult;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
-import org.apache.helix.zookeeper.util.HttpRoutingDataReader;
+import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
+import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
import org.apache.helix.zookeeper.zkclient.IZkConnection;
@@ -86,9 +87,11 @@ public class DedicatedZkClient implements RealmAwareZkClient {
// Get the routing data from a static Singleton HttpRoutingDataReader
String msdsEndpoint = connectionConfig.getMsdsEndpoint();
if (msdsEndpoint == null || msdsEndpoint.isEmpty()) {
- _metadataStoreRoutingData = HttpRoutingDataReader.getMetadataStoreRoutingData();
+ _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData();
} else {
- _metadataStoreRoutingData = HttpRoutingDataReader.getMetadataStoreRoutingData(msdsEndpoint);
+ // TODO: Make RoutingDataReaderType configurable
+ _metadataStoreRoutingData =
+ RoutingDataManager.getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, msdsEndpoint);
}
_zkRealmShardingKey = connectionConfig.getZkRealmShardingKey();
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
index f5de9d8..22b9fe5 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
@@ -31,8 +31,9 @@ import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.api.client.ChildrenSubscribeResult;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
-import org.apache.helix.zookeeper.util.HttpRoutingDataReader;
+import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
@@ -98,9 +99,11 @@ public class FederatedZkClient implements RealmAwareZkClient {
// Attempt to get MetadataStoreRoutingData
String msdsEndpoint = connectionConfig.getMsdsEndpoint();
if (msdsEndpoint == null || msdsEndpoint.isEmpty()) {
- _metadataStoreRoutingData = HttpRoutingDataReader.getMetadataStoreRoutingData();
+ _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData();
} else {
- _metadataStoreRoutingData = HttpRoutingDataReader.getMetadataStoreRoutingData(msdsEndpoint);
+ // TODO: Make RoutingDataReaderType configurable
+ _metadataStoreRoutingData =
+ RoutingDataManager.getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, msdsEndpoint);
}
_isClosed = false;
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 4a83712..341731e 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
@@ -29,8 +29,9 @@ import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.api.client.ChildrenSubscribeResult;
import org.apache.helix.zookeeper.api.client.HelixZkClient;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
-import org.apache.helix.zookeeper.util.HttpRoutingDataReader;
+import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
@@ -78,9 +79,11 @@ public class SharedZkClient implements RealmAwareZkClient {
// Get the routing data from a static Singleton HttpRoutingDataReader
String msdsEndpoint = connectionConfig.getMsdsEndpoint();
if (msdsEndpoint == null || msdsEndpoint.isEmpty()) {
- _metadataStoreRoutingData = HttpRoutingDataReader.getMetadataStoreRoutingData();
+ _metadataStoreRoutingData = RoutingDataManager.getMetadataStoreRoutingData();
} else {
- _metadataStoreRoutingData = HttpRoutingDataReader.getMetadataStoreRoutingData(msdsEndpoint);
+ // TODO: Make RoutingDataReaderType configurable
+ _metadataStoreRoutingData =
+ RoutingDataManager.getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, msdsEndpoint);
}
_zkRealmShardingKey = connectionConfig.getZkRealmShardingKey();
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpRoutingDataReader.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpRoutingDataReader.java
new file mode 100644
index 0000000..4ef9881
--- /dev/null
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpRoutingDataReader.java
@@ -0,0 +1,132 @@
+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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+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.exception.MultiZkException;
+import org.apache.http.HttpEntity;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.DefaultBackoffStrategy;
+import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+
+
+/**
+ * HTTP-based RoutingDataReader that makes an HTTP call to Metadata Store Directory Service (REST)
+ * to fetch routing data.
+ */
+public class HttpRoutingDataReader implements RoutingDataReader {
+ private static final int DEFAULT_HTTP_TIMEOUT_IN_MS = 5000;
+
+ /**
+ * Returns an object form of metadata store routing data.
+ * @param endpoint
+ * @return
+ */
+ @Override
+ public MetadataStoreRoutingData getMetadataStoreRoutingData(String endpoint) {
+ try {
+ return new TrieRoutingData(getRawRoutingData(endpoint));
+ } catch (InvalidRoutingDataException e) {
+ throw new MultiZkException(e);
+ }
+ }
+
+ /**
+ * Returns a map form of metadata store routing data.
+ * The map fields stand for metadata store realm address (key), and a corresponding list of ZK
+ * path sharding keys (key).
+ * @param endpoint
+ * @return
+ */
+ @Override
+ public Map<String, List<String>> getRawRoutingData(String endpoint) {
+ try {
+ String routingDataJson = getAllRoutingData(endpoint);
+ return parseRoutingData(routingDataJson);
+ } catch (IOException e) {
+ throw new MultiZkException(e);
+ }
+ }
+
+ /**
+ * Makes an HTTP call to fetch all routing data.
+ * @return
+ * @throws IOException
+ */
+ private String getAllRoutingData(String msdsEndpoint) throws IOException {
+ // Note that MSDS_ENDPOINT should provide high-availability - it risks becoming a single point
+ // of failure if it's backed by a single IP address/host
+ // Retry count is 3 by default.
+ HttpGet requestAllData = new HttpGet(
+ msdsEndpoint + MetadataStoreRoutingConstants.MSDS_GET_ALL_ROUTING_DATA_ENDPOINT);
+
+ // Define timeout configs
+ RequestConfig config = RequestConfig.custom().setConnectTimeout(DEFAULT_HTTP_TIMEOUT_IN_MS)
+ .setConnectionRequestTimeout(DEFAULT_HTTP_TIMEOUT_IN_MS)
+ .setSocketTimeout(DEFAULT_HTTP_TIMEOUT_IN_MS).build();
+
+ try (CloseableHttpClient httpClient = HttpClients.custom().setDefaultRequestConfig(config)
+ .setConnectionBackoffStrategy(new DefaultBackoffStrategy())
+ .setRetryHandler(new DefaultHttpRequestRetryHandler()).build()) {
+ // Return the JSON because try-resources clause closes the CloseableHttpResponse
+ HttpEntity entity = httpClient.execute(requestAllData).getEntity();
+ if (entity == null) {
+ throw new IOException("Response's entity is null!");
+ }
+ return EntityUtils.toString(entity, "UTF-8");
+ }
+ }
+
+ /**
+ * Returns the raw routing data in a Map< ZkRealm, List of shardingKeys > format.
+ * @param routingDataJson
+ * @return
+ */
+ private Map<String, List<String>> parseRoutingData(String routingDataJson) throws IOException {
+ if (routingDataJson != null) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> resultMap = new ObjectMapper().readValue(routingDataJson, Map.class);
+ @SuppressWarnings("unchecked")
+ List<Map<String, Object>> routingDataList =
+ (List<Map<String, Object>>) resultMap.get(MetadataStoreRoutingConstants.ROUTING_DATA);
+ @SuppressWarnings("unchecked")
+ Map<String, List<String>> routingData = routingDataList.stream().collect(Collectors.toMap(
+ realmKeyPair -> (String) realmKeyPair
+ .get(MetadataStoreRoutingConstants.SINGLE_METADATA_STORE_REALM),
+ mapEntry -> (List<String>) mapEntry.get(MetadataStoreRoutingConstants.SHARDING_KEYS)));
+ return routingData;
+ }
+ return Collections.emptyMap();
+ }
+}
\ No newline at end of file
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
new file mode 100644
index 0000000..f8aafb2
--- /dev/null
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
@@ -0,0 +1,181 @@
+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 static 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 static final String DEFAULT_MSDS_ENDPOINT =
+ 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 static final Map<String, Map<String, List<String>>> _rawRoutingDataMap =
+ new ConcurrentHashMap<>();
+ // The following map stands for (RoutingDataReaderType_endpoint ID, MetadataStoreRoutingData)
+ private static final Map<String, MetadataStoreRoutingData> _metadataStoreRoutingDataMap =
+ new ConcurrentHashMap<>();
+
+ /**
+ * This class is a Singleton.
+ */
+ private RoutingDataManager() {
+ }
+
+ /**
+ * Fetches routing data from the data source via HTTP by querying the MSDS configured in the JVM
+ * config.
+ * @return
+ * @throws IOException
+ */
+ public static Map<String, List<String>> getRawRoutingData() throws IOException {
+ if (DEFAULT_MSDS_ENDPOINT == null || DEFAULT_MSDS_ENDPOINT.isEmpty()) {
+ throw new IllegalStateException(
+ "HttpRoutingDataReader was unable to find a valid MSDS endpoint String in System "
+ + "Properties!");
+ }
+ return getRawRoutingData(RoutingDataReaderType.HTTP, DEFAULT_MSDS_ENDPOINT);
+ }
+
+ /**
+ * 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 static 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 static MetadataStoreRoutingData getMetadataStoreRoutingData()
+ throws InvalidRoutingDataException {
+ if (DEFAULT_MSDS_ENDPOINT == null || DEFAULT_MSDS_ENDPOINT.isEmpty()) {
+ throw new IllegalStateException(
+ "HttpRoutingDataReader was unable to find a valid MSDS endpoint String in System "
+ + "Properties!");
+ }
+ return getMetadataStoreRoutingData(RoutingDataReaderType.HTTP, DEFAULT_MSDS_ENDPOINT);
+ }
+
+ /**
+ * Returns the routing data read from MSDS as a MetadataStoreRoutingData object.
+ * @param routingDataReaderType
+ * @param endpoint
+ * @return
+ * @throws IOException
+ * @throws InvalidRoutingDataException
+ */
+ public static 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 static void reset() {
+ _rawRoutingDataMap.clear();
+ _metadataStoreRoutingDataMap.clear();
+ }
+
+ /**
+ * Returns an appropriate instance of RoutingDataReader given the type.
+ * @param routingDataReaderType
+ * @return
+ */
+ private static RoutingDataReader resolveRoutingDataReader(
+ RoutingDataReaderType routingDataReaderType) {
+ // RoutingDataReaderType.HTTP by default if not found
+ routingDataReaderType =
+ routingDataReaderType == null ? RoutingDataReaderType.HTTP : 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(
+ "RoutingDataManager: failed to instantiate RoutingDataReader! ReaderType className: "
+ + routingDataReaderType.getClassName(), e);
+ }
+ }
+
+ /**
+ * Constructs a key for the cache lookup.
+ * @param routingDataReaderType
+ * @param endpoint
+ * @return
+ */
+ private static String getRoutingDataCacheKey(RoutingDataReaderType routingDataReaderType,
+ String endpoint) {
+ return routingDataReaderType.name() + "_" + endpoint;
+ }
+}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataReader.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataReader.java
new file mode 100644
index 0000000..9e6c258
--- /dev/null
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataReader.java
@@ -0,0 +1,45 @@
+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.util.List;
+import java.util.Map;
+
+import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
+
+
+public interface RoutingDataReader {
+
+ /**
+ * Returns an object form of metadata store routing data.
+ * @param endpoint
+ * @return
+ */
+ MetadataStoreRoutingData getMetadataStoreRoutingData(String endpoint);
+
+ /**
+ * Returns a map form of metadata store routing data.
+ * The map fields stand for metadata store realm address (key), and a corresponding list of ZK
+ * path sharding keys (key).
+ * @param endpoint
+ * @return
+ */
+ Map<String, List<String>> getRawRoutingData(String endpoint);
+}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/util/HttpRoutingDataReader.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/util/HttpRoutingDataReader.java
deleted file mode 100644
index 6c12046..0000000
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/util/HttpRoutingDataReader.java
+++ /dev/null
@@ -1,197 +0,0 @@
-package org.apache.helix.zookeeper.util;
-
-/*
- * 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.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-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.http.HttpEntity;
-import org.apache.http.client.config.RequestConfig;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.DefaultBackoffStrategy;
-import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
-import org.apache.http.impl.client.HttpClients;
-import org.apache.http.util.EntityUtils;
-
-
-public class HttpRoutingDataReader {
- private static final String SYSTEM_MSDS_ENDPOINT =
- System.getProperty(MetadataStoreRoutingConstants.MSDS_SERVER_ENDPOINT_KEY);
- private static final int HTTP_TIMEOUT_IN_MS = 5000;
-
- /** Double-checked locking requires that the following fields be volatile */
- // The following map stands for (MSDS endpoint, Raw Routing Data)
- private static volatile Map<String, Map<String, List<String>>> _rawRoutingDataMap =
- new ConcurrentHashMap<>();
- // The following map stands for (MSDS endpoint, MetadataStoreRoutingData)
- private static volatile Map<String, MetadataStoreRoutingData> _metadataStoreRoutingDataMap =
- new ConcurrentHashMap<>();
-
- /**
- * This class is a Singleton.
- */
- private HttpRoutingDataReader() {
- }
-
- /**
- * Fetches routing data from the data source via HTTP by querying the MSDS configured in the JVM config.
- * @return
- * @throws IOException
- */
- public static Map<String, List<String>> getRawRoutingData() throws IOException {
- if (SYSTEM_MSDS_ENDPOINT == null || SYSTEM_MSDS_ENDPOINT.isEmpty()) {
- throw new IllegalStateException(
- "HttpRoutingDataReader was unable to find a valid MSDS endpoint String in System Properties!");
- }
- return getRawRoutingData(SYSTEM_MSDS_ENDPOINT);
- }
-
- /**
- * 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 msdsEndpoint Metadata Store Directory Store endpoint to query from
- */
- public static Map<String, List<String>> getRawRoutingData(String msdsEndpoint)
- throws IOException {
- Map<String, List<String>> rawRoutingData = _rawRoutingDataMap.get(msdsEndpoint);
- if (rawRoutingData == null) {
- synchronized (HttpRoutingDataReader.class) {
- rawRoutingData = _rawRoutingDataMap.get(msdsEndpoint);
- if (rawRoutingData == null) {
- String routingDataJson = getAllRoutingData(msdsEndpoint);
- // Update the reference if reading routingData over HTTP is successful
- rawRoutingData = parseRoutingData(routingDataJson);
- _rawRoutingDataMap.put(msdsEndpoint, 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
- * @throws IOException
- * @throws InvalidRoutingDataException
- */
- public static MetadataStoreRoutingData getMetadataStoreRoutingData()
- throws IOException, InvalidRoutingDataException {
- if (SYSTEM_MSDS_ENDPOINT == null || SYSTEM_MSDS_ENDPOINT.isEmpty()) {
- throw new IllegalStateException(
- "HttpRoutingDataReader was unable to find a valid MSDS endpoint String in System Properties!");
- }
- return getMetadataStoreRoutingData(SYSTEM_MSDS_ENDPOINT);
- }
-
- /**
- * Returns the routing data read from MSDS in a MetadataStoreRoutingData format.
- * @param msdsEndpoint Metadata Store Directory Store endpoint to query from
- * @return MetadataStoreRoutingData
- * @throws IOException if there is an issue connecting to MSDS
- * @throws InvalidRoutingDataException if the raw routing data is not valid
- */
- public static MetadataStoreRoutingData getMetadataStoreRoutingData(String msdsEndpoint)
- throws IOException, InvalidRoutingDataException {
- MetadataStoreRoutingData metadataStoreRoutingData =
- _metadataStoreRoutingDataMap.get(msdsEndpoint);
- if (metadataStoreRoutingData == null) {
- synchronized (HttpRoutingDataReader.class) {
- metadataStoreRoutingData = _metadataStoreRoutingDataMap.get(msdsEndpoint);
- if (metadataStoreRoutingData == null) {
- metadataStoreRoutingData = new TrieRoutingData(getRawRoutingData(msdsEndpoint));
- _metadataStoreRoutingDataMap.put(msdsEndpoint, metadataStoreRoutingData);
- }
- }
- }
- return metadataStoreRoutingData;
- }
-
- /**
- * Clears the statically-cached routing data in HttpRoutingDataReader.
- */
- public static void reset() {
- _rawRoutingDataMap.clear();
- _metadataStoreRoutingDataMap.clear();
- }
-
- /**
- * Makes an HTTP call to fetch all routing data.
- * @return
- * @throws IOException
- */
- private static String getAllRoutingData(String msdsEndpoint) throws IOException {
- // Note that MSDS_ENDPOINT should provide high-availability - it risks becoming a single point
- // of failure if it's backed by a single IP address/host
- // Retry count is 3 by default.
- HttpGet requestAllData = new HttpGet(
- msdsEndpoint + MetadataStoreRoutingConstants.MSDS_GET_ALL_ROUTING_DATA_ENDPOINT);
-
- // Define timeout configs
- RequestConfig config = RequestConfig.custom().setConnectTimeout(HTTP_TIMEOUT_IN_MS)
- .setConnectionRequestTimeout(HTTP_TIMEOUT_IN_MS).setSocketTimeout(HTTP_TIMEOUT_IN_MS)
- .build();
-
- try (CloseableHttpClient httpClient = HttpClients.custom().setDefaultRequestConfig(config)
- .setConnectionBackoffStrategy(new DefaultBackoffStrategy())
- .setRetryHandler(new DefaultHttpRequestRetryHandler()).build()) {
- // Return the JSON because try-resources clause closes the CloseableHttpResponse
- HttpEntity entity = httpClient.execute(requestAllData).getEntity();
- if (entity == null) {
- throw new IOException("Response's entity is null!");
- }
- return EntityUtils.toString(entity, "UTF-8");
- }
- }
-
- /**
- * Returns the raw routing data in a Map< ZkRealm, List of shardingKeys > format.
- * @param routingDataJson
- * @return
- */
- private static Map<String, List<String>> parseRoutingData(String routingDataJson)
- throws IOException {
- if (routingDataJson != null) {
- @SuppressWarnings("unchecked")
- Map<String, Object> resultMap = new ObjectMapper().readValue(routingDataJson, Map.class);
- @SuppressWarnings("unchecked")
- List<Map<String, Object>> routingDataList =
- (List<Map<String, Object>>) resultMap.get(MetadataStoreRoutingConstants.ROUTING_DATA);
- @SuppressWarnings("unchecked")
- Map<String, List<String>> routingData = routingDataList.stream().collect(Collectors.toMap(
- realmKeyPair -> (String) realmKeyPair
- .get(MetadataStoreRoutingConstants.SINGLE_METADATA_STORE_REALM),
- mapEntry -> (List<String>) mapEntry.get(MetadataStoreRoutingConstants.SHARDING_KEYS)));
- return routingData;
- }
- return Collections.emptyMap();
- }
-}
\ No newline at end of file
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestHttpRoutingDataReader.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
similarity index 90%
rename from zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestHttpRoutingDataReader.java
rename to zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
index 5c52d05..ad16fb3 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestHttpRoutingDataReader.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
@@ -35,17 +35,18 @@ import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.msdcommon.mock.MockMetadataStoreDirectoryServer;
import org.apache.helix.zookeeper.constant.TestConstants;
import org.apache.helix.zookeeper.impl.ZkTestBase;
+import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-public class TestHttpRoutingDataReader extends ZkTestBase {
+public class TestRoutingDataManager extends ZkTestBase {
private MockMetadataStoreDirectoryServer _msdsServer;
private final String _host = "localhost";
private final int _port = 1991;
- private final String _namespace = "TestHttpRoutingDataReader";
+ private final String _namespace = "TestRoutingDataManager";
@BeforeClass
public void beforeClass() throws IOException {
@@ -66,14 +67,14 @@ public class TestHttpRoutingDataReader extends ZkTestBase {
@Test
public void testGetRawRoutingData() throws IOException {
- Map<String, List<String>> rawRoutingData = HttpRoutingDataReader.getRawRoutingData();
+ Map<String, List<String>> rawRoutingData = RoutingDataManager.getRawRoutingData();
TestConstants.FAKE_ROUTING_DATA.forEach((realm, keys) -> Assert
.assertEquals(new HashSet(rawRoutingData.get(realm)), new HashSet(keys)));
}
@Test(dependsOnMethods = "testGetRawRoutingData")
public void testGetMetadataStoreRoutingData() throws IOException, InvalidRoutingDataException {
- MetadataStoreRoutingData data = HttpRoutingDataReader.getMetadataStoreRoutingData();
+ MetadataStoreRoutingData data = RoutingDataManager.getMetadataStoreRoutingData();
Map<String, String> allMappings = data.getAllMappingUnderPath("/");
Map<String, Set<String>> groupedMappings = allMappings.entrySet().stream().collect(Collectors
.groupingBy(Map.Entry::getValue,
@@ -101,7 +102,7 @@ public class TestHttpRoutingDataReader extends ZkTestBase {
// HttpRoutingDataReader should still return old data because it's static
// Make sure the results don't contain the new realm
- Map<String, List<String>> rawRoutingData = HttpRoutingDataReader.getRawRoutingData();
+ Map<String, List<String>> rawRoutingData = RoutingDataManager.getRawRoutingData();
Assert.assertFalse(rawRoutingData.containsKey(newRealm));
// Remove newRealm and check for equality
@@ -110,7 +111,7 @@ public class TestHttpRoutingDataReader extends ZkTestBase {
TestConstants.FAKE_ROUTING_DATA.forEach((realm, keys) -> Assert
.assertEquals(new HashSet(rawRoutingData.get(realm)), new HashSet(keys)));
- MetadataStoreRoutingData data = HttpRoutingDataReader.getMetadataStoreRoutingData();
+ MetadataStoreRoutingData data = RoutingDataManager.getMetadataStoreRoutingData();
Map<String, String> allMappings = data.getAllMappingUnderPath("/");
Map<String, Set<String>> groupedMappings = allMappings.entrySet().stream().collect(Collectors
.groupingBy(Map.Entry::getValue,
[helix] 05/12: Add HttpZkFallbackRoutingDataReader
Posted by hu...@apache.org.
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
commit 0ff7e95c21c6a2d85009cfdf4c457a3d5b9b961d
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Wed Jul 22 11:09:49 2020 -0700
Add HttpZkFallbackRoutingDataReader
Implement an Http-ZK fallback routing data reader. ZkRoutingDataReader will follow.
---
.../routing/HttpZkFallbackRoutingDataReader.java | 79 +++++++++++++++++++++-
.../TestHttpZkFallbackRoutingDataReader.java | 71 ++++++++++++++++++-
2 files changed, 148 insertions(+), 2 deletions(-)
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpZkFallbackRoutingDataReader.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpZkFallbackRoutingDataReader.java
index 1b4713c..add034f 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpZkFallbackRoutingDataReader.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/HttpZkFallbackRoutingDataReader.java
@@ -1,4 +1,81 @@
package org.apache.helix.zookeeper.routing;
-public class HttpZkFallbackRoutingDataReader {
+/*
+ * 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.util.List;
+import java.util.Map;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
+import org.apache.helix.zookeeper.exception.MultiZkException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * HTTP and ZK-based RoutingDataReader that first tries an HTTP call to MSDS and upon failure,
+ * falls back to ZK for routing data.
+ * HttpZkFallbackRoutingDataReader does not maintain a ZK connection - it establishes for reading
+ * and closes it right away.
+ */
+public class HttpZkFallbackRoutingDataReader implements RoutingDataReader {
+ private static final Logger LOG = LoggerFactory.getLogger(HttpZkFallbackRoutingDataReader.class);
+
+ /**
+ * Returns a map form of metadata store routing data.
+ * The map fields stand for metadata store realm address (key), and a corresponding list of ZK
+ * path sharding keys (key).
+ * @param endpoint two comma-separated endpoints. <HTTP endpoint>,<ZK address>
+ * @return
+ */
+ @Override
+ public Map<String, List<String>> getRawRoutingData(String endpoint) {
+ Map<RoutingDataReaderType, String> endpointMap = parseEndpoint(endpoint);
+ try {
+ return new HttpRoutingDataReader()
+ .getRawRoutingData(endpointMap.get(RoutingDataReaderType.HTTP));
+ } catch (MultiZkException e) {
+ LOG.warn(
+ "HttpZkFallbackRoutingDataReader::getRawRoutingData: failed to read routing data via "
+ + "HTTP. Falling back to ZK!", e);
+ // TODO: increment failure count and emit as a metric
+ return new ZkRoutingDataReader().getRawRoutingData(endpointMap.get(RoutingDataReaderType.ZK));
+ }
+ }
+
+ /**
+ * For a fallback routing data reader, endpoints are given in a comma-separated string in the
+ * fallback order. This method parses the string and returns a map of routing data source type to
+ * its respective endpoint.
+ * @param endpointString
+ * @return
+ */
+ private Map<RoutingDataReaderType, String> parseEndpoint(String endpointString) {
+ String[] endpoints = endpointString.split(",");
+ if (endpoints.length != 2) {
+ throw new MultiZkException(
+ "HttpZkFallbackRoutingDataReader::parseEndpoint: endpoint string does not contain two "
+ + "proper comma-separated endpoints for HTTP and ZK! Endpoint string: "
+ + endpointString);
+ }
+ return ImmutableMap
+ .of(RoutingDataReaderType.HTTP, endpoints[0], RoutingDataReaderType.ZK, endpoints[1]);
+ }
}
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestHttpZkFallbackRoutingDataReader.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestHttpZkFallbackRoutingDataReader.java
index 85896af..d959ad2 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestHttpZkFallbackRoutingDataReader.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/routing/TestHttpZkFallbackRoutingDataReader.java
@@ -1,4 +1,73 @@
package org.apache.helix.zookeeper.routing;
-public class TestHttpZkFallbackRoutingDataReader {
+/*
+ * 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.HashSet;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.helix.msdcommon.mock.MockMetadataStoreDirectoryServer;
+import org.apache.helix.zookeeper.constant.TestConstants;
+import org.apache.helix.zookeeper.exception.MultiZkException;
+import org.apache.helix.zookeeper.impl.ZkTestBase;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class TestHttpZkFallbackRoutingDataReader extends ZkTestBase {
+ private MockMetadataStoreDirectoryServer _msdsServer;
+ private static final String HOST = "localhost";
+ private static final int PORT = 1992;
+ private static final String NAMESPACE = "TestHttpZkFallbackRoutingDataReader";
+ private static final String MSDS_ENDPOINT =
+ "http://" + HOST + ":" + PORT + "/admin/v2/namespaces/" + NAMESPACE;
+
+ @BeforeClass
+ public void beforeClass() throws IOException {
+ // Start MockMSDS
+ _msdsServer = new MockMetadataStoreDirectoryServer(HOST, PORT, NAMESPACE,
+ TestConstants.FAKE_ROUTING_DATA);
+ _msdsServer.startServer();
+ }
+
+ @Test
+ public void testGetRawRoutingData() {
+ HttpZkFallbackRoutingDataReader reader = new HttpZkFallbackRoutingDataReader();
+
+ // This read should read from HTTP
+ String endpointString = MSDS_ENDPOINT + "," + ZK_ADDR;
+ Map<String, List<String>> rawRoutingData = reader.getRawRoutingData(endpointString);
+ TestConstants.FAKE_ROUTING_DATA.forEach((realm, keys) -> Assert
+ .assertEquals(new HashSet(rawRoutingData.get(realm)), new HashSet(keys)));
+
+ // Shut down MSDS so that it would read from ZK (fallback)
+ _msdsServer.stopServer();
+ try {
+ reader.getRawRoutingData(endpointString);
+ Assert.fail("Must encounter a MultiZkException since the path in ZK does not exist!");
+ } catch (MultiZkException e) {
+ // Check the exception message to ensure that it's reading from ZK
+ Assert.assertTrue(e.getMessage()
+ .contains("Routing data directory ZNode /METADATA_STORE_ROUTING_DATA does not exist"));
+ }
+ }
}
[helix] 03/12: Remove unnecessary IOException
Posted by hu...@apache.org.
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
commit 0a9584045a206313f4128ca3e6c96860bd7f76f7
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Tue Jul 21 10:44:04 2020 -0700
Remove unnecessary IOException
---
.../org/apache/helix/manager/zk/ZKHelixAdmin.java | 19 +++++++------------
.../helix/zookeeper/routing/RoutingDataManager.java | 3 +--
.../helix/zookeeper/util/TestRoutingDataManager.java | 2 +-
3 files changed, 9 insertions(+), 15 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
index bce8105..348f8d8 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
@@ -966,19 +966,14 @@ public class ZKHelixAdmin implements HelixAdmin {
// If on multi-zk mode, we retrieve cluster information from Metadata Store Directory Service.
Map<String, List<String>> realmToShardingKeys;
String msdsEndpoint = _zkClient.getRealmAwareZkConnectionConfig().getMsdsEndpoint();
- try {
- if (msdsEndpoint == null || msdsEndpoint.isEmpty()) {
- realmToShardingKeys = RoutingDataManager.getRawRoutingData();
- } else {
- // TODO: Make RoutingDataReaderType configurable
- realmToShardingKeys =
- RoutingDataManager.getRawRoutingData(RoutingDataReaderType.HTTP, msdsEndpoint);
- }
- } catch (IOException e) {
- throw new HelixException(
- "ZKHelixAdmin: Failed to read raw routing data from Metadata Store Directory Service! MSDS endpoint used: "
- + msdsEndpoint, e);
+ if (msdsEndpoint == null || msdsEndpoint.isEmpty()) {
+ realmToShardingKeys = RoutingDataManager.getRawRoutingData();
+ } else {
+ // TODO: Make RoutingDataReaderType configurable
+ realmToShardingKeys =
+ RoutingDataManager.getRawRoutingData(RoutingDataReaderType.HTTP, msdsEndpoint);
}
+
if (realmToShardingKeys == null || realmToShardingKeys.isEmpty()) {
return Collections.emptyList();
}
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
index f8aafb2..aa3986d 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/routing/RoutingDataManager.java
@@ -61,9 +61,8 @@ public class RoutingDataManager {
* Fetches routing data from the data source via HTTP by querying the MSDS configured in the JVM
* config.
* @return
- * @throws IOException
*/
- public static Map<String, List<String>> getRawRoutingData() throws IOException {
+ public static Map<String, List<String>> getRawRoutingData() {
if (DEFAULT_MSDS_ENDPOINT == null || DEFAULT_MSDS_ENDPOINT.isEmpty()) {
throw new IllegalStateException(
"HttpRoutingDataReader was unable to find a valid MSDS endpoint String in System "
diff --git a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
index ad16fb3..e7d0c43 100644
--- a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestRoutingDataManager.java
@@ -66,7 +66,7 @@ public class TestRoutingDataManager extends ZkTestBase {
}
@Test
- public void testGetRawRoutingData() throws IOException {
+ public void testGetRawRoutingData() {
Map<String, List<String>> rawRoutingData = RoutingDataManager.getRawRoutingData();
TestConstants.FAKE_ROUTING_DATA.forEach((realm, keys) -> Assert
.assertEquals(new HashSet(rawRoutingData.get(realm)), new HashSet(keys)));