feat(connectors): use TCP client for Flink sink

foreign/java/.gitignore

@@ -2,6 +2,7 @@
 .project
 .settings/
 .gradle/
+.kotlin/
 build/
 out/
 

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

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);
+        }
+    }
+}

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.
  *
- * <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 static <T> IggySinkBuilder<T> builder() {
 
     @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);
         }
     }
 

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<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 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<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 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;
+        }
     }
 
     /**