컨슈머가 무한 리밸런스 루프에 빠졌다

while (running) {
    ConsumerRecords<String, ImageEvent> records = consumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, ImageEvent> record : records) {
        // 외부 리사이즈 서비스 호출 — 평소 400ms, 부하 시 2.5s
        byte[] thumb = resizeClient.resize(record.value().getSourceUrl());
        objectStore.put(thumbKey(record), thumb);
    }
    consumer.commitSync();   // 배치 전체를 처리한 뒤 커밋
}

WARN  [Consumer clientId=thumbnailer-2, groupId=image-thumbnailer] consumer poll timeout has expired. This means the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
INFO  [Consumer clientId=thumbnailer-2, groupId=image-thumbnailer] Member thumbnailer-2-4b7d1c9a-5f31-4e02-9a1d-0c8f7e2b6a44 sending LeaveGroup request to coordinator kafka-2:9092 (id: 2 rack: null) due to consumer poll timeout has expired.
INFO  [Consumer clientId=thumbnailer-2, groupId=image-thumbnailer] Resetting generation and member id due to: consumer pro-actively leaving the group
INFO  [Consumer clientId=thumbnailer-2, groupId=image-thumbnailer] Request joining group due to: consumer pro-actively leaving the group
INFO  [Consumer clientId=thumbnailer-2, groupId=image-thumbnailer] (Re-)joining group

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:...)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:...)

WARN  [Consumer clientId=thumbnailer-2, groupId=image-thumbnailer] Asynchronous auto-commit of offsets {image-uploaded-4=OffsetAndMetadata{offset=1841020, ...}} failed: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.

INFO [GroupCoordinator id=2] [Group image-thumbnailer] Member thumbnailer-2-4b7d1c9a-... has left group through explicit `LeaveGroup` request; client reason: consumer poll timeout has expired.
INFO [GroupCoordinator id=2] Preparing to rebalance group image-thumbnailer in state PreparingRebalance with old generation 214 (reason: removing member thumbnailer-2-4b7d1c9a-... on LeaveGroup).
INFO [GroupCoordinator id=2] Stabilized group image-thumbnailer generation 215 with 3 members.
INFO [GroupCoordinator id=2] [Group image-thumbnailer] Member thumbnailer-0-1a09f4de-... has left group through explicit `LeaveGroup` request; client reason: consumer poll timeout has expired.
INFO [GroupCoordinator id=2] Preparing to rebalance group image-thumbnailer in state PreparingRebalance with old generation 215 (reason: removing member thumbnailer-0-1a09f4de-... on LeaveGroup).

$ kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
    --describe --group image-thumbnailer --state

GROUP              COORDINATOR (ID)  ASSIGNMENT-STRATEGY  STATE                #MEMBERS
image-thumbnailer  kafka-2:9092 (2)  range                PreparingRebalance   3

$ # 4초 뒤 다시
GROUP              COORDINATOR (ID)  ASSIGNMENT-STRATEGY  STATE                #MEMBERS
image-thumbnailer  kafka-2:9092 (2)  range                CompletingRebalance  4

$ kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
    --describe --group image-thumbnailer

Warning: Consumer group 'image-thumbnailer' is rebalancing.

GROUP              TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG      CONSUMER-ID  HOST  CLIENT-ID
image-thumbnailer  image-uploaded  0          1841020         2094882         253862   -            -     -
image-thumbnailer  image-uploaded  1          1839774         2093010         253236   -            -     -
image-thumbnailer  image-uploaded  2          1840119         2093551         253432   -            -     -
...

$ # 30초 뒤 — CURRENT-OFFSET 이 동일합니다. LAG 만 늘어납니다.

services:
  broker:
    image: apache/kafka:4.3.1
    hostname: broker
    container_name: broker
    ports:
      - '9092:9092'
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
      KAFKA_LISTENERS: 'CONTROLLER://:29093,PLAINTEXT://:19092,PLAINTEXT_HOST://:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker:19092,PLAINTEXT_HOST://localhost:9092'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      CLUSTER_ID: '4L6g3nShT-eMCtK--X86sw'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class SlowConsumer {
    public static void main(String[] args) {
        Properties p = new Properties();
        p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        p.put(ConsumerConfig.GROUP_ID_CONFIG, "slow-group");
        p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        // 재현 포인트 1 — 폴 간격 한계를 10초로 줄인다 (기본 300000)
        p.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 10_000);
        // 재현 포인트 2 — 한 배치에 20건. 20 x 1s = 20s > 10s 이므로 반드시 초과한다
        p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20);

        try (KafkaConsumer<String, String> c = new KafkaConsumer<>(p)) {
            c.subscribe(List.of("slow-topic"));
            while (true) {
                ConsumerRecords<String, String> records = c.poll(Duration.ofSeconds(1));
                System.out.println("polled " + records.count());
                for (ConsumerRecord<String, String> r : records) {
                    Thread.sleep(1_000);           // 건당 1초 처리
                    System.out.println("  done " + r.offset());
                }
                c.commitSync();                    // 여기서 CommitFailedException 이 터집니다
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

docker compose up -d

# 1. 파티션 2개 토픽
docker exec -it broker /opt/kafka/bin/kafka-topics.sh --create \
  --topic slow-topic --partitions 2 --replication-factor 1 \
  --bootstrap-server localhost:9092

# 2. 200건 적재
docker exec -i broker /opt/kafka/bin/kafka-console-producer.sh \
  --topic slow-topic --bootstrap-server localhost:9092 \
  < <(seq 1 200)

# 3. SlowConsumer 를 두 개 띄운다 (같은 group.id)
#    → 10초마다 WARN "consumer poll timeout has expired" 가 반복되고
#      commitSync() 에서 CommitFailedException 이 발생합니다.
java -cp kafka-clients-4.3.1.jar:. SlowConsumer &
java -cp kafka-clients-4.3.1.jar:. SlowConsumer &

# 4. 그룹 상태가 Stable 에 머물지 못하는 것을 확인
watch -n 2 'docker exec -it broker /opt/kafka/bin/kafka-consumer-groups.sh \
  --describe --group slow-group --state --bootstrap-server localhost:9092'

# 5. CURRENT-OFFSET 이 전진하지 않는 것을 확인
docker exec -it broker /opt/kafka/bin/kafka-consumer-groups.sh \
  --describe --group slow-group --bootstrap-server localhost:9092

# 20 x 2.5s = 50s. max.poll.interval.ms 기본 300s 안에 충분히 들어옵니다.
max.poll.records=20

while (running) {
    ConsumerRecords<String, ImageEvent> records = consumer.poll(Duration.ofMillis(500));

    for (TopicPartition tp : records.partitions()) {
        for (ConsumerRecord<String, ImageEvent> r : records.records(tp)) {
            workerPool.submit(() -> process(r));   // 느린 작업은 여기서
        }
        // 워커 큐가 깊어지면 그 파티션의 fetch 를 잠시 멈춘다.
        // poll() 자체는 계속 호출되므로 max.poll.interval.ms 를 넘기지 않는다.
        if (queueDepth(tp) > HIGH_WATERMARK) consumer.pause(List.of(tp));
    }

    // 워커가 완료한 지점까지만 커밋한다 (완료 오프셋 추적기 필요)
    Map<TopicPartition, OffsetAndMetadata> done = completedOffsets();
    if (!done.isEmpty()) consumer.commitSync(done);

    // 큐가 비면 다시 재개
    for (TopicPartition tp : consumer.paused()) {
        if (queueDepth(tp) < LOW_WATERMARK) consumer.resume(List.of(tp));
    }
}