Fail permit-based operations with ConsumerShutdownException after consumer termination

[tiagomota]

Problem

Once the background consumer fiber has died (any exception thrown by poll), the permit-based operations subscribe, unsubscribe and assign block forever. withPermit offers a Request.WithPermit to the requests queue and then waits on a Deferred that nothing will ever complete, because the actor loop that services the queue is gone.

The request helper used by the commit path already handles this case by racing awaitTermination, so committing on a dead consumer fails fast with ConsumerShutdownException while unsubscribing on one hangs forever. That shutdown guarantee dates back to #66 ("All KafkaConsumer requests now check to ensure the consumer has not already shutdown") and was lost for the permit-based operations introduced in #906.

Why it matters

The idiomatic place to call unsubscribe is a Resource / bracket finalizer around the record stream. fs2 runs finalizers before surfacing the stream's error, and finalizers are uncancelable. So when the poll loop dies (for example an out-of-range fetch under AutoOffsetReset.None):

1. the record streams are correctly interrupted via awaitTermination,
2. the scope closes and runs the finalizer,
3. unsubscribe deadlocks,
4. the original error is never raised, and because the finalizer is uncancelable, no surrounding .timeout, fiber cancel or IO.race can fire either.

Fix

Mirror the request helper's race in withPermit:

requests.offer(
  Request.WithPermit(fa, deferred.complete(_: Either[Throwable, A]).void)
) >> F.race(awaitTermination.as(ConsumerShutdownException()), deferred.get.rethrow)

Behaviour matches the commit path: if the fiber died with an error, awaitTermination rethrows it; if the consumer terminated normally, the operation fails with ConsumerShutdownException. Live consumers are unaffected, the deferred wins the race.

[mergify]

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

• Queue this pull request