This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 64ae1e9cb feat(connectors): use TCP client for Flink sink (#3657)
64ae1e9cb is described below
commit 64ae1e9cbbb0e3c95285cf280b85836150b28e46
Author: Alan Tang <[email protected]>
AuthorDate: Fri Jul 24 20:09:14 2026 +0800
feat(connectors): use TCP client for Flink sink (#3657)
The Flink connector used different transports for reading and writing
messages. The source already used the native TCP client, while the sink
sent messages through the HTTP client and hardcoded the HTTP port to
3000.
This caused the same IggyConnectionConfig to behave differently between
the source and sink. It also required deployments to expose the HTTP API
even when the configured Iggy endpoint used the TCP port.
The source and sink also parsed server addresses independently. The
source relied on splitting the address by `:`, while the sink used URI
parsing. This produced inconsistent behavior and did not reliably
support IPv6 or validate malformed endpoints.
Closes #2484
---
foreign/java/.gitignore | 1 +
.../iggy-connector-library/README.md | 33 ++++++
.../apache/iggy/connector/config/TcpEndpoint.java | 92 +++++++++++++++
.../apache/iggy/connector/flink/sink/IggySink.java | 91 +++++++--------
.../iggy/connector/flink/sink/IggySinkWriter.java | 39 +++++--
.../iggy/connector/flink/source/IggySource.java | 24 ++--
.../iggy/connector/config/TcpEndpointTest.java | 130 +++++++++++++++++++++
7 files changed, 333 insertions(+), 77 deletions(-)
diff --git a/foreign/java/.gitignore b/foreign/java/.gitignore
index 2d92e8251..4ee6a2872 100644
--- a/foreign/java/.gitignore
+++ b/foreign/java/.gitignore
@@ -2,6 +2,7 @@
.project
.settings/
.gradle/
+.kotlin/
build/
out/
diff --git
a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/README.md
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/README.md
index 85e4712ac..3799b4cb8 100644
---
a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/README.md
+++
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/README.md
@@ -54,6 +54,39 @@ dependencies {
}
```
+### TCP Connection
+
+The Flink source and sink use Iggy's native TCP transport. Configure both with
+the same `IggyConnectionConfig`:
+
+```java
+IggyConnectionConfig connectionConfig = IggyConnectionConfig.builder()
+ .serverAddress("localhost:8090")
+ .username("iggy")
+ .password("iggy")
+ .connectionTimeout(Duration.ofSeconds(30))
+ .requestTimeout(Duration.ofSeconds(30))
+ .maxRetries(3)
+ .retryBackoff(Duration.ofMillis(100))
+ .enableTls(false)
+ .build();
+```
+
+The server address accepts the following formats:
+
+- `host`, using the default TCP port `8090`
+- `host:port`
+- `tcp://host:port`
+- bracketed IPv6, such as `tcp://[::1]:8090`
+
+Only TCP endpoints are accepted. HTTP URLs and addresses containing user info,
+paths, queries, or fragments are rejected during client creation. When TLS is
+enabled, the connector still uses the configured TCP endpoint.
+
+The sink uses the blocking TCP client because Flink's `SinkWriter` flush API is
+synchronous. Closing the writer flushes pending messages before closing the TCP
+connection.
+
### Building from Source
```bash
diff --git
a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/config/TcpEndpoint.java
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/config/TcpEndpoint.java
new file mode 100644
index 000000000..2586375c6
--- /dev/null
+++
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/config/TcpEndpoint.java
@@ -0,0 +1,92 @@
+/*
+ * 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.iggy.connector.config;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+/**
+ * Host and port of an Iggy TCP endpoint.
+ *
+ * @param host server host
+ * @param port server TCP port
+ */
+public record TcpEndpoint(String host, int port) {
+
+ public static final int DEFAULT_PORT = 8090;
+
+ /**
+ * Parses an Iggy TCP server address.
+ *
+ * @param serverAddress address in host, host:port, or tcp://host:port form
+ * @return parsed TCP endpoint
+ */
+ public static TcpEndpoint parse(String serverAddress) {
+ if (serverAddress == null || serverAddress.isBlank()) {
+ throw new IllegalArgumentException("TCP server address cannot be
null or blank");
+ }
+
+ URI uri = parseUri(serverAddress);
+ validateScheme(uri);
+ String host = extractHost(uri, serverAddress);
+ validateComponents(uri, serverAddress);
+
+ int port = uri.getPort() >= 0 ? uri.getPort() : DEFAULT_PORT;
+ validatePort(port, serverAddress);
+ return new TcpEndpoint(host, port);
+ }
+
+ private static URI parseUri(String serverAddress) {
+ try {
+ return serverAddress.contains("://") ? new URI(serverAddress) :
new URI("tcp://" + serverAddress);
+ } catch (URISyntaxException e) {
+ throw new IllegalArgumentException("Invalid TCP server address: "
+ serverAddress, e);
+ }
+ }
+
+ private static void validateScheme(URI uri) {
+ if (!"tcp".equalsIgnoreCase(uri.getScheme())) {
+ throw new IllegalArgumentException("Unsupported TCP server address
scheme: " + uri.getScheme());
+ }
+ }
+
+ private static String extractHost(URI uri, String serverAddress) {
+ String host = uri.getHost();
+ if (host == null || host.isBlank()) {
+ throw new IllegalArgumentException("Cannot extract host from TCP
server address: " + serverAddress);
+ }
+ return host;
+ }
+
+ private static void validateComponents(URI uri, String serverAddress) {
+ if (uri.getUserInfo() != null
+ || !uri.getPath().isEmpty()
+ || uri.getQuery() != null
+ || uri.getFragment() != null) {
+ throw new IllegalArgumentException("Invalid TCP server address: "
+ serverAddress);
+ }
+ }
+
+ private static void validatePort(int port, String serverAddress) {
+ if (port == 0 || port > 65535) {
+ throw new IllegalArgumentException("Invalid TCP port in server
address: " + serverAddress);
+ }
+ }
+}
diff --git
a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySink.java
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySink.java
index b8faf5251..7a70218eb 100644
---
a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySink.java
+++
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySink.java
@@ -22,32 +22,35 @@ package org.apache.iggy.connector.flink.sink;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.connector.sink2.WriterInitContext;
-import org.apache.iggy.client.blocking.http.IggyHttpClient;
+import org.apache.iggy.client.blocking.tcp.IggyTcpClient;
+import org.apache.iggy.config.RetryPolicy;
import org.apache.iggy.connector.config.IggyConnectionConfig;
+import org.apache.iggy.connector.config.TcpEndpoint;
import org.apache.iggy.connector.serialization.SerializationSchema;
import java.io.IOException;
import java.io.Serializable;
-import java.net.URISyntaxException;
import java.time.Duration;
/**
* Flink Sink implementation for writing to Iggy streams.
* Implements the Flink Sink V2 API for integration with DataStream API.
*
- * <p>Example usage:
+ * <p>
+ * Example usage:
+ *
* <pre>{@code
* events.sinkTo(
- * IggySink.<Event>builder()
- * .setConnectionConfig(connectionConfig)
- * .setStreamId("my-stream")
- * .setTopicId("my-topic")
- * .setSerializer(new JsonSerializationSchema<>())
- * .setBatchSize(100)
- * .setFlushInterval(Duration.ofSeconds(5))
- * .withBalancedPartitioning()
- * .build()
- * ).name("Iggy Sink");
+ * IggySink.<Event>builder()
+ * .setConnectionConfig(connectionConfig)
+ * .setStreamId("my-stream")
+ * .setTopicId("my-topic")
+ * .setSerializer(new JsonSerializationSchema<>())
+ * .setBatchSize(100)
+ * .setFlushInterval(Duration.ofSeconds(5))
+ * .withBalancedPartitioning()
+ * .build())
+ * .name("Iggy Sink");
* }</pre>
*
* @param <T> the type of records to write
@@ -68,12 +71,12 @@ public class IggySink<T> implements Sink<T>, Serializable {
* Creates a new Iggy sink.
* Use {@link #builder()} to construct instances.
*
- * @param connectionConfig the connection configuration
- * @param streamId the stream identifier
- * @param topicId the topic identifier
- * @param serializer the serialization schema
- * @param batchSize the batch size for buffering
- * @param flushInterval the maximum flush interval
+ * @param connectionConfig the connection configuration
+ * @param streamId the stream identifier
+ * @param topicId the topic identifier
+ * @param serializer the serialization schema
+ * @param batchSize the batch size for buffering
+ * @param flushInterval the maximum flush interval
* @param partitioningStrategy the partitioning strategy
*/
public IggySink(
@@ -106,46 +109,32 @@ public class IggySink<T> implements Sink<T>, Serializable
{
@Override
public SinkWriter<T> createWriter(WriterInitContext context) throws
IOException {
- IggyHttpClient httpClient = createHttpClient();
+ IggyTcpClient tcpClient = createTcpClient();
return new IggySinkWriter<>(
- httpClient, streamId, topicId, serializer, batchSize,
flushInterval, partitioningStrategy);
+ tcpClient, streamId, topicId, serializer, batchSize,
flushInterval, partitioningStrategy);
}
/**
- * Creates an HTTP Iggy client based on connection configuration.
+ * Creates a TCP Iggy client based on connection configuration.
*
- * @return configured HTTP Iggy client
+ * @return configured TCP Iggy client
*/
- private IggyHttpClient createHttpClient() {
+ private IggyTcpClient createTcpClient() {
try {
- // Build HTTP URL from server address using URI for proper parsing
- String serverAddress = connectionConfig.getServerAddress();
-
- // Parse server address to extract host
- java.net.URI uri = serverAddress.contains("://")
- ? new java.net.URI(serverAddress)
- : new java.net.URI("tcp://" + serverAddress);
-
- String host = uri.getHost();
- if (host == null) {
- throw new IllegalArgumentException("Cannot extract host from:
" + serverAddress);
- }
-
- // Build HTTP URL with port 3000 (Iggy HTTP API default port)
- String httpUrl = "http://"; + host + ":3000";
-
- // Create HTTP client
- IggyHttpClient httpClient = new IggyHttpClient(httpUrl);
-
- // Login
- httpClient.users().login(connectionConfig.getUsername(),
connectionConfig.getPassword());
-
- return httpClient;
-
- } catch (URISyntaxException e) {
- throw new RuntimeException("Invalid server address format: " +
connectionConfig.getServerAddress(), e);
+ TcpEndpoint endpoint =
TcpEndpoint.parse(connectionConfig.getServerAddress());
+
+ return IggyTcpClient.builder()
+ .host(endpoint.host())
+ .port(endpoint.port())
+ .credentials(connectionConfig.getUsername(),
connectionConfig.getPassword())
+ .connectionTimeout(connectionConfig.getConnectionTimeout())
+ .requestTimeout(connectionConfig.getRequestTimeout())
+ .retryPolicy(RetryPolicy.fixedDelay(
+ connectionConfig.getMaxRetries(),
connectionConfig.getRetryBackoff()))
+ .tls(connectionConfig.isEnableTls())
+ .buildAndLogin();
} catch (RuntimeException e) {
- throw new RuntimeException("Failed to create HTTP Iggy client", e);
+ throw new RuntimeException("Failed to create TCP Iggy client", e);
}
}
diff --git
a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java
index 244750b17..9e0383088 100644
---
a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java
+++
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java
@@ -21,7 +21,7 @@ package org.apache.iggy.connector.flink.sink;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.connector.sink2.SinkWriter;
-import org.apache.iggy.client.blocking.http.IggyHttpClient;
+import org.apache.iggy.client.blocking.tcp.IggyTcpClient;
import org.apache.iggy.connector.error.ConnectorException;
import org.apache.iggy.connector.serialization.SerializationSchema;
import org.apache.iggy.identifier.StreamId;
@@ -49,7 +49,7 @@ public class IggySinkWriter<T> implements SinkWriter<T> {
private static final Logger log =
LoggerFactory.getLogger(IggySinkWriter.class);
- private final IggyHttpClient httpClient;
+ private final IggyTcpClient tcpClient;
private final String streamId;
private final String topicId;
private final SerializationSchema<T> serializer;
@@ -73,7 +73,7 @@ public class IggySinkWriter<T> implements SinkWriter<T> {
/**
* Creates a new sink writer.
*
- * @param httpClient the HTTP Iggy client
+ * @param tcpClient the TCP Iggy client
* @param streamId the stream identifier
* @param topicId the topic identifier
* @param serializer the serialization schema
@@ -82,15 +82,15 @@ public class IggySinkWriter<T> implements SinkWriter<T> {
* @param partitioningStrategy the partitioning strategy
*/
public IggySinkWriter(
- IggyHttpClient httpClient,
+ IggyTcpClient tcpClient,
String streamId,
String topicId,
SerializationSchema<T> serializer,
int batchSize,
Duration flushInterval,
PartitioningStrategy partitioningStrategy) {
- if (httpClient == null) {
- throw new IllegalArgumentException("httpClient cannot be null");
+ if (tcpClient == null) {
+ throw new IllegalArgumentException("tcpClient cannot be null");
}
if (StringUtils.isBlank(streamId)) {
throw new IllegalArgumentException("streamId cannot be null or
empty");
@@ -108,7 +108,7 @@ public class IggySinkWriter<T> implements SinkWriter<T> {
throw new IllegalArgumentException("flushInterval must be
positive");
}
- this.httpClient = httpClient;
+ this.tcpClient = tcpClient;
this.streamId = streamId;
this.topicId = topicId;
this.serializer = serializer;
@@ -158,12 +158,11 @@ public class IggySinkWriter<T> implements SinkWriter<T> {
// Determine partitioning
Partitioning partitioning = determinePartitioning();
- // Send messages to Iggy via HTTP
StreamId stream = parseStreamId(streamId);
TopicId topic = parseTopicId(topicId);
log.debug("IggySinkWriter: Sending {} messages with
partitioning={}", messages.size(), partitioning);
- httpClient.messages().sendMessages(stream, topic, partitioning,
messages);
+ tcpClient.messages().sendMessages(stream, topic, partitioning,
messages);
totalWritten += buffer.size();
log.debug("IggySinkWriter: Successfully sent {} messages. Total
written: {}", buffer.size(), totalWritten);
@@ -180,9 +179,25 @@ public class IggySinkWriter<T> implements SinkWriter<T> {
@Override
public void close() throws Exception {
- // Flush any remaining buffered records
- flush(true);
- // Note: HTTP client doesn't have close() method - connections managed
by Java HttpClient pool
+ Exception flushException = null;
+ try {
+ flush(true);
+ } catch (IOException | RuntimeException e) {
+ flushException = e;
+ }
+
+ try {
+ tcpClient.close();
+ } catch (RuntimeException e) {
+ if (flushException == null) {
+ throw e;
+ }
+ flushException.addSuppressed(e);
+ }
+
+ if (flushException != null) {
+ throw flushException;
+ }
}
/**
diff --git
a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySource.java
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySource.java
index 365624125..82fead000 100644
---
a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySource.java
+++
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySource.java
@@ -27,8 +27,10 @@ import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
+import org.apache.iggy.config.RetryPolicy;
import org.apache.iggy.connector.config.IggyConnectionConfig;
import org.apache.iggy.connector.config.OffsetConfig;
+import org.apache.iggy.connector.config.TcpEndpoint;
import org.apache.iggy.consumergroup.Consumer;
import java.io.Serializable;
@@ -154,24 +156,18 @@ public class IggySource<T> implements Source<T,
IggySourceSplit, IggySourceEnume
*/
private AsyncIggyTcpClient createAsyncIggyClient() {
try {
- // Parse host and port from server address
- String serverAddress = connectionConfig.getServerAddress();
- String host;
- int port = 8090; // Default TCP port
-
- if (serverAddress.contains(":")) {
- String[] parts = serverAddress.split(":");
- host = parts[0];
- port = Integer.parseInt(parts[1]);
- } else {
- host = serverAddress;
- }
+ TcpEndpoint endpoint =
TcpEndpoint.parse(connectionConfig.getServerAddress());
// Create async TCP client using builder pattern with auto connect
and login
return AsyncIggyTcpClient.builder()
- .host(host)
- .port(port)
+ .host(endpoint.host())
+ .port(endpoint.port())
.credentials(connectionConfig.getUsername(),
connectionConfig.getPassword())
+ .connectionTimeout(connectionConfig.getConnectionTimeout())
+ .requestTimeout(connectionConfig.getRequestTimeout())
+ .retryPolicy(RetryPolicy.fixedDelay(
+ connectionConfig.getMaxRetries(),
connectionConfig.getRetryBackoff()))
+ .tls(connectionConfig.isEnableTls())
.connectionPoolSize(4)
.buildAndLogin()
.join();
diff --git
a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/config/TcpEndpointTest.java
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/config/TcpEndpointTest.java
new file mode 100644
index 000000000..785b44e4e
--- /dev/null
+++
b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/config/TcpEndpointTest.java
@@ -0,0 +1,130 @@
+/*
+ * 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.iggy.connector.config;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class TcpEndpointTest {
+
+ @Test
+ void shouldUseDefaultPortForHost() {
+ assertThat(TcpEndpoint.parse("localhost")).isEqualTo(new
TcpEndpoint("localhost", TcpEndpoint.DEFAULT_PORT));
+ }
+
+ @Test
+ void shouldUseDefaultPortForTcpUri() {
+ assertThat(TcpEndpoint.parse("tcp://iggy.example.com"))
+ .isEqualTo(new TcpEndpoint("iggy.example.com",
TcpEndpoint.DEFAULT_PORT));
+ }
+
+ @Test
+ void shouldParseHostAndPort() {
+ assertThat(TcpEndpoint.parse("iggy:8091")).isEqualTo(new
TcpEndpoint("iggy", 8091));
+ }
+
+ @Test
+ void shouldParseTcpUri() {
+ assertThat(TcpEndpoint.parse("tcp://iggy.example.com:8092"))
+ .isEqualTo(new TcpEndpoint("iggy.example.com", 8092));
+ }
+
+ @Test
+ void shouldParseIpv6Address() {
+ assertThat(TcpEndpoint.parse("tcp://[::1]:8093")).isEqualTo(new
TcpEndpoint("[::1]", 8093));
+ }
+
+ @Test
+ void shouldParseIpv6AddressWithoutScheme() {
+ assertThat(TcpEndpoint.parse("[2001:db8::1]:8094")).isEqualTo(new
TcpEndpoint("[2001:db8::1]", 8094));
+ }
+
+ @Test
+ void shouldRejectUnsupportedScheme() {
+ assertThatThrownBy(() -> TcpEndpoint.parse("http://localhost:3000";))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unsupported TCP server address scheme");
+ }
+
+ @Test
+ void shouldRejectBlankAddress() {
+ assertThatThrownBy(() -> TcpEndpoint.parse(" "))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("cannot be null or blank");
+ }
+
+ @Test
+ void shouldRejectNullAddress() {
+ assertThatThrownBy(() -> TcpEndpoint.parse(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("cannot be null or blank");
+ }
+
+ @Test
+ void shouldRejectMissingHost() {
+ assertThatThrownBy(() -> TcpEndpoint.parse("tcp://:8090"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot extract host");
+ }
+
+ @Test
+ void shouldRejectZeroPort() {
+ assertThatThrownBy(() -> TcpEndpoint.parse("localhost:0"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid TCP port");
+ }
+
+ @Test
+ void shouldRejectPortAboveMaximum() {
+ assertThatThrownBy(() -> TcpEndpoint.parse("localhost:65536"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid TCP port");
+ }
+
+ @Test
+ void shouldRejectUserInfo() {
+ assertThatThrownBy(() ->
TcpEndpoint.parse("tcp://user@localhost:8090"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid TCP server address");
+ }
+
+ @Test
+ void shouldRejectPath() {
+ assertThatThrownBy(() ->
TcpEndpoint.parse("tcp://localhost:8090/messages"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid TCP server address");
+ }
+
+ @Test
+ void shouldRejectQuery() {
+ assertThatThrownBy(() ->
TcpEndpoint.parse("tcp://localhost:8090?tls=true"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid TCP server address");
+ }
+
+ @Test
+ void shouldRejectFragment() {
+ assertThatThrownBy(() ->
TcpEndpoint.parse("tcp://localhost:8090#server"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid TCP server address");
+ }
+}
