Repository navigation
Redis Streams Imperative creating massive connections that lead to connection problems #3288
Description
When creating a StreamMessageListenerContainer with Spring boot 4.0.1 written in Kotlin the schedulers seem to open massive connections until the connections stack is full and results in the following error:
java.net.BindException: Can't assign requested address
val create = StreamMessageListenerContainer.create(
connectionFactory,
StreamMessageListenerContainer.StreamMessageListenerContainerOptions
.builder()
.pollTimeout(Duration.ofMillis(blockMillis))
.targetType(JobRunInfo::class.java)
.errorHandler { e ->
log.error("Redis Stream error", e) }
.build()
)
create.receive(
Consumer.from(groupName, consumerName),
StreamOffset.fromStart(streamKey), {
handler.invoke(it)
})
create.start()Full stacktrace:
org.springframework.data.redis.RedisConnectionFailureException: Unable to connect to Redis
at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory$ExceptionTranslatingConnectionProvider.translateException(LettuceConnectionFactory.java:1868) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory$ExceptionTranslatingConnectionProvider.getConnection(LettuceConnectionFactory.java:1799) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.connection.lettuce.LettuceConnection.doGetAsyncDedicatedConnection(LettuceConnection.java:968) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.connection.lettuce.LettuceConnection.getOrCreateDedicatedConnection(LettuceConnection.java:1022) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.connection.lettuce.LettuceConnection.getAsyncDedicatedConnection(LettuceConnection.java:953) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.connection.lettuce.LettuceStreamCommands.getAsyncDedicatedConnection(LettuceStreamCommands.java:328) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.connection.lettuce.LettuceStreamCommands.xReadGroup(LettuceStreamCommands.java:292) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.stream.DefaultStreamMessageListenerContainer.lambda$getReadFunction$1(DefaultStreamMessageListenerContainer.java:245) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:413) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:380) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:368) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.stream.DefaultStreamMessageListenerContainer.lambda$getReadFunction$0(DefaultStreamMessageListenerContainer.java:244) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.stream.StreamPollTask.readRecords(StreamPollTask.java:147) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.stream.StreamPollTask.doLoop(StreamPollTask.java:128) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.stream.StreamPollTask.run(StreamPollTask.java:113) ~[spring-data-redis-4.0.1.jar:4.0.1]
at java.base/java.lang.Thread.run(Thread.java:1583) ~[na:na]
Caused by: io.lettuce.core.RedisConnectionException: Unable to connect to localhost/<unresolved>:49952
at io.lettuce.core.RedisConnectionException.create(RedisConnectionException.java:63) ~[lettuce-core-6.8.1.RELEASE.jar:6.8.1.RELEASE/ec0535e]
at io.lettuce.core.RedisConnectionException.create(RedisConnectionException.java:41) ~[lettuce-core-6.8.1.RELEASE.jar:6.8.1.RELEASE/ec0535e]
at io.lettuce.core.AbstractRedisClient.getConnection(AbstractRedisClient.java:352) ~[lettuce-core-6.8.1.RELEASE.jar:6.8.1.RELEASE/ec0535e]
at io.lettuce.core.RedisClient.connect(RedisClient.java:220) ~[lettuce-core-6.8.1.RELEASE.jar:6.8.1.RELEASE/ec0535e]
at org.springframework.data.redis.connection.lettuce.StandaloneConnectionProvider.lambda$getConnection$1(StandaloneConnectionProvider.java:113) ~[spring-data-redis-4.0.1.jar:4.0.1]
at java.base/java.util.Optional.orElseGet(Optional.java:364) ~[na:na]
at org.springframework.data.redis.connection.lettuce.StandaloneConnectionProvider.getConnection(StandaloneConnectionProvider.java:113) ~[spring-data-redis-4.0.1.jar:4.0.1]
at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory$ExceptionTranslatingConnectionProvider.getConnection(LettuceConnectionFactory.java:1797) ~[spring-data-redis-4.0.1.jar:4.0.1]
... 14 common frames omitted
Caused by: io.netty.channel.AbstractChannel$AnnotatedSocketException: Can't assign requested address: localhost/127.0.0.1:49952
Caused by: java.net.BindException: Can't assign requested address
at java.base/sun.nio.ch.Net.connect0(Native Method) ~[na:na]
at java.base/sun.nio.ch.Net.connect(Net.java:589) ~[na:na]
at java.base/sun.nio.ch.Net.connect(Net.java:596) ~[na:na]
at java.base/sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:880) ~[na:na]
at io.netty.util.internal.SocketUtils$3.run(SocketUtils.java:91) ~[netty-common-4.2.9.Final.jar:4.2.9.Final]
at io.netty.util.internal.SocketUtils$3.run(SocketUtils.java:88) ~[netty-common-4.2.9.Final.jar:4.2.9.Final]
at java.base/java.security.AccessController.doPrivileged(AccessController.java:571) ~[na:na]
at io.netty.util.internal.SocketUtils.connect(SocketUtils.java:88) ~[netty-common-4.2.9.Final.jar:4.2.9.Final]
at io.netty.channel.socket.nio.NioSocketChannel.doConnect(NioSocketChannel.java:315) ~[netty-transport-4.2.9.Final.jar:4.2.9.Final]
at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.connect(AbstractNioChannel.java:295) ~[netty-transport-4.2.9.Final.jar:4.2.9.Final]
at io.netty.channel.DefaultChannelPipeline$HeadContext.connect(DefaultChannelPipeline.java:1361) ~[netty-transport-4.2.9.Final.jar:4.2.9.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeConnect(AbstractChannelHandlerContext.java:547) ~[netty-transport-4.2.9.Final.jar:4.2.9.Final]
at io.netty.channel.AbstractChannelHandlerContext.connect(AbstractChannelHandlerContext.java:526) ~[netty-transport-4.2.9.Final.jar:4.2.9.Final]
at io.netty.channel.AbstractChannelHandlerContext.connect(AbstractChannelHandlerContext.java:510) ~[netty-transport-4.2.9.Final.jar:4.2.9.Final]
at io.netty.channel.DefaultChannelPipeline.connect(DefaultChannelPipeline.java:999) ~[netty-transport-4.2.9.Final.jar:4.2.9.Final]
at io.netty.channel.Channel.connect(Channel.java:297) ~[netty-transport-4.2.9.Final.jar:4.2.9.Final]
at io.netty.bootstrap.Bootstrap$3.run(Bootstrap.java:264) ~[netty-transport-4.2.9.Final.jar:4.2.9.Final]
at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:148) ~[netty-common-4.2.9.Final.jar:4.2.9.Final]
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:141) ~[netty-common-4.2.9.Final.jar:4.2.9.Final]
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java) ~[netty-common-4.2.9.Final.jar:4.2.9.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:535) ~[netty-common-4.2.9.Final.jar:4.2.9.Final]
at io.netty.channel.SingleThreadIoEventLoop.run(SingleThreadIoEventLoop.java:201) ~[netty-transport-4.2.9.Final.jar:4.2.9.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:1195) ~[netty-common-4.2.9.Final.jar:4.2.9.Final]
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.2.9.Final.jar:4.2.9.Final]
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.2.9.Final.jar:4.2.9.Final]
at java.base/java.lang.Thread.run(Thread.java:1583) ~[na:na]
Tested on Mac OS.
I created a repro project: https://github.com/arne-kroeger/demo-redis-repro (here the setup is done via TestController SmartLifecycle start.
When I remove the group configuration Consumer.from(groupName, consumerName) everything works perfect.
Activity
mp911de commentedon Jan 6, 2026
Seems that the block timeout is so small that read requests are issued in rapid succession leading to a state where available ports are blocked by TCP connections in
FIN_WAITstate.Please either increase the block timeout to avoid exhausting the connection address space or use connection pooling.
We have seen reports however in which
XREADGROUPreturns immediately (without considering theBLOCKtimeout). We weren't however able to get hold of a reproducer that is able to demonstrate the behavior.arne-kroeger commentedon Jan 7, 2026
@mp911de - I have also tried it with 20000 ms and this resulted in the same result.
mp911de commentedon Jan 7, 2026
What about pooling?
arne-kroeger commentedon Jan 7, 2026
@mp911de - that fixed that problem in the first place regarding this exception but still no data was processed.
The problem was that I specified for the Consumer Group a stream offset of
StreamOffset.fromStart(streamKey). Which does not make any sense. By changing it toStreamOffset.create(streamKey, ReadOffset.lastConsumed())it fixed the connection problem without specifying a connection pool.Seems to be that by specifying the start index it lead to these problems. Maybe it is worth catching this as an assertion case?
mp911de commentedon Jan 7, 2026
Thanks for additional details. It likely is a combination that we haven't tested and considered yet and its worth revisiting the arrangement to see whether we can improve here.
arne-kroeger commentedon Jan 7, 2026
Thanks for your support. Shall I close this ticket for now or do you want to use it as a reference?
mp911de commentedon Jan 7, 2026
Please keep it open, we want to explore how we can do better.
jewoodev commentedon Jun 2, 2026
Hi @arne-kroeger, I looked into this and I think the diagnosis you arrived at matches the lower-level behavior I found — the combination is being silently lost on the Redis side, then amplified by the polling/connection model on the Spring side.
Per the Redis docs, when
XREADGROUPis called with a history/concrete ID such as0-0instead of>(which is whatStreamOffset.fromStart(...)resolves to, viaReadOffset.from("0-0")), the command enters the PEL/history mode andBLOCK,NOACK, andCLAIMare ignored; it returns synchronously even with an empty result. This is documented at https://redis.io/docs/latest/commands/xreadgroup/ and visible int_stream.c'sxreadCommand(serve_history=1→serve_synchronously=1branch).From the current code, Spring Data Redis treats this case as if the call were really blocking:
LettuceStreamCommands#xReadGroupacquires a dedicated connection wheneverreadOptions.isBlocking()is true, regardless of the offset value.StreamPollTask#doLooppolls back-to-back withThread.sleep(0)between iterations.LettuceConnectionFactory,RedisTemplate.execute's per-call acquire/release means every poll opens and closes a fresh native connection, which is what exhausts the ephemeral port range and surfaces asBindException.So
Can't assign requested addressis the secondary effect of the silently-ignoredBLOCK, not the primary fault.I'd be happy to work on a focused fix. My plan would be:
readOptions.isBlocking()is true, aConsumeris set, and the offset is a history/concrete stream ID such as0-0. The natural place is inDefaultStreamMessageListenerContainer—ConsumerStreamReadRequestBuilderitself doesn't see the container'spollTimeout, but the container does.fromStart(...)or other explicit stream IDs with aConsumer, I'd suggest a deprecation-first step: log a clear WARN that explains the Redis-side ignoredBLOCKand points atReadOffset.lastConsumed(), and only escalate toIllegalArgumentExceptionin 5.0 (or behind an opt-in flag if you prefer it sooner).ReadOffset.from(String),StreamReadOptions#block, andStreamMessageListenerContainerOptions#pollTimeoutto note that history/concrete offsets putXREADGROUPinto PEL mode where Redis ignoresBLOCK.I also considered making the polling task reuse a single dedicated connection (which would reduce the connection-churn symptom for this immediate-return polling path), but that changes the
LettuceConnection/RedisTemplatecallback lifecycle more broadly, so I'd rather start with validation + docs and leave that as a separate discussion if you'd like to pursue it.I'd add focused unit coverage for the new WARN/exception; the existing
*StreamMessageListenerContainerIntegrationTestspaths uselastConsumed()and should stay untouched.Please let me know whether you prefer a WARN-first path, an immediate hard fail, or another direction.