[Bug Report]: Paused consumer stuck after rebalance

[massada]

Prerequisites

• I have searched issues to ensure it has not already been reported

Description

As per issue reported by @SonicGD in the retry extensions repo (see issue 151), if a rebalance is issued when a message is in the retry loop, it will not be retried after partitions reassignment.

I've pin-pointed the issue to the revoke handler in the consumer and will open a PR soon.

Retry loop flow on error:

1. Call next middleware
2. Pause the consumer if needed
3. Goto 1 if an handled exception is catched, or
4. if an unhandled exception is catched or if the worker is stopped, resume the consumer if needed

Pause flow:

• ConsumerFlowManager saves the topic+partition internally, pauses the consumer and starts a heartbeat task to keep the consumer alive

Resume flow:

• ConsumerFlowManager resumes the consumer if the topic+partition is known and stops the heartbeat task

Revoke handling flow:

• Kafka driver calls the Consumer revoke handler
• Consumer revoke handler cleans up and stops the ConsumerFlowManager
  • ConsumerFlowManager clears the topic+partition list and stops the heartbeat
• Consumer calls any registered revoke handlers, one of which is in ConsumerManager, which stops the worker pool

Issue lies in stopping the ConsumerFlowManager before the worker pool, since stopping the latter triggers the retry loop cancellation, which tries to resume the consumer, but by that time the ConsumerFlowManager's topic+partition list was already cleared and so the resume does nothing.

Steps to reproduce

See the issue in the retry extensions repo for code reproducing the problem.

Expected behavior

A new message should be consumed after partitions reassignment.

Actual behavior

The consumer stays paused and no message is consumed after partitions reassignment.

KafkaFlow version

v3.0.9