Rebalance coordinator should not wait on a failed stream

[Awethon]

When a consumer receives an unhandled exception during processing, it keeps working until timeout:

timestamp=2025-09-10T19:26:22.789653Z level=ERROR thread=#zio-fiber-1028656011 message="Error in withStream fiber in runWithGracefulShutdown" cause="java.lang.RuntimeException: Simulated failure

timestamp=2025-09-10T19:13:18.599604Z level=WARN thread=#zio-fiber-1784753688 message="Exceeded deadline waiting for streams (of revoked partitions) to commit the offsets of the records they consumed; the rebalance will continue. This might cause another consumer to process some records again. Revoked partitions: kafka-support-test-topic-0: stream ended, last pulled offset=1, last committed offset=none, not committed" location=zio.kafka.consumer.internal.RebalanceCoordinator.doAwaitStreamCommits.logFinalStreamCompletionStatuses file=RebalanceCoordinator.scala line=125

But since the exception was already returned for a record, there's no option to commit it, and no other records are provided to the consumer.
This prevents a quick recovery (redeployment or restart of a consumer) and also tests the fail-fast behavior of a consumer idle for a long time.

[erikvanoosten]

Hi @Awethon thanks for your report.

What is this 'unhandled exception' in the consumer?

Related: please see ConsumerSettings.withMaxRebalanceDuration where you see that the default timeout is very conservative. Most of the time you can set this much lower because you know how fast records can be processed.

[Awethon]

Hey @erikvanoosten, I managed to reproduce it in zio-kafka. The initial problem was encountered in our wrapper for zio-kafka, so I didn't have any code to share.

        val kvs = (1 to 5).toList.map(i => (s"key$i", s"msg$i"))
        for {
          topic  <- randomTopic
          client <- randomClient
          group  <- randomGroup

          producer <- KafkaTestUtils.makeProducer
          _        <- KafkaTestUtils.produceMany(producer, topic, kvs)

          consumer <- KafkaTestUtils.makeConsumer(
                        client,
                        Some(group),
                        rebalanceSafeCommits = true,
                        maxRebalanceDuration = 40.seconds
                      )
          _ <- consumer
                 .plainStream(Subscription.Topics(Set(topic)), Serde.string, Serde.string)
                 .mapZIOChunked { x =>
                   Console.printLine(s"Starting processing ${x.key}") *>
                     ZIO.die(new RuntimeException(s"Failed consuming message ${x.key}")).as(x.offset)
                 }
                 .aggregateAsync(Consumer.offsetBatches)
                 .mapZIOChunked(_.commit)
                 .take(5)
                 .runDrain
        } yield assertTrue(true)

On receiving an exception consumer hangs until it hits rebalancing deadline (which is 25 seconds if we run it in ConsumerSpec):

22:11:32.134 [zio-kafka-runloop-thread-0] [] INFO z.k.c.internal.RebalanceCoordinator - 1 partitions are assigned
Starting processing key1
22:11:32.269 [zio-kafka-runloop-thread-0] [] INFO z.k.c.internal.RebalanceCoordinator - 1 partitions are revoked
Rebalancing deadline is PT25S
22:11:57.281 [zio-kafka-runloop-thread-0] [] WARN z.k.c.i.R.doAwaitStreamCommits - Exceeded deadline waiting for streams (of revoked partitions) to commit the offsets of the records they consumed; the rebalance will continue. This might cause another consumer to process some records again. Revoked partitions: consumespec-topic-c8dd119b-9b28-4e2f-8c48-fe7a6d3ae0ca-0: stream ended, last pulled offset=4, last committed offset=none, not committed
- [test name]
java.lang.RuntimeException: Failed consuming message key1

I'd expect consumer to be terminated immediately in this situation.

I have also increased these timeouts because docker kafka doesn't allow low session timeouts:

          ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG       -> "50000",
          ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG    -> "10000",

Also, do you by any chance know the motivation to calculate it as a diff between maxRebalanceDuration and commitTimeout val deadline = java.lang.System.nanoTime() + maxRebalanceDuration.toNanos - commitTimeoutNanos ?

[Awethon]

@erikvanoosten please reopen the issue. It wasn't fixed by #1576.

[erikvanoosten]

@erikvanoosten please reopen the issue. It wasn't fixed by #1576.

Of course. You could have done it as well I think.

[Awethon]

No, it says I have no permissions to do so :(

[erikvanoosten]

Thanks for re-opening. I only read this issue properly for the first time now (#1573 was filling all my free time).

I have also increased these timeouts because docker kafka doesn't allow low session timeouts:

Do you have some more information about that? We're adding settings' validation in #1580 and that would fit right in.

Also, do you by any chance know the motivation to calculate it as a diff between maxRebalanceDuration and commitTimeout. val deadline = java.lang.System.nanoTime() + maxRebalanceDuration.toNanos - commitTimeoutNanos ?

I do (I wrote it). Initially the commitTimeout didn't play a role in this code. After sending a commit to the broker, you have to wait for it to complete (that is the whole point of rebalance-safe-commits). Unfortunately, you can't just assume it completes in zero time. It is necessary to reserve some time in order to guarantee staying below maxRebalanceDuration. How much time is needed? We could use average commit time or something like that, but those are not guarantees. The only number we have is the commitTimeout and since a commit is declared a failure when it takes longer, commitTimeout is equal to the worse case commit time.

I'd expect consumer to be terminated immediately in this situation.

When rebalance-safe-commits is enabled, zio-kafka tries it best to let every partition stream complete its processing. If one partition stream throws an exception, it is safer (to prevent duplicates as much as possible) to let the other partition streams continue.

In the example code however, there is only 1 partition (it is implicitly created), so in this case the rebalance should not be held up. So indeed, this is valid bug report. We can probably track the exit status for each partition and use that as an additional condition in the rebalance coordinator.

[Awethon]

Do you have some more information about that?

When I keep default timeouts of zio-kafka tests but switch execution from embedded kafka to local kafka, when I try to initialize tests I see in docker logs that it fails saying session timeout is too low, but I don't have more info on that.

I didn't think of multiple partitions, makes sense now. How I imagine it should work is that failed partition immediately initiates stream shutdown and if other partitions if they exist and don't have any processing to do, they will also shutdown immediately.

[erikvanoosten]

I didn't think of multiple partitions, makes sense now. How I imagine it should work is that failed partition immediately initiates stream shutdown and if other partitions if they exist don't have any processing to do, they will also shutdown immediately.

if they exist don't have any processing to do

Right. Or, if they all failed, that should also lead to immediate shutdown.

[erikvanoosten]

@Awethon I am going to ignore this issue for a while. Now that I am looking at the rebalance coordinator code again I remember that this not something we can easily fix in zio-kafka (otherwise, we would have). The problem is that as soon as we hand out a partition stream to the user application, we no longer know what it is doing. The only thing we know is whether elements are being pulled from the stream, or whether we reached the end of that stream. From zio-kafka we can't see the difference between a program that is taking its time to process the elements, or one that crashed and burned. The only solution is to run the partition stream in the context of a zio-kafka call. This is what the new and experimental Consumer.*WithControl methods make possible. But even then, I am not sure if we can see crashes at partition stream granularity. Perhaps that @svroonland can shine a light on this when he is available again.