카프카4.0 부터는 eager rebalancing protocol이 삭제됩니다.

https://issues.apache.org/jira/browse/KAFKA-18839

[KAFKA-18839] Drop support for eager rebalancing in Streams - ASF JIRA

 

이 이슈(KAFKA-18839)는 Kafka Streams에서 기존에 사용되던 EAGER 재분배(eager rebalancing) 프로토콜에 대한 지원을 완전히 제거하는 작업입니다. Kafka Streams는 버전 3.1부터 EAGER 프로토콜을 사용 중단(deprecated)했고, 4.0 버전부터는 이 프로토콜을 아예 제거하기로 결정했습니다.

 

Why drop?

EAGER 프로토콜은 오래된 방식으로, 유지보수와 코드 복잡성을 증가시키는 요인이었습니다. 더 나은 성능과 안정성을 제공하는 COOPERATIVE 재분배 프로토콜로 전환하면서, EAGER 프로토콜은 불필요한 잔재물이 되었습니다. 단일 재분배 프로토콜에 집중하여 스트림즈 개선을 할 수 있게 됩니다.

 

이후 버전에서는?

 

• 이제부터 Kafka Streams는 오직 COOPERATIVE 재분배 프로토콜만 지원합니다.
• 기존에 EAGER 프로토콜을 사용하던 설정은 더 이상 유효하지 않으며, 시스템 구성 시 반드시 COOPERATIVE 프로토콜로 전환해야 합니다.

EAGER 리밸런싱 프로토콜이란?

컨슈머가 그룹에 합류/실패/종료 할 경우 그룹 코디네이터가 이를 감지하고 전체 파티션에 대해 Revoke(할당 취소 또는 연결 해제)를 수행. 이후, 그룹 코디네이터는 사전에 할당된 알고리즘(range, roundrobin 등)을 통해 재분배 수행합니다. 로직이 단순하지만 전체 파티션에 대해 재분배가 이루어져서 전체적으로 리밸런싱에 시간이 많이 소요되게 됩니다. 

 

반면, COOPERATIVE는 부분적으로 파티션을 재분배하기 때문에 점진적 리밸런싱을 통해 컨슈머들에게 끼치는 영향을 최소화 합니다.

 

코드로 설정하는 방법?

뭐 별도로 있는 것은 아니고 Cooperative가 붙어있는 Rebalancing 을 선택하면 됨. CooperativeStickyAssignor가 대표적입니다.

public class CooperativeStickyAssignorExample {
    public static void main(String[] args) {
    
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "cooperative-sticky-group");

        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // cooperative rebalancing을 사용하기 위한 assignor 지정
        props.put("partition.assignment.strategy", 
            "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");