feat(java): implement Async Connection Pooling using FixedChannelPool by rythm-sachdeva · Pull Request #2606 · apache/iggy

rythm-sachdeva · 2026-01-24T09:23:48Z

Description

Implements connection pooling for AsyncTcpConnection using Netty's FixedChannelPool to handle concurrent async operations efficiently.

Changes

Connection Pooling: Replaced single channel with FixedChannelPool for concurrent request handling
Pool Metrics: Added PoolMetrics class to track:
- Active connections
- Total requests and errors
- Wait times (min/max/average in nanoseconds, microseconds, and milliseconds)
- Error rate percentage
Configuration: Added TCPConnectionPoolConfig with builder pattern for customizing:
- Max connections (default: 5)
- Max pending acquires (default: 1000)
- Acquire timeout (default: 3000ms)
Request Correlation: Implemented FIFO queue-based request/response matching using ConcurrentLinkedQueue
Broadcast Operations: Added broadcastAsync() method to execute commands across all pooled connections (useful for authentication)
Client Integration: Exposed pool configuration via AsyncIggyTcpClient.Builder.connectionPoolSize()

Implementation Details

Uses Netty's FixedChannelPool with ChannelHealthChecker.ACTIVE
Each channel has its own IggyResponseHandler with a response queue
Pool automatically releases channels after request completion
Handles connection errors gracefully with proper cleanup

Testing

Added integration test testConnectionPoolMetrics() to verify metrics tracking
Existing async integration tests validate concurrent operations
Pool properly handles concurrent sends and polls

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java

github-actions · 2026-02-04T00:34:55Z

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!

rythm-sachdeva · 2026-02-09T14:21:08Z

Hi @mmodzelewski Can you please review .

mmodzelewski · 2026-02-09T21:20:40Z

Hi @mmodzelewski Can you please review .

Hey @rythm-sachdeva, I will, but I need some time. I might not get round to it until the beginning of next week.

slbotbm · 2026-02-17T17:46:03Z

@rythm-sachdeva Please see why CI is failing and fix the errors.

mmodzelewski · 2026-02-17T18:17:26Z

@rythm-sachdeva Please see why CI is failing and fix the errors.

@rythm-sachdeva I've added a verification if all ByteBufs are properly released (not leaked) and it appears that the new code does not release all resources. See the logs, or run the tests locally, you'll see reports of the leaks in there.

codecov · 2026-02-22T08:58:29Z

Codecov Report

❌ Patch coverage is 57.81991% with 89 lines in your changes missing coverage. Please review.
✅ Project coverage is 68.50%. Comparing base (aff29a5) to head (181682b).

Files with missing lines	Patch %	Lines
...ache/iggy/client/async/tcp/AsyncTcpConnection.java	60.86%	40 Missing and 14 partials ⚠️
.../org/apache/iggy/client/async/tcp/PoolMetrics.java	45.28%	26 Missing and 3 partials ⚠️
...g/apache/iggy/client/async/tcp/UsersTcpClient.java	50.00%	6 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #2606      +/-   ##
============================================
- Coverage     68.52%   68.50%   -0.02%     
- Complexity      656      673      +17     
============================================
  Files           743      744       +1     
  Lines         62808    62934     +126     
  Branches      59221    59235      +14     
============================================
+ Hits          43039    43114      +75     
- Misses        17656    17700      +44     
- Partials       2113     2120       +7

Flag	Coverage Δ
java	`52.42% <57.81%> (+0.23%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...ache/iggy/client/async/tcp/AsyncIggyTcpClient.java	`98.11% <100.00%> (+0.28%)`	⬆️
...g/apache/iggy/client/async/tcp/UsersTcpClient.java	`73.33% <50.00%> (-3.59%)`	⬇️
.../org/apache/iggy/client/async/tcp/PoolMetrics.java	`45.28% <45.28%> (ø)`
...ache/iggy/client/async/tcp/AsyncTcpConnection.java	`61.62% <60.86%> (+3.44%)`	⬆️

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

mmodzelewski

Hey @rythm-sachdeva, I left a few comments, please let me know if you have any questions.

mmodzelewski · 2026-02-24T11:03:28Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java

    public CompletableFuture<Void> connect() {
-        connection = new AsyncTcpConnection(host, port, enableTls, tlsCertificate);
+        TCPConnectionPoolConfig.Builder poolConfigBuilder = new TCPConnectionPoolConfig.Builder();
+        if (connectionPoolSize.isPresent()) {


instead of isPresent/get, it would be more idiomatic to just call ifPresent with an update lambda

mmodzelewski · 2026-02-24T11:07:00Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java


    public AsyncTcpConnection(String host, int port) {
-        this(host, port, false, Optional.empty());
+        this(host, port, false, Optional.empty(), new TCPConnectionPoolConfig(5, 1000, 1000));


Please create a no-args constructor for the config, and use it here instead.

mmodzelewski · 2026-02-24T11:08:21Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java

+        public static final class Builder {
+            private int maxConnections = 5;
+            private int maxPendingAcquires = 1000;
+            private long acquireTimeoutMillis = 5000;


assign above values to constants, and use them in no-args constructor for the config

mmodzelewski · 2026-02-24T11:10:13Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java

        }
    }
+
+    // Inner Class for Channel pool configurations


Please drop all comments that do not introduce any new information. Comments are justified in situations where code is not obvious, something is done in a non-standard way, etc.

mmodzelewski · 2026-02-24T11:16:09Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java

+     */
+    public CompletableFuture<Void> connect() {
+        if (isClosed.get()) {
+            return CompletableFuture.failedFuture(new IllegalStateException("Client is Closed"));


in the SDK code, please only use exceptions that inherit from IggyException

mmodzelewski · 2026-02-24T19:04:04Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java

+            poolConfigBuilder.setMaxConnections(connectionPoolSize.get());
+        }
+        if (connectionTimeout.isPresent()) {
+            poolConfigBuilder.setAcquireTimeoutMillis(connectionTimeout.get().toMillis());


seems like connection timeout should map to CONNECT_TIMEOUT_MILLIS, acquire timeout is semantically different thing

I think by mistake I have given the wrong name it should be acquireTimeout

mmodzelewski · 2026-02-24T19:06:03Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PoolMetrics.java

+ * Metrics for monitoring connection pool performance.
+ * Tracks active connections, wait times, and errors.
+ */
+public class PoolMetrics {


I'd drop the metrics for now. Let's make sure that the connection pool is working fine first. This can be a follow up.

mmodzelewski · 2026-02-24T19:09:28Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java

+            if (channelPool != null) {
+                channelPool.close();
+            }
+            return CompletableFuture.runAsync(eventLoopGroup::shutdownGracefully);


It would make sense to wait for the result of shutdownGracefully. Currently, it is ignored.

mmodzelewski · 2026-02-24T19:10:54Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java

+
+            public Builder() {}
+
+            public Builder setMaxConnections(int maxConnections) {


let's at least add some simple validations for the values inserted to the config

mmodzelewski · 2026-02-24T19:23:07Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java

+     * Returns the result of the LAST connection's execution, allowing the caller
+     * to treat this like a single request.
+     */
+    public CompletableFuture<ByteBuf> broadcastAsync(int commandCode, ByteBuf payload) {


broadcastAsync() for authentication is unreliable - FixedChannelPool creates channels lazily, so sending poolSize login requests may hit the same channel multiple times (it gets released in channelRead0 and re-acquired) rather than reaching every distinct connection. Connections created later (e.g. reconnect after health-check failure) will also be unauthenticated.

A simpler approach: authenticate lazily in send(). After acquiring a channel, check a channel attribute (e.g. channel.attr(AUTH_KEY)). If not authenticated, send login first, set the attribute on success, then send the actual command. This works naturally with the async model, handles all channel lifecycle events (pool growth, reconnection, idle eviction), and eliminates the need for broadcastAsync() entirely.

Currently the login is called by the user once After client creation. In this case how would the flow work ?

The user-facing API stays the same - login() is still called once. Internally, login() would store the credentials and authenticate the one channel it acquires. Then in send(), after acquiring a channel, you check a channel attribute (e.g. AUTH_KEY). If the channel isn't authenticated yet, send login with the stored credentials first, set the attribute, then send the actual command. This way, every channel gets authenticated transparently on first use, including channels created later by the pool.

mmodzelewski · 2026-02-24T19:35:07Z

foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java

        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {


A channel should also be released to the pool in case of exception

fix(java-sdk):Added Missing PoolMetrics and linting issues fix(java):fixed connection error fix(java):Fixed Linting Issues fix(java):fixed builder issues fix(java):fixed ci issues fix(java): fixed memory leaks fix(java):fixed memory leak at channelread0 fix(java):fixed broadcastasync memory leak feat(java): Added attribute based channel login fix(java): fixed error handling in inbound handler

ex172000 reviewed Jan 27, 2026

View reviewed changes

mmodzelewski linked an issue Jan 27, 2026 that may be closed by this pull request

[Java SDK] Implement Connection Pooling for Async Client #2232

Open

20 tasks

github-actions bot added the stale Inactive issue or pull request label Feb 4, 2026

hubcio changed the title ~~feat(java): Implemented Async Connection Pooling using FixedChannelPool (#2232)~~ feat(java): implement Async Connection Pooling using FixedChannelPool Feb 4, 2026

github-actions bot removed the stale Inactive issue or pull request label Feb 5, 2026

rythm-sachdeva force-pushed the feature/async-connection-pool branch 2 times, most recently from 2cdb131 to 50e8a36 Compare February 7, 2026 06:56

rythm-sachdeva force-pushed the feature/async-connection-pool branch from 50e8a36 to ca4583f Compare February 17, 2026 15:08

rythm-sachdeva force-pushed the feature/async-connection-pool branch 4 times, most recently from 4eaf354 to 181682b Compare February 22, 2026 01:40

mmodzelewski requested changes Feb 24, 2026

View reviewed changes

mmodzelewski reviewed Feb 24, 2026

View reviewed changes

rythm-sachdeva force-pushed the feature/async-connection-pool branch from 181682b to a31643a Compare February 28, 2026 13:48


		public Builder() {}

		public Builder setMaxConnections(int maxConnections) {

Conversation

rythm-sachdeva commented Jan 24, 2026

Description

Changes

Implementation Details

Testing

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

github-actions bot commented Feb 4, 2026

Uh oh!

rythm-sachdeva commented Feb 9, 2026

Uh oh!

mmodzelewski commented Feb 9, 2026

Uh oh!

slbotbm commented Feb 17, 2026

Uh oh!

mmodzelewski commented Feb 17, 2026

Uh oh!

codecov bot commented Feb 22, 2026

Codecov Report

Uh oh!

mmodzelewski left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants