Support cooperative rebalancing protocol

[ocadaruma]

• Kafka 2.4.0 introduces cooperative rebalancing protocol, which mitigates the stop-the-world effect upon rebalancing
  • https://cwiki.apache.org/confluence/display/KAFKA/KIP-429%3A+Kafka+Consumer+Incremental+Rebalance+Protocol
• Users can opt-in cooperative rebalancing by specifying CooperativeStickyAssignor in partition.assignment.strategy config or implementing ConsumerPartitionAssignor which returns RebalanceProtocol.COOPERATIVE from supportedProtocols
• If cooperative rebalancing is enabled, Kafka consumer will invoke ConsumerRebalanceListener as like below:
  • onPartitionsRevoked : called only when there's at least one partition that revoked from current assignment
  • onPartitionsAssigned: called with newly assigned partitions
  • refs: https://cwiki.apache.org/confluence/display/KAFKA/KIP-429%3A+Kafka+Consumer+Incremental+Rebalance+Protocol#KIP429:KafkaConsumerIncrementalRebalanceProtocol-ConsumerRebalanceListenerandConsumerPartitionAssignorSemantics
• onPartitionsAssigned's behavior change makes Decaton to not working, because AssignmentManager assumes partitions which previously assigned and kept assigned after rebalance as revoked unintentionally.
  • https://github.com/line/decaton/blob/master/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/AssignmentManager.java#L109

[ocadaruma]

Cooperative assignor will be used by default from Kafka 3.3.0 https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+3.3.0