C0urante commented on a change in pull request #10530:
URL: https://github.com/apache/kafka/pull/10530#discussion_r613339599
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java
##########
@@ -369,9 +370,31 @@ else if (serverConnector != null &&
serverConnector.getHost() != null && serverC
else if (serverConnector != null && serverConnector.getPort() > 0)
builder.port(serverConnector.getPort());
- log.info("Advertised URI: {}", builder.build());
+ URI uri = builder.build();
+ validateUriHost(uri);
+ log.info("Advertised URI: {}", uri);
- return builder.build();
+ return uri;
+ }
+
+ /**
+ * Parses the uri and throws a more definitive error
+ * when the internal node to node communication can't happen due to an
invalid host name.
+ */
+ private void validateUriHost(URI uri) {
+ if (uri.getHost() == null) {
+ String host = Utils.getHost(uri.getAuthority());
+ String errorMsg = "Invalid host=" + host + ", in url=" +
uri.toString();
Review comment:
Yeah, changing to a new REST client is probably fine as a follow-up
ticket 👍
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java
##########
@@ -185,6 +188,36 @@ public void testAdvertisedUri() {
Assert.assertEquals("http://plaintext-localhost:4761/";,
server.advertisedUrl().toString());
}
+ @Test
+ public void testValidateUriHost() {
+ Map<String, String> configMap = new HashMap<>(baseWorkerProps());
+ configMap.put(WorkerConfig.LISTENERS_CONFIG,
"http://localhost:8080,https://localhost:8443";);
+ DistributedConfig config = new DistributedConfig(configMap);
+
+ server = new RestServer(config);
+ server.validateUriHost(URI.create("http://localhost:8080";));
+ server.validateUriHost(URI.create("http://172.217.2.110:80";));
+
server.validateUriHost(URI.create("http://[2607:f8b0:4006:818::2004]:80";));
+ }
+
+ @Test
+ public void testValidateUriInvalidHost() {
+ Map<String, String> configMap = new HashMap<>(baseWorkerProps());
+ configMap.put(WorkerConfig.LISTENERS_CONFIG,
"http://localhost:8080,https://localhost:8443";);
+ DistributedConfig config = new DistributedConfig(configMap);
+
+ server = new RestServer(config);
+
+ ConnectException exception = assertThrows(ConnectException.class, ()
-> server.validateUriHost(URI.create("http://kafka_connect-0.dev-2:8080";)));
+ assertTrue(exception.getMessage().contains("RFC 1123"));
+ //invalid uri with / in the end
Review comment:
Nice case!
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java
##########
@@ -185,6 +188,36 @@ public void testAdvertisedUri() {
Assert.assertEquals("http://plaintext-localhost:4761/";,
server.advertisedUrl().toString());
}
+ @Test
+ public void testValidateUriHost() {
+ Map<String, String> configMap = new HashMap<>(baseWorkerProps());
+ configMap.put(WorkerConfig.LISTENERS_CONFIG,
"http://localhost:8080,https://localhost:8443";);
+ DistributedConfig config = new DistributedConfig(configMap);
+
+ server = new RestServer(config);
Review comment:
Nit: seems a little strange to have to instantiate a `RestServer` with
one set of REST listeners and no advertised listeners, but then use it to
validate a different set of advertised listeners. Would it make sense to change
`RestServer::validateUriHost` to a `static` method to make the testing here
cleaner?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
