feat(java): Add TCP transport support for Flink connector #2560

GaneshPatil7517 · 2026-01-14T13:11:12Z

Summary

Adds TCP transport protocol support to the Apache Iggy Flink connector, addressing the limitation where the sink only supported HTTP transport.

Resolves #2484

Changes

New Files

TransportType.java - Enum defining HTTP (port 3000) and TCP (port 8090) transport protocols
IggyClientFactory.java - Factory class for creating HTTP or TCP clients based on configuration

Modified Files

IggyConnectionConfig.java - Added transportType, httpPort, and tcpPort configuration options
IggySink.java - Updated to use IggyClientFactory instead of hardcoded HTTP client
IggySinkWriter.java - Changed to use IggyBaseClient interface, refactored validation
README.md - Added documentation for TCP transport configuration

Tests

TransportTypeTest.java - Unit tests for transport type enum
IggyClientFactoryTest.java - Unit tests for client factory
IggyConnectionConfigTest.java - Extended with transport configuration tests

Usage

// TCP Transport
IggyConnectionConfig config = IggyConnectionConfig.builder()
    .serverAddress("localhost")
    .transportType(TransportType.TCP)
    .tcpPort(8090)
    .username("iggy")
    .password("iggy")
    .build();

mmodzelewski · 2026-01-14T13:19:57Z

@GaneshPatil7517 Hey, I haven't reviewed the code fully yet, but seems like the redshift connector are unrelated to this change. Please clean the changes and let me know once this is ready to be reviewed.

GaneshPatil7517 · 2026-01-14T13:28:42Z

Hi @mmodzelewski, thanks for the quick review

You're right - the Redshift connector commits were accidentally included. I've cleaned up the branch and force-pushed. The PR now contains only the TCP transport support changes:

Single commit: feat(flink): add TCP transport support for Flink connector

Changes summary:

Added TransportType enum (HTTP/TCP)
Created IggyClientFactory for transport-agnostic client creation
Updated IggyConnectionConfig with transport settings
Modified sink to use IggyBaseClient interface
Added unit tests and documentation

All 137 tests pass locally. Ready for review!

hubcio · 2026-01-14T13:39:12Z

@GaneshPatil7517 your PR title is invalid, see https://github.com/apache/iggy/actions/runs/20995847599/job/60353117214?pr=2560 use java or connectors

Also fix other lints. Dont ping maintainers if CI is not passing for your PR.

GaneshPatil7517 · 2026-01-14T13:41:18Z

@GaneshPatil7517 your PR title is invalid, see https://github.com/apache/iggy/actions/runs/20995847599/job/60353117214?pr=2560 use java or connectors

Also fix other lints. Dont ping maintainers if CI is not passing for your PR.

ok ill fix it....

- Add TransportType enum with HTTP (default) and TCP options - Create IggyClientFactory for transport-agnostic client creation - Update IggyConnectionConfig with transport type and port settings - Modify IggySink/IggySinkWriter to use IggyBaseClient interface - Add comprehensive unit tests for new components - Update README with TCP configuration examples - Fix checkstyle NPath complexity in IggySinkWriter - Fix markdown table formatting in README Resolves apache#2484

ex172000 · 2026-01-15T06:05:20Z

foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/README.md

+
+| Transport            | Default Port | Use Case                                                     |
+| -------------------- | ------------ | ------------------------------------------------------------ |
+| **HTTP** (default)   | 3000         | Development, debugging, environments with HTTP proxies       |


Not sure if we should ask users to leverage production settings as much as possible? Thus making TCP the default makes more sense?

ex172000 · 2026-01-15T06:08:12Z

...y-connector-library/src/main/java/org/apache/iggy/connector/config/IggyConnectionConfig.java

+     *
+     * @return the port for the current transport type
+     */
+    public int getEffectivePort() {


I don't see it's used anywhere except the test, can we remove it or restrict its access to tests only?

Copilot

Pull request overview

This PR adds TCP transport protocol support to the Apache Iggy Flink connector, which previously only supported HTTP transport. The change provides users with a more performant option for high-throughput streaming workloads.

Changes:

Introduced TransportType enum to define HTTP and TCP transport options with default ports
Created IggyClientFactory to abstract client creation based on transport configuration
Extended IggyConnectionConfig with transport type, HTTP port, and TCP port configuration fields

Reviewed changes

Copilot reviewed 9 out of 9 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
TransportType.java	Defines transport protocol enum with HTTP (port 3000) and TCP (port 8090) options
IggyClientFactory.java	Factory class that creates appropriate client (HTTP/TCP) based on configuration
IggyConnectionConfig.java	Extended with transport type and port configuration options, including validation
IggySink.java	Updated to use `IggyClientFactory` instead of inline HTTP client creation
IggySinkWriter.java	Modified to use `IggyBaseClient` interface for transport flexibility and refactored validation logic
README.md	Added documentation for TCP transport configuration and usage examples
TransportTypeTest.java	Unit tests for transport type enum parsing and validation
IggyClientFactoryTest.java	Unit tests for host extraction from various server address formats
IggyConnectionConfigTest.java	Extended tests to cover new transport configuration options

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-01-15T07:36:10Z

...ggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java

+        // Close TCP client if applicable (TCP clients need explicit close)
+        if (iggyClient instanceof IggyTcpClient) {
+            log.debug("Closing TCP client connection");
+            // TCP client connections are managed internally and closed when JVM exits
+            // or when connection pool is explicitly closed
+        }
+        // HTTP client doesn't need explicit close - connections managed by Java HttpClient pool


The close() method checks if the client is a TCP client but doesn't actually call any close method on it. If TCP clients don't need explicit cleanup, this instanceof check and the empty block add unnecessary complexity. Consider removing the instanceof check entirely, or if TCP clients will need cleanup in the future, add a TODO comment explaining what needs to be done.

Suggested change

// Close TCP client if applicable (TCP clients need explicit close)

if (iggyClient instanceof IggyTcpClient) {

log.debug("Closing TCP client connection");

// TCP client connections are managed internally and closed when JVM exits

// or when connection pool is explicitly closed

}

// HTTP client doesn't need explicit close - connections managed by Java HttpClient pool

// Underlying Iggy clients manage their own connections; no explicit close required here.

Copilot · 2026-01-15T07:36:10Z

...ggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java

+
+    private static void validatePositiveDuration(Duration value, String name) {
+        if (value == null || value.isNegative()) {
+            throw new IllegalArgumentException(name + " must be positive");


The validation allows zero duration (Duration.ZERO), but the error message says 'must be positive'. A zero flush interval would mean immediate flushing which may not be intended. Either allow zero explicitly by adjusting the error message to 'must be non-negative', or check for zero and negative values with value.isNegative() || value.isZero() to enforce truly positive durations.

Suggested change

throw new IllegalArgumentException(name + " must be positive");

throw new IllegalArgumentException(name + " must be non-negative");

Copilot · 2026-01-15T07:36:11Z

.../iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/IggyClientFactory.java

+        if (address.contains(":")) {
+            String[] parts = address.split(":");
+            if (parts.length >= 1 && !parts[0].isBlank()) {
+                return parts[0];
+            }
+            throw new IllegalArgumentException("Cannot extract host from address: " + serverAddress);
+        }


The host extraction logic for addresses with colons can fail for IPv6 addresses (e.g., '[::1]:8090' or '2001:db8::1'). IPv6 addresses contain multiple colons which would cause incorrect splitting. Consider handling IPv6 addresses explicitly, either by checking for brackets or by using more sophisticated URI parsing for all cases.

…nstanceof check, fix error message, handle IPv6 addresses

github-actions · 2026-01-30T16:56:50Z

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs.

If you need a review, please ensure CI is green and the PR is rebased on the latest master. Don't hesitate to ping the maintainers - either @core on Discord or by mentioning them directly here on the PR.

Thank you for your contribution!

github-actions · 2026-02-07T00:36:25Z

This pull request was automatically closed because it has been inactive for 14 days. Feel free to reopen it if you'd like to continue working on it.

GaneshPatil7517 force-pushed the feature/tcp-transport-flink-connector branch from bee34ee to 964d07c Compare January 14, 2026 13:27

GaneshPatil7517 force-pushed the feature/tcp-transport-flink-connector branch from f3e3991 to 6b46df4 Compare January 14, 2026 14:10

GaneshPatil7517 changed the title ~~feat(flink): Add TCP transport support for Flink connector~~ feat(java): Add TCP transport support for Flink connector Jan 14, 2026

GaneshPatil7517 added 2 commits January 14, 2026 19:43

Merge branch 'master' into feature/tcp-transport-flink-connector

ec57787

Merge branch 'master' into feature/tcp-transport-flink-connector

0d8cda6

ex172000 reviewed Jan 15, 2026

View reviewed changes

fix: Add trailing newline and fix markdown table alignment

aa210d6

Copilot AI review requested due to automatic review settings January 15, 2026 07:29

Copilot started reviewing on behalf of GaneshPatil7517 January 15, 2026 07:30 View session

Copilot AI reviewed Jan 15, 2026

View reviewed changes

refactor: Implement AI code review suggestions - remove unnecessary i…

bc5ed4f

…nstanceof check, fix error message, handle IPv6 addresses

hubcio closed this Jan 21, 2026

hubcio reopened this Jan 21, 2026

Merge branch 'master' into feature/tcp-transport-flink-connector

4b0f513

github-actions bot added the stale Inactive issue or pull request label Jan 30, 2026

github-actions bot closed this Feb 7, 2026

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(java): Add TCP transport support for Flink connector #2560

feat(java): Add TCP transport support for Flink connector #2560

Uh oh!

GaneshPatil7517 commented Jan 14, 2026 •

edited

Loading

Uh oh!

mmodzelewski commented Jan 14, 2026

Uh oh!

GaneshPatil7517 commented Jan 14, 2026

Uh oh!

hubcio commented Jan 14, 2026 •

edited

Loading

Uh oh!

GaneshPatil7517 commented Jan 14, 2026

Uh oh!

ex172000 Jan 15, 2026

Uh oh!

ex172000 Jan 15, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Jan 15, 2026

Uh oh!

Copilot AI Jan 15, 2026

Uh oh!

Copilot AI Jan 15, 2026

Uh oh!

github-actions bot commented Jan 30, 2026

Uh oh!

github-actions bot commented Feb 7, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

	throw new IllegalArgumentException(name + " must be positive");
	throw new IllegalArgumentException(name + " must be non-negative");

feat(java): Add TCP transport support for Flink connector #2560

feat(java): Add TCP transport support for Flink connector #2560

Uh oh!

Conversation

GaneshPatil7517 commented Jan 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Changes

New Files

Modified Files

Tests

Usage

Uh oh!

mmodzelewski commented Jan 14, 2026

Uh oh!

GaneshPatil7517 commented Jan 14, 2026

Uh oh!

hubcio commented Jan 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

GaneshPatil7517 commented Jan 14, 2026

Uh oh!

ex172000 Jan 15, 2026

Choose a reason for hiding this comment

Uh oh!

ex172000 Jan 15, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Jan 15, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 15, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 15, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented Jan 30, 2026

Uh oh!

github-actions bot commented Feb 7, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

GaneshPatil7517 commented Jan 14, 2026 •

edited

Loading

hubcio commented Jan 14, 2026 •

edited

Loading