Support cooperative rebalancing fixes #98,#118

[ocadaruma]

Summary

• Kafka 2.4 added cooperative rebalancing, which aims to minimize well-known "stop-the-world" effect upon consumer rebalance
  • Essentially, what cooperative rebalancing achieves is to revoke/assign only necessary partitions on rebalance and allows to fetch on still-owned partitions even during rebalance.
  • We can opt-in cooperative rebalancing protocol by specifying ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG property
• However, currently, Decaton doesn't work with cooperative rebalancing for some reasons. This PR gonna fix them

Problems of current Decaton implementation to support cooperative rebalancing

• Deprecated Consumer#poll(long) is used
  • It's already deprecated and should be replaced with Consumer#poll(Duration)
  • There are several behavior changes in Consumer#poll(Duration) though, ~from cooperative rebalancing-support perspective~, the most notable one is poll(Duration) doesn't block for entire rebalancing procedure unlike poll(long)
    • i.e. When rebalance is initiated, after onPartitionRevoked is called, poll() may return without waiting onPartitionAssigned invocation in same poll. (It will be eventually invoked in later poll() though)
  • Currently, Decaton is relying on the behavior that onPartitionRevoked and onPartitionAssigned are executed in same poll()
    • Once onPartitionRevoked returns, revoked partitions are no longer owned by KafkaConsumer so any pause/resume/offset commit may cause RuntimeException and kills subscription thread
    • However, since Decaton postpones destroying PartitionContexts to at onPartitionAssigned (this was to minimize context-recreation as much as possible. In short, Decaton itself did similar thing to cooperative-rebalancing from eager-rebalancing era)
    • Hence, we need some mechanism to remember which partitions are already revoked, until we actually destroy contexts in subsequent onPartitionAssigned
• Decaton relies on the eager-rebalancing's behavior that "all assigned partitions are revoked in onPartitionsRevoked => new assignment is added in onPartitionsAssigned".
  • With cooperative-rebalancing, onPartitionsAssigned will be called only on newly-added partitions

Summary of changes

• Store revoking status in PartitionContext, which is used to prevent calling pause/resume/offset commit for revoked partitions during rebalance
• Fix ConsumeManager to work even if subset of partitions (i.e. not full assignment) are passed in ConsumerRebalanceListener's callbacks
• SubscriptionStateListener state transition may change
  • Because poll() may return after onPartitionsRevoked before onPartitionsAssigned, the transition REBALANCING -> SHUTTING_DOWN is now possible.
  • This should be considered as breaking change

Appendix

• Since the behavior change that "onPartitionsRevoked and onPartitionsAssigned may be called in different poll() introduces certain complexity, I wrote small TLA+ spec to get a confidence about the design
  • https://gist.github.com/ocadaruma/226c2f4e65cb97aba3610341903955b0

[ocadaruma]

Minimum supported version of Kafka-client should maybe be documented?

Sorry, overlooked.

After this PR gets merged, Decaton will require kafka-clients at least 2.0.0 because poll(Duration) is used but I think it's quite long ago so not necessary to be explicitly documented?

[mauhiz]

Thanks, I'm good. Thanks for the nice feature!

Regarding the minimum version of Kafka client, I think it's better documented than not!

[ocadaruma]

Okay then, I'll followup in separate PR about minimum kafka-clients ver.