diff --git a/foreign/java/.gitignore b/foreign/java/.gitignore
index 2d92e82516..4ee6a28721 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 85e4712acf..3799b4cb84 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 0000000000..2586375c6b
--- /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 b8faf5251a..7a70218eb7 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 @@
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.
*
- * Example usage:
+ *
+ * Example usage:
+ *
*
{@code
* events.sinkTo(
- * IggySink.builder()
- * .setConnectionConfig(connectionConfig)
- * .setStreamId("my-stream")
- * .setTopicId("my-topic")
- * .setSerializer(new JsonSerializationSchema<>())
- * .setBatchSize(100)
- * .setFlushInterval(Duration.ofSeconds(5))
- * .withBalancedPartitioning()
- * .build()
- * ).name("Iggy Sink");
+ * IggySink.builder()
+ * .setConnectionConfig(connectionConfig)
+ * .setStreamId("my-stream")
+ * .setTopicId("my-topic")
+ * .setSerializer(new JsonSerializationSchema<>())
+ * .setBatchSize(100)
+ * .setFlushInterval(Duration.ofSeconds(5))
+ * .withBalancedPartitioning()
+ * .build())
+ * .name("Iggy Sink");
* }
*
* @param the type of records to write
@@ -68,12 +71,12 @@ public class IggySink implements Sink, 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 static IggySinkBuilder builder() {
@Override
public SinkWriter 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 244750b17c..5a0b6238d0 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 @@
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 implements SinkWriter {
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 serializer;
@@ -73,7 +73,7 @@ public enum PartitioningStrategy {
/**
* 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 enum PartitioningStrategy {
* @param partitioningStrategy the partitioning strategy
*/
public IggySinkWriter(
- IggyHttpClient httpClient,
+ IggyTcpClient tcpClient,
String streamId,
String topicId,
SerializationSchema 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 IggySinkWriter(
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 void flush(boolean endOfInput) throws IOException {
// 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 void flush(boolean endOfInput) throws IOException {
@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
+ IOException flushException = null;
+ try {
+ flush(true);
+ } catch (IOException 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 3656241258..50c25eff6b 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
@@ -29,6 +29,7 @@
import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
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,23 +155,12 @@ public SimpleVersionedSerializer getEnumeratorCheckpo
*/
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())
.connectionPoolSize(4)
.buildAndLogin()
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 0000000000..785b44e4e9
--- /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");
+ }
+}