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/04/08 22:53:51 UTC
[helix] 21/50: Add HttpRoutingDataReader (#775)
This is an automated email from the ASF dual-hosted git repository.
hulee pushed a commit to branch zooscalability_merge
in repository https://gitbox.apache.org/repos/asf/helix.git
commit f8fa75ab6c9677abced2aa59ec57380e54ffe111
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Tue Feb 25 14:45:28 2020 -0800
Add HttpRoutingDataReader (#775)
HttpRoutingDataReader is a component used by new ZkClient APIs to make an HTTP read request to the metadata store directory service to retrieve routing data. ZkClient APIs will construct an internal MetadataStoreRoutingData instance based on the raw routing data retrieved from MSDS.
Note: this change contains modifications to MockMSDS because the actual endpoint names changed. The methods and http server contexts have been updated.
---
.../mock/TestMockMetadataStoreDirectoryServer.java | 84 ------------
metadata-store-directory-common/pom.xml | 5 +
.../constant/MetadataStoreRoutingConstants.java | 9 ++
.../mock/MockMetadataStoreDirectoryServer.java | 68 +++++++---
.../mock/TestMockMetadataStoreDirectoryServer.java | 118 ++++++++++++++++
zookeeper-api/pom.xml | 5 +
.../zookeeper/util/HttpRoutingDataReader.java | 149 +++++++++++++++++++++
.../zookeeper/util/TestHttpRoutingDataReader.java | 128 ++++++++++++++++++
zookeeper-api/zookeeper-api-0.9.2-SNAPSHOT.ivy | 1 +
9 files changed, 465 insertions(+), 102 deletions(-)
diff --git a/helix-rest/src/test/java/org/apache/helix/rest/metadatastore/mock/TestMockMetadataStoreDirectoryServer.java b/helix-rest/src/test/java/org/apache/helix/rest/metadatastore/mock/TestMockMetadataStoreDirectoryServer.java
deleted file mode 100644
index 5e71089..0000000
--- a/helix-rest/src/test/java/org/apache/helix/rest/metadatastore/mock/TestMockMetadataStoreDirectoryServer.java
+++ /dev/null
@@ -1,84 +0,0 @@
-package org.apache.helix.rest.metadatastore.mock;
-
-/*
- * 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.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import com.google.common.collect.ImmutableList;
-import org.apache.http.client.methods.CloseableHttpResponse;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClients;
-import org.junit.Test;
-import org.testng.Assert;
-
-
-public class TestMockMetadataStoreDirectoryServer {
- @Test
- public void testMockMetadataStoreDirectoryServer()
- throws IOException {
- // Create fake routing data
- Map<String, List<String>> routingData = new HashMap<>();
- routingData.put("zk-0", ImmutableList.of("sharding-key-0", "sharding-key-1", "sharding-key-2"));
- routingData.put("zk-1", ImmutableList.of("sharding-key-3", "sharding-key-4", "sharding-key-5"));
- routingData.put("zk-2", ImmutableList.of("sharding-key-6", "sharding-key-7", "sharding-key-8"));
-
- // Start MockMSDS
- String host = "localhost";
- int port = 11000;
- String endpoint = "http://" + host + ":" + port;
- String namespace = "MY-HELIX-NAMESPACE";
- MockMetadataStoreDirectoryServer server =
- new MockMetadataStoreDirectoryServer(host, port, namespace, routingData);
- server.startServer();
- CloseableHttpClient httpClient = HttpClients.createDefault();
-
- // Send a GET request
- String testZkRealm = "zk-0";
- HttpGet getRequest = new HttpGet(
- endpoint + MockMetadataStoreDirectoryServer.REST_PREFIX + namespace
- + MockMetadataStoreDirectoryServer.ZK_REALM_ENDPOINT + testZkRealm);
- try {
- CloseableHttpResponse getResponse = httpClient.execute(getRequest);
- List<String> shardingKeyList = MockMetadataStoreDirectoryServer.OBJECT_MAPPER
- .readValue(getResponse.getEntity().getContent(), List.class);
- Assert.assertEquals(shardingKeyList, routingData.get(testZkRealm));
- } catch (IOException e) {
- e.printStackTrace();
- }
-
- // Try sending a POST request (not supported)
- HttpPost postRequest = new HttpPost(
- endpoint + MockMetadataStoreDirectoryServer.REST_PREFIX + namespace
- + MockMetadataStoreDirectoryServer.ZK_REALM_ENDPOINT + testZkRealm);
- try {
- CloseableHttpResponse postResponse = httpClient.execute(postRequest);
- } catch (IOException e) {
- e.printStackTrace();
- }
-
- // Shutdown
- server.stopServer();
- }
-}
diff --git a/metadata-store-directory-common/pom.xml b/metadata-store-directory-common/pom.xml
index cc81a33..1b0d964 100644
--- a/metadata-store-directory-common/pom.xml
+++ b/metadata-store-directory-common/pom.xml
@@ -44,6 +44,11 @@ under the License.
<dependencies>
<dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>4.5.8</version>
+ </dependency>
+ <dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.10.2</version>
diff --git a/metadata-store-directory-common/src/main/java/org/apache/helix/msdcommon/constant/MetadataStoreRoutingConstants.java b/metadata-store-directory-common/src/main/java/org/apache/helix/msdcommon/constant/MetadataStoreRoutingConstants.java
index 2358bcd..e3d541b 100644
--- a/metadata-store-directory-common/src/main/java/org/apache/helix/msdcommon/constant/MetadataStoreRoutingConstants.java
+++ b/metadata-store-directory-common/src/main/java/org/apache/helix/msdcommon/constant/MetadataStoreRoutingConstants.java
@@ -62,4 +62,13 @@ public class MetadataStoreRoutingConstants {
* }
*/
public static final String SHARDING_KEY_PATH_PREFIX = "prefix";
+
+ // System Property Metadata Store Directory Server endpoint key
+ public static final String MSDS_SERVER_ENDPOINT_KEY = "metadataStoreDirectoryServerEndpoint";
+
+ // MSDS resource getAllRealms endpoint string
+ public static final String MSDS_GET_ALL_REALMS_ENDPOINT = "/metadata-store-realms";
+
+ // MSDS resource get all routing data endpoint string
+ public static final String MSDS_GET_ALL_ROUTING_DATA_ENDPOINT = "/routing-data";
}
diff --git a/helix-rest/src/test/java/org/apache/helix/rest/metadatastore/mock/MockMetadataStoreDirectoryServer.java b/metadata-store-directory-common/src/main/java/org/apache/helix/msdcommon/mock/MockMetadataStoreDirectoryServer.java
similarity index 61%
rename from helix-rest/src/test/java/org/apache/helix/rest/metadatastore/mock/MockMetadataStoreDirectoryServer.java
rename to metadata-store-directory-common/src/main/java/org/apache/helix/msdcommon/mock/MockMetadataStoreDirectoryServer.java
index ae0f85d..f2a59d6 100644
--- a/helix-rest/src/test/java/org/apache/helix/rest/metadatastore/mock/MockMetadataStoreDirectoryServer.java
+++ b/metadata-store-directory-common/src/main/java/org/apache/helix/msdcommon/mock/MockMetadataStoreDirectoryServer.java
@@ -1,4 +1,4 @@
-package org.apache.helix.rest.metadatastore.mock;
+package org.apache.helix.msdcommon.mock;
/*
* Licensed to the Apache Software Foundation (ASF) under one
@@ -22,13 +22,19 @@ package org.apache.helix.rest.metadatastore.mock;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
+import java.util.stream.Collectors;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;
+import org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,14 +47,15 @@ public class MockMetadataStoreDirectoryServer {
private static final Logger LOG = LoggerFactory.getLogger(MockMetadataStoreDirectoryServer.class);
protected static final String REST_PREFIX = "/admin/v2/namespaces/";
- protected static final String ZK_REALM_ENDPOINT = "/METADATA_STORE_ROUTING_DATA/";
+ protected static final String ZK_REALM_ENDPOINT =
+ MetadataStoreRoutingConstants.MSDS_GET_ALL_REALMS_ENDPOINT;
protected static final int NOT_IMPLEMENTED = 501;
protected static final int OK = 200;
protected static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
protected final String _hostname;
protected final int _mockServerPort;
- protected final Map<String, List<String>> _routingDataMap;
+ protected final Map<String, Collection<String>> _routingDataMap;
protected final String _namespace;
protected HttpServer _server;
protected final ThreadPoolExecutor _executor =
@@ -68,7 +75,7 @@ public class MockMetadataStoreDirectoryServer {
* @param routingData <ZK realm, List of ZK path sharding keys>
*/
public MockMetadataStoreDirectoryServer(String hostname, int port, String namespace,
- Map<String, List<String>> routingData) {
+ Map<String, Collection<String>> routingData) {
if (hostname == null || hostname.isEmpty()) {
throw new IllegalArgumentException("hostname cannot be null or empty!");
}
@@ -108,20 +115,45 @@ public class MockMetadataStoreDirectoryServer {
* Dynamically generates HTTP server contexts based on the routing data given.
*/
private void generateContexts() {
+ // Get all routing data endpoint
+ // Get the result to be in the MetadataStoreShardingKeysByRealm format
+ List<Map<String, Object>> result = _routingDataMap.entrySet().stream().map(entry -> ImmutableMap
+ .of(MetadataStoreRoutingConstants.SINGLE_METADATA_STORE_REALM, entry.getKey(),
+ MetadataStoreRoutingConstants.SHARDING_KEYS, entry.getValue()))
+ .collect(Collectors.toList());
+ _server.createContext(
+ REST_PREFIX + _namespace + MetadataStoreRoutingConstants.MSDS_GET_ALL_ROUTING_DATA_ENDPOINT,
+ createHttpHandler(ImmutableMap
+ .of(MetadataStoreRoutingConstants.SINGLE_METADATA_STORE_NAMESPACE, _namespace,
+ MetadataStoreRoutingConstants.ROUTING_DATA, result)));
+
+ // Get all realms endpoint
+ _server.createContext(REST_PREFIX + _namespace + ZK_REALM_ENDPOINT, createHttpHandler(
+ ImmutableMap
+ .of(MetadataStoreRoutingConstants.METADATA_STORE_REALMS, _routingDataMap.keySet())));
+
+ // Get all sharding keys for a realm endpoint
_routingDataMap.forEach((zkRealm, shardingKeyList) -> _server
- .createContext(REST_PREFIX + _namespace + ZK_REALM_ENDPOINT + zkRealm, httpExchange -> {
- OutputStream outputStream = httpExchange.getResponseBody();
- String htmlResponse;
- if (SupportedHttpVerbs.GET.name().equals(httpExchange.getRequestMethod())) {
- htmlResponse = OBJECT_MAPPER.writeValueAsString(shardingKeyList);
- httpExchange.sendResponseHeaders(OK, htmlResponse.length());
- } else {
- htmlResponse = httpExchange.getRequestMethod() + " is not supported!\n";
- httpExchange.sendResponseHeaders(NOT_IMPLEMENTED, htmlResponse.length());
- }
- outputStream.write(htmlResponse.getBytes());
- outputStream.flush();
- outputStream.close();
- }));
+ .createContext(REST_PREFIX + _namespace + ZK_REALM_ENDPOINT + "/" + zkRealm,
+ createHttpHandler(ImmutableMap
+ .of(MetadataStoreRoutingConstants.SINGLE_METADATA_STORE_REALM, zkRealm,
+ MetadataStoreRoutingConstants.SHARDING_KEYS, shardingKeyList))));
+ }
+
+ private HttpHandler createHttpHandler(Map<String, Object> keyValuePairs) {
+ return httpExchange -> {
+ OutputStream outputStream = httpExchange.getResponseBody();
+ String htmlResponse;
+ if (SupportedHttpVerbs.GET.name().equals(httpExchange.getRequestMethod())) {
+ htmlResponse = OBJECT_MAPPER.writeValueAsString(keyValuePairs);
+ httpExchange.sendResponseHeaders(OK, htmlResponse.length());
+ } else {
+ htmlResponse = httpExchange.getRequestMethod() + " is not supported!\n";
+ httpExchange.sendResponseHeaders(NOT_IMPLEMENTED, htmlResponse.length());
+ }
+ outputStream.write(htmlResponse.getBytes());
+ outputStream.flush();
+ outputStream.close();
+ };
}
}
diff --git a/metadata-store-directory-common/src/test/java/org/apache/helix/msdcommon/mock/TestMockMetadataStoreDirectoryServer.java b/metadata-store-directory-common/src/test/java/org/apache/helix/msdcommon/mock/TestMockMetadataStoreDirectoryServer.java
new file mode 100644
index 0000000..1dc006a
--- /dev/null
+++ b/metadata-store-directory-common/src/test/java/org/apache/helix/msdcommon/mock/TestMockMetadataStoreDirectoryServer.java
@@ -0,0 +1,118 @@
+package org.apache.helix.msdcommon.mock;
+
+/*
+ * 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.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.junit.Test;
+import org.testng.Assert;
+
+
+public class TestMockMetadataStoreDirectoryServer {
+ @Test
+ public void testMockMetadataStoreDirectoryServer() throws IOException {
+ // Create fake routing data
+ Map<String, Collection<String>> routingData = new HashMap<>();
+ routingData.put("zk-0", ImmutableList.of("sharding-key-0", "sharding-key-1", "sharding-key-2"));
+ routingData.put("zk-1", ImmutableList.of("sharding-key-3", "sharding-key-4", "sharding-key-5"));
+ routingData.put("zk-2", ImmutableList.of("sharding-key-6", "sharding-key-7", "sharding-key-8"));
+
+ // Start MockMSDS
+ String host = "localhost";
+ int port = 11000;
+ String endpoint = "http://" + host + ":" + port;
+ String namespace = "MY-HELIX-NAMESPACE";
+
+ MockMetadataStoreDirectoryServer server =
+ new MockMetadataStoreDirectoryServer(host, port, namespace, routingData);
+ server.startServer();
+ try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
+ // Send a GET request for all routing data
+ HttpGet getRequest = new HttpGet(
+ endpoint + MockMetadataStoreDirectoryServer.REST_PREFIX + namespace
+ + MetadataStoreRoutingConstants.MSDS_GET_ALL_ROUTING_DATA_ENDPOINT);
+
+ CloseableHttpResponse getResponse = httpClient.execute(getRequest);
+ Map<String, Object> resultMap = MockMetadataStoreDirectoryServer.OBJECT_MAPPER
+ .readValue(getResponse.getEntity().getContent(), Map.class);
+ List<Map<String, Object>> routingDataList =
+ (List<Map<String, Object>>) resultMap.get(MetadataStoreRoutingConstants.ROUTING_DATA);
+ Collection<String> allRealms = routingDataList.stream().map(mapEntry -> (String) mapEntry
+ .get(MetadataStoreRoutingConstants.SINGLE_METADATA_STORE_REALM))
+ .collect(Collectors.toSet());
+ Assert.assertEquals(allRealms, routingData.keySet());
+ Map<String, List<String>> retrievedRoutingData = routingDataList.stream().collect(Collectors
+ .toMap(mapEntry -> (String) mapEntry
+ .get(MetadataStoreRoutingConstants.SINGLE_METADATA_STORE_REALM),
+ mapEntry -> (List<String>) mapEntry
+ .get(MetadataStoreRoutingConstants.SHARDING_KEYS)));
+ Assert.assertEquals(retrievedRoutingData, routingData);
+
+ // Send a GET request for all realms
+ getRequest = new HttpGet(endpoint + MockMetadataStoreDirectoryServer.REST_PREFIX + namespace
+ + MockMetadataStoreDirectoryServer.ZK_REALM_ENDPOINT);
+ getResponse = httpClient.execute(getRequest);
+ Map<String, Collection<String>> allRealmsMap = MockMetadataStoreDirectoryServer.OBJECT_MAPPER
+ .readValue(getResponse.getEntity().getContent(), Map.class);
+ Assert.assertTrue(
+ allRealmsMap.containsKey(MetadataStoreRoutingConstants.METADATA_STORE_REALMS));
+ allRealms = allRealmsMap.get(MetadataStoreRoutingConstants.METADATA_STORE_REALMS);
+ Assert.assertEquals(allRealms, routingData.keySet());
+
+ // Send a GET request for testZkRealm
+ String testZkRealm = "zk-0";
+ getRequest = new HttpGet(endpoint + MockMetadataStoreDirectoryServer.REST_PREFIX + namespace
+ + MockMetadataStoreDirectoryServer.ZK_REALM_ENDPOINT + "/" + testZkRealm);
+ getResponse = httpClient.execute(getRequest);
+ Map<String, Object> shardingKeysMap = MockMetadataStoreDirectoryServer.OBJECT_MAPPER
+ .readValue(getResponse.getEntity().getContent(), Map.class);
+ Assert.assertTrue(
+ shardingKeysMap.containsKey(MetadataStoreRoutingConstants.SINGLE_METADATA_STORE_REALM));
+ Assert.assertTrue(shardingKeysMap.containsKey(MetadataStoreRoutingConstants.SHARDING_KEYS));
+ String zkRealm =
+ (String) shardingKeysMap.get(MetadataStoreRoutingConstants.SINGLE_METADATA_STORE_REALM);
+ Collection<String> shardingKeyList =
+ (Collection) shardingKeysMap.get(MetadataStoreRoutingConstants.SHARDING_KEYS);
+ Assert.assertEquals(zkRealm, testZkRealm);
+ Assert.assertEquals(shardingKeyList, routingData.get(testZkRealm));
+
+ // Try sending a POST request (not supported)
+ HttpPost postRequest = new HttpPost(
+ endpoint + MockMetadataStoreDirectoryServer.REST_PREFIX + namespace
+ + MockMetadataStoreDirectoryServer.ZK_REALM_ENDPOINT + "/" + testZkRealm);
+ CloseableHttpResponse postResponse = httpClient.execute(postRequest);
+ } finally {
+ // Shutdown
+ server.stopServer();
+ }
+ }
+}
diff --git a/zookeeper-api/pom.xml b/zookeeper-api/pom.xml
index 91b448f..6d3e8c7 100644
--- a/zookeeper-api/pom.xml
+++ b/zookeeper-api/pom.xml
@@ -60,6 +60,11 @@ under the License.
</exclusions>
</dependency>
<dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>4.5.8</version>
+ </dependency>
+ <dependency>
<groupId>org.codehaus.jackson</groupId>
<artifactId>jackson-mapper-asl</artifactId>
<version>1.9.13</version>
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
new file mode 100644
index 0000000..b795de0
--- /dev/null
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/util/HttpRoutingDataReader.java
@@ -0,0 +1,149 @@
+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.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 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 */
+ private static volatile Map<String, List<String>> _rawRoutingData;
+ private static volatile MetadataStoreRoutingData _metadataStoreRoutingData;
+
+ /**
+ * This class is a Singleton.
+ */
+ private HttpRoutingDataReader() {
+ }
+
+ /**
+ * 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
+ */
+ public static Map<String, List<String>> getRawRoutingData() throws IOException {
+ if (MSDS_ENDPOINT == null || MSDS_ENDPOINT.isEmpty()) {
+ throw new IllegalStateException(
+ "HttpRoutingDataReader was unable to find a valid MSDS endpoint String in System Properties!");
+ }
+ if (_rawRoutingData == null) {
+ synchronized (HttpRoutingDataReader.class) {
+ if (_rawRoutingData == null) {
+ String routingDataJson = getAllRoutingData();
+ // Update the reference if reading routingData over HTTP is successful
+ _rawRoutingData = parseRoutingData(routingDataJson);
+ }
+ }
+ }
+ return _rawRoutingData;
+ }
+
+ /**
+ * Returns the routing data read from MSDS in a MetadataStoreRoutingData format.
+ * @return
+ * @throws IOException if there is an issue connecting to MSDS
+ * @throws InvalidRoutingDataException if the raw routing data is not valid
+ */
+ public static MetadataStoreRoutingData getMetadataStoreRoutingData()
+ throws IOException, InvalidRoutingDataException {
+ if (_metadataStoreRoutingData == null) {
+ synchronized (HttpRoutingDataReader.class) {
+ if (_metadataStoreRoutingData == null) {
+ _metadataStoreRoutingData = new TrieRoutingData(getRawRoutingData());
+ }
+ }
+ }
+ return _metadataStoreRoutingData;
+ }
+
+ /**
+ * Makes an HTTP call to fetch all routing data.
+ * @return
+ * @throws IOException
+ */
+ private static String getAllRoutingData() 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(
+ MSDS_ENDPOINT + 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/TestHttpRoutingDataReader.java
new file mode 100644
index 0000000..83ee471
--- /dev/null
+++ b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/util/TestHttpRoutingDataReader.java
@@ -0,0 +1,128 @@
+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.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.ImmutableSet;
+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.impl.ZkTestBase;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class TestHttpRoutingDataReader extends ZkTestBase {
+ private MockMetadataStoreDirectoryServer _msdsServer;
+ private Map<String, Collection<String>> _testRawRoutingData;
+ private final String _host = "localhost";
+ private final int _port = 1991;
+ private final String _namespace = "TestHttpRoutingDataReader";
+
+ @BeforeClass
+ public void beforeClass() throws IOException {
+ // Create fake routing data
+ _testRawRoutingData = new HashMap<>();
+ _testRawRoutingData
+ .put("zk-0", ImmutableSet.of("/sharding-key-0", "/sharding-key-1", "/sharding-key-2"));
+ _testRawRoutingData
+ .put("zk-1", ImmutableSet.of("/sharding-key-3", "/sharding-key-4", "/sharding-key-5"));
+ _testRawRoutingData
+ .put("zk-2", ImmutableSet.of("/sharding-key-6", "/sharding-key-7", "/sharding-key-8"));
+
+ // Start MockMSDS
+ _msdsServer =
+ new MockMetadataStoreDirectoryServer(_host, _port, _namespace, _testRawRoutingData);
+ _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);
+ }
+
+ @AfterClass
+ public void afterClass() {
+ _msdsServer.stopServer();
+ }
+
+ @Test
+ public void testGetRawRoutingData() throws IOException {
+ Map<String, List<String>> rawRoutingData = HttpRoutingDataReader.getRawRoutingData();
+ _testRawRoutingData.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();
+ Map<String, String> allMappings = data.getAllMappingUnderPath("/");
+ Map<String, Set<String>> groupedMappings = allMappings.entrySet().stream().collect(Collectors
+ .groupingBy(Map.Entry::getValue,
+ Collectors.mapping(Map.Entry::getKey, Collectors.toSet())));
+ _testRawRoutingData.forEach((realm, keys) -> {
+ Assert.assertEquals(groupedMappings.get(realm), new HashSet(keys));
+ });
+ }
+
+ /**
+ * Test that the static methods in HttpRoutingDataReader returns consistent results even though MSDS's data have been updated.
+ */
+ @Test(dependsOnMethods = "testGetMetadataStoreRoutingData")
+ public void testStaticMapping() throws IOException, InvalidRoutingDataException {
+ // Modify routing data
+ String newRealm = "newRealm";
+ _testRawRoutingData.put(newRealm, ImmutableSet.of("/newKey"));
+
+ // Kill MSDS and restart with a new mapping
+ _msdsServer.stopServer();
+ _msdsServer =
+ new MockMetadataStoreDirectoryServer(_host, _port, _namespace, _testRawRoutingData);
+ _msdsServer.startServer();
+
+ // 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();
+ Assert.assertFalse(rawRoutingData.containsKey(newRealm));
+
+ // Remove newRealm and check for equality
+ _testRawRoutingData.remove(newRealm);
+ Assert.assertEquals(rawRoutingData.keySet(), _testRawRoutingData.keySet());
+ _testRawRoutingData.forEach((realm, keys) -> Assert
+ .assertEquals(new HashSet(rawRoutingData.get(realm)), new HashSet(keys)));
+
+ MetadataStoreRoutingData data = HttpRoutingDataReader.getMetadataStoreRoutingData();
+ Map<String, String> allMappings = data.getAllMappingUnderPath("/");
+ Map<String, Set<String>> groupedMappings = allMappings.entrySet().stream().collect(Collectors
+ .groupingBy(Map.Entry::getValue,
+ Collectors.mapping(Map.Entry::getKey, Collectors.toSet())));
+ Assert.assertFalse(groupedMappings.containsKey(newRealm));
+ }
+}
diff --git a/zookeeper-api/zookeeper-api-0.9.2-SNAPSHOT.ivy b/zookeeper-api/zookeeper-api-0.9.2-SNAPSHOT.ivy
index 19a9ed8..2f7b6cb 100644
--- a/zookeeper-api/zookeeper-api-0.9.2-SNAPSHOT.ivy
+++ b/zookeeper-api/zookeeper-api-0.9.2-SNAPSHOT.ivy
@@ -46,5 +46,6 @@ under the License.
<dependency org="org.codehaus.jackson" name="jackson-core-asl" rev="1.8.5" force="true" conf="compile->compile(*),master(*);runtime->runtime(*)"/>
<dependency org="org.codehaus.jackson" name="jackson-mapper-asl" rev="1.8.5" force="true" conf="compile->compile(*),master(*);runtime->runtime(*)"/>
<dependency org="commons-cli" name="commons-cli" rev="1.2" force="true" conf="compile->compile(*),master(*);runtime->runtime(*)"/>
+ <dependency org="org.apache.httpcomponents" name="httpclient" rev="4.5.8" force="true" conf="compile->compile(*),master(*);runtime->runtime(*)"/>
</dependencies>
</ivy-module>