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:14:05 UTC
[helix] 10/12: Implement routing data update upon cache miss for
FederatedZkClient
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