This is an automated email from the ASF dual-hosted git repository.
zhoujinsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 9e63e8ebd [AMORO-4232][common] Improve AmsThriftUrl parsing and
validation (#4372)
9e63e8ebd is described below
commit 9e63e8ebda0f9664b3efa1b4075a2ec695609893
Author: Sivakumar Mahalingam
<[email protected]>
AuthorDate: Tue Sep 22 10:30:48 2026 +0400
[AMORO-4232][common] Improve AmsThriftUrl parsing and validation (#4372)
fix: improve AmsThriftUrl parsing and validation
---
.../java/org/apache/amoro/client/AmsThriftUrl.java | 91 ++++++++++------
.../org/apache/amoro/client/TestAmsThriftUrl.java | 115 +++++++++++++++++++++
2 files changed, 173 insertions(+), 33 deletions(-)
diff --git
a/amoro-common/src/main/java/org/apache/amoro/client/AmsThriftUrl.java
b/amoro-common/src/main/java/org/apache/amoro/client/AmsThriftUrl.java
index 860d5c6dc..238a08a24 100644
--- a/amoro-common/src/main/java/org/apache/amoro/client/AmsThriftUrl.java
+++ b/amoro-common/src/main/java/org/apache/amoro/client/AmsThriftUrl.java
@@ -34,7 +34,6 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
-import java.util.Locale;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -44,8 +43,12 @@ public class AmsThriftUrl {
public static final String ZOOKEEPER_FLAG = "zookeeper";
public static final String THRIFT_URL_FORMAT = "thrift://%s:%d/%s%s";
public static final int MAX_RETRIES = 3;
+ private static final String THRIFT_SCHEME = "thrift";
+ private static final String ZOOKEEPER_SCHEME_PREFIX = ZOOKEEPER_FLAG + ":";
+ private static final int MAX_PORT = 65535;
private static final Logger logger =
LoggerFactory.getLogger(AmsThriftUrl.class);
- private static final Pattern PATTERN =
Pattern.compile("zookeeper://(\\S+)/([\\w-]+)");
+ private static final Pattern PATTERN =
+ Pattern.compile("zookeeper://(\\S+)/([\\w-]+)",
Pattern.CASE_INSENSITIVE);
private final String schema;
private final String host;
private final int port;
@@ -77,7 +80,7 @@ public class AmsThriftUrl {
if (url == null) {
throw new IllegalArgumentException("thrift url is null");
}
- if (url.startsWith(ZOOKEEPER_FLAG)) {
+ if (url.regionMatches(true, 0, ZOOKEEPER_SCHEME_PREFIX, 0,
ZOOKEEPER_SCHEME_PREFIX.length())) {
return parserZookeeperUrl(url, serviceName);
} else {
return parserThriftUrl(url);
@@ -85,31 +88,57 @@ public class AmsThriftUrl {
}
private static AmsThriftUrl parserThriftUrl(String url) {
- int socketTimeout = DEFAULT_SOCKET_TIMEOUT;
try {
- URI uri = new URI(url.toLowerCase(Locale.ROOT));
+ URI uri = new URI(url);
String schema = uri.getScheme();
+ if (!THRIFT_SCHEME.equalsIgnoreCase(schema)) {
+ throw new IllegalArgumentException(
+ String.format("Unsupported thrift URL scheme '%s' in URL: %s",
schema, url));
+ }
+
+ uri = uri.parseServerAuthority();
String host = uri.getHost();
+ if (host == null || host.isEmpty()) {
+ throw new IllegalArgumentException(String.format("Missing host in
thrift URL: %s", url));
+ }
+
int port = uri.getPort();
+ if (port == -1) {
+ throw new IllegalArgumentException(String.format("Missing port in
thrift URL: %s", url));
+ } else if (port <= 0 || port > MAX_PORT) {
+ throw new IllegalArgumentException(
+ String.format("Invalid port in thrift URL, expected 1-%d: %s",
MAX_PORT, url));
+ }
+
String path = uri.getPath();
if (path != null && path.startsWith("/")) {
path = path.substring(1);
}
- if (uri.getQuery() != null) {
- for (String paramExpression : uri.getQuery().split("&")) {
- String[] paramSplit = paramExpression.split("=");
- if (paramSplit.length == 2) {
- if (paramSplit[0].equalsIgnoreCase(PARAM_SOCKET_TIMEOUT)) {
- socketTimeout = Integer.parseInt(paramSplit[1]);
- }
+ String catalogName = path;
+ return new AmsThriftUrl(
+ THRIFT_SCHEME, host, port, catalogName,
parseSocketTimeout(uri.getQuery()), url);
+ } catch (URISyntaxException e) {
+ throw new IllegalArgumentException(
+ String.format("Invalid thrift URL '%s': %s", url, e.getMessage()),
e);
+ }
+ }
+
+ private static int parseSocketTimeout(String query) {
+ int socketTimeout = DEFAULT_SOCKET_TIMEOUT;
+ if (query != null) {
+ for (String paramExpression : query.split("&")) {
+ String[] paramSplit = paramExpression.split("=");
+ if (paramSplit.length == 2 &&
paramSplit[0].equalsIgnoreCase(PARAM_SOCKET_TIMEOUT)) {
+ try {
+ socketTimeout = Integer.parseInt(paramSplit[1]);
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ String.format("Invalid socketTimeout value '%s'",
paramSplit[1]), e);
}
}
}
- String catalogName = path;
- return new AmsThriftUrl(schema, host, port, catalogName, socketTimeout,
url);
- } catch (URISyntaxException e) {
- throw new IllegalArgumentException("parse metastore url failed", e);
}
+ return socketTimeout;
}
private static AmsThriftUrl parserZookeeperUrl(String url, String
serviceName) {
@@ -132,33 +161,25 @@ public class AmsThriftUrl {
zkServerAddress = m.group(1);
cluster = m.group(2);
}
+ int socketTimeout = parseSocketTimeout(query.replace("?", ""));
int retryCount = 0;
while (retryCount < MAX_RETRIES) {
try {
AmsServerInfo serverInfo = findAmsServerInfo(serviceName,
zkServerAddress, cluster);
- url =
+ String resolvedUrl =
String.format(
THRIFT_URL_FORMAT,
serverInfo.getHost(),
serverInfo.getThriftBindPort(),
catalog,
query);
- int socketTimeout = DEFAULT_SOCKET_TIMEOUT;
- for (String paramExpression : query.replace("?", "").split("&")) {
- String[] paramSplit = paramExpression.split("=");
- if (paramSplit.length == 2) {
- if (paramSplit[0].equalsIgnoreCase(PARAM_SOCKET_TIMEOUT)) {
- socketTimeout = Integer.parseInt(paramSplit[1]);
- }
- }
- }
return new AmsThriftUrl(
- "thrift",
+ THRIFT_SCHEME,
serverInfo.getHost(),
serverInfo.getThriftBindPort(),
- catalog.toLowerCase(),
+ catalog,
socketTimeout,
- url);
+ resolvedUrl);
} catch (KeeperException.AuthFailedException authFailedException) {
// If kerberos authentication is not enabled on the zk,
// an error occurs when the thread carrying kerberos authentication
information accesses
@@ -182,11 +203,14 @@ public class AmsThriftUrl {
retryCount++;
logger.error(
String.format("Caught exception, retrying... (retry count: %s)",
retryCount), e);
- throw new RuntimeException(String.format("invalid ams url %s", url));
+ throw new RuntimeException(
+ String.format("Failed to resolve AMS URL from ZooKeeper URL:
%s", url), e);
}
}
} else {
- throw new RuntimeException(String.format("invalid ams url %s", url));
+ throw new IllegalArgumentException(
+ String.format(
+ "Invalid ZooKeeper URL, expected
zookeeper://host:port/cluster[/catalog]: %s", url));
}
return null;
}
@@ -217,7 +241,7 @@ public class AmsThriftUrl {
}
private static List<AmsServerInfo>
parserZookeeperUrlListForMasterSlaveMode(String url) {
- if (!url.startsWith(ZOOKEEPER_FLAG)) {
+ if (!url.regionMatches(true, 0, ZOOKEEPER_SCHEME_PREFIX, 0,
ZOOKEEPER_SCHEME_PREFIX.length())) {
throw new IllegalArgumentException(
"parseMasterSlaveAmsNodes only supports ZooKeeper URL format:
zookeeper://host:port/cluster");
}
@@ -227,7 +251,8 @@ public class AmsThriftUrl {
}
Matcher m = PATTERN.matcher(thriftUrl);
if (!m.matches()) {
- throw new RuntimeException(String.format("invalid ams url %s", url));
+ throw new IllegalArgumentException(
+ String.format("Invalid ZooKeeper URL, expected
zookeeper://host:port/cluster: %s", url));
}
String zkServerAddress;
String cluster;
diff --git
a/amoro-common/src/test/java/org/apache/amoro/client/TestAmsThriftUrl.java
b/amoro-common/src/test/java/org/apache/amoro/client/TestAmsThriftUrl.java
new file mode 100644
index 000000000..55e57a329
--- /dev/null
+++ b/amoro-common/src/test/java/org/apache/amoro/client/TestAmsThriftUrl.java
@@ -0,0 +1,115 @@
+/*
+ * 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.
+ */
+
+package org.apache.amoro.client;
+
+import org.apache.amoro.Constants;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestAmsThriftUrl {
+ @Test
+ public void testParseMixedCaseThriftUrl() {
+ String url = "ThRiFt://LOCALHOST:1260/MyCatalog?socketTimeout=6000";
+
+ AmsThriftUrl thriftUrl = AmsThriftUrl.parse(url, null);
+
+ Assertions.assertEquals("thrift", thriftUrl.schema());
+ Assertions.assertEquals("LOCALHOST", thriftUrl.host());
+ Assertions.assertEquals(1260, thriftUrl.port());
+ Assertions.assertEquals("MyCatalog", thriftUrl.catalogName());
+ Assertions.assertEquals(6000, thriftUrl.socketTimeout());
+ Assertions.assertEquals(url, thriftUrl.url());
+ }
+
+ @Test
+ public void testParseThriftUrl() {
+ AmsThriftUrl thriftUrl =
+ AmsThriftUrl.parse(
+ "thrift://127.0.0.1:1260/catalog?socketTimeout=7000",
+ Constants.THRIFT_TABLE_SERVICE_NAME);
+
+ Assertions.assertEquals("127.0.0.1", thriftUrl.host());
+ Assertions.assertEquals(1260, thriftUrl.port());
+ Assertions.assertEquals("catalog", thriftUrl.catalogName());
+ Assertions.assertEquals(7000, thriftUrl.socketTimeout());
+ }
+
+ @Test
+ public void testRejectUnsupportedScheme() {
+ assertInvalidUrl("http://127.0.0.1:1260/catalog";, "scheme");
+ assertInvalidUrl("zookeeperx://127.0.0.1:2181/cluster", "scheme");
+ }
+
+ @Test
+ public void testRejectMissingHost() {
+ assertInvalidUrl("thrift:///catalog", "host");
+ }
+
+ @Test
+ public void testRejectMissingPort() {
+ assertInvalidUrl("thrift://127.0.0.1/catalog", "port");
+ }
+
+ @Test
+ public void testRejectInvalidPort() {
+ assertInvalidUrl("thrift://127.0.0.1:65536/catalog", "port");
+ }
+
+ @Test
+ public void testRejectInvalidZookeeperUrl() {
+ assertInvalidUrl("zookeeper:/127.0.0.1:2181/cluster", "ZooKeeper URL");
+ }
+
+ @Test
+ public void testParseZookeeperUrlBeforeServiceResolution() {
+ String url =
"ZoOkEePeR://127.0.0.1:2181/test-cluster/MyCatalog?socketTimeout=8000";
+
+ RuntimeException exception =
+ Assertions.assertThrows(
+ RuntimeException.class, () -> AmsThriftUrl.parse(url,
"unsupported-service"));
+
+ Assertions.assertTrue(
+ exception.getMessage().contains("Failed to resolve AMS URL from
ZooKeeper URL"),
+ exception.getMessage());
+ Assertions.assertNotNull(exception.getCause());
+ Assertions.assertTrue(
+ exception.getCause().getMessage().contains("invalid service name
unsupported-service"),
+ exception.getCause().getMessage());
+ }
+
+ @Test
+ public void testRejectInvalidZookeeperSocketTimeout() {
+ String url =
"ZoOkEePeR://127.0.0.1:2181/test-cluster/MyCatalog?socketTimeout=invalid";
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> AmsThriftUrl.parse(url,
Constants.THRIFT_TABLE_SERVICE_NAME));
+
+ Assertions.assertTrue(
+ exception.getMessage().contains("Invalid socketTimeout value"),
exception.getMessage());
+ }
+
+ private void assertInvalidUrl(String url, String messagePart) {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> AmsThriftUrl.parse(url,
null));
+ Assertions.assertTrue(exception.getMessage().contains(messagePart),
exception.getMessage());
+ }
+}
