큰 메시지가 무한 재시도로 쌓였다

ERROR Failed to publish product-changed event productId=P-8841902
org.apache.kafka.common.errors.RecordTooLargeException: The message is 4718392 bytes when serialized which is larger than 1048576, which is the value of the max.request.size configuration.
	at org.apache.kafka.clients.producer.KafkaProducer.ensureValidRecordSize(KafkaProducer.java:1163)
	at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:1024)

# producer.properties
max.request.size=10485760      # 1048576 → 10MB

# 브로커 message.max.bytes 는 기본값 1048588 그대로
# 토픽 max.message.bytes 도 미설정 → 1048588
# replica.fetch.max.bytes 도 기본값 1048576 그대로
# 컨슈머 max.partition.fetch.bytes 도 기본값 1048576 그대로

WARN [Producer clientId=catalog-publisher-2] Got error produce response in correlation id 1841992 on topic-partition product-changed-7, splitting and retrying (2147483647 attempts left). Error: MESSAGE_TOO_LARGE
WARN [Producer clientId=catalog-publisher-2] Got error produce response in correlation id 1841998 on topic-partition product-changed-7, splitting and retrying (2147483647 attempts left). Error: MESSAGE_TOO_LARGE
WARN [Producer clientId=catalog-publisher-2] Got error produce response in correlation id 1842004 on topic-partition product-changed-7, splitting and retrying (2147483647 attempts left). Error: MESSAGE_TOO_LARGE

org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for product-changed-7:120004 ms has passed since batch creation. The request has not been sent, or no server response has been received yet.

org.apache.kafka.common.errors.RecordTooLargeException: The request included a message larger than the max message size the server will accept.

ERROR [ReplicaManager broker=1] Error processing append operation on partition product-changed-7
org.apache.kafka.common.errors.RecordTooLargeException: The record batch size in the append to product-changed-7 is 4718392 bytes which exceeds the maximum configured value of 1048588.

① 프로듀서 클라이언트          max.request.size (1048576, 비압축 기준)
       │                       실패 → RecordTooLargeException (클라이언트에서 즉시)
       ↓
② 브로커 리더 append          message.max.bytes (1048588) 또는
       │                       토픽 max.message.bytes (오버라이드, 압축 후 기준)
       │                       실패 → MESSAGE_TOO_LARGE → Sender 가 배치 분할 재시도
       ↓
③ 팔로워 복제                 replica.fetch.max.bytes (1048576)
       │                       replica.fetch.response.max.bytes (10485760)
       │                       → 4.x 는 첫 배치를 반환해 진행을 보장
       ↓
④ 컨슈머 fetch                max.partition.fetch.bytes (1048576)
                               fetch.max.bytes (52428800)
                               → 4.x 는 첫 배치를 반환해 진행을 보장

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'
      # message.max.bytes 는 기본값 1048588 을 그대로 둡니다 (재현 포인트)

docker compose up -d
K=/opt/kafka/bin

docker exec -it broker $K/kafka-topics.sh --create --topic big-msg \
  --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092

# 3MiB 짜리 한 줄 파일을 만든다
docker exec broker sh -c 'head -c 3145728 /dev/zero | tr "\0" "x" > /tmp/big.txt'

# ── 관문 ① 프로듀서 max.request.size (기본 1048576)
docker exec -i broker sh -c "$K/kafka-console-producer.sh --topic big-msg \
  --bootstrap-server localhost:9092 < /tmp/big.txt"
#   → org.apache.kafka.common.errors.RecordTooLargeException:
#     The message is 3145729 bytes when serialized which is larger than 1048576,
#     which is the value of the max.request.size configuration.

# ── 관문 ① 를 열어 준다 → 관문 ② 브로커 message.max.bytes (기본 1048588) 에서 막힘
docker exec -i broker sh -c "$K/kafka-console-producer.sh --topic big-msg \
  --bootstrap-server localhost:9092 \
  --producer-property max.request.size=10485760 \
  --producer-property delivery.timeout.ms=15000 < /tmp/big.txt"
#   → Got error produce response ... Error: MESSAGE_TOO_LARGE
#     그리고 delivery.timeout.ms 초과 후:
#     org.apache.kafka.common.errors.RecordTooLargeException: The request included a message
#     larger than the max message size the server will accept.

# ── 관문 ② 를 토픽 레벨로 열어 준다
docker exec -it broker $K/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name big-msg \
  --alter --add-config max.message.bytes=10485760

docker exec -i broker sh -c "$K/kafka-console-producer.sh --topic big-msg \
  --bootstrap-server localhost:9092 \
  --producer-property max.request.size=10485760 < /tmp/big.txt"
#   → 성공합니다. 브로커 message.max.bytes 를 건드리지 않고
#     토픽 max.message.bytes 만 올려도 통과한다는 점을 확인하세요.

# ── 관문 ④ 컨슈머 — 기본값 그대로도 "읽힙니다" (4.x 동작 확인)
docker exec -it broker $K/kafka-console-consumer.sh --topic big-msg \
  --from-beginning --timeout-ms 20000 --bootstrap-server localhost:9092 \
  --property print.value=false --property print.offset=true
#   → max.partition.fetch.bytes(1048576) 보다 큰 레코드지만 정상적으로 반환됩니다.
#     "컨슈머가 멈춘다"는 오래된 설명이 현행 버전에서 틀렸음을 확인하는 단계입니다.

# 토픽을 다시 기본 한계로 되돌린다
docker exec -it broker $K/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name big-msg \
  --alter --delete-config max.message.bytes

# 작은 레코드 사이에 큰 레코드를 섞어 한 배치에 들어가게 만듭니다.
# linger.ms 를 크게 주면 배치가 뭉쳐져 분할 재시도가 잘 보입니다.
docker exec broker sh -c '{ \
  for i in $(seq 1 50); do echo "small-$i"; done; \
  head -c 2097152 /dev/zero | tr "\0" "y"; echo; \
  for i in $(seq 51 100); do echo "small-$i"; done; \
} > /tmp/mixed.txt'

docker exec -i broker sh -c "$K/kafka-console-producer.sh --topic big-msg \
  --bootstrap-server localhost:9092 \
  --producer-property max.request.size=10485760 \
  --producer-property linger.ms=1000 \
  --producer-property batch.size=5242880 \
  --producer-property delivery.timeout.ms=20000 < /tmp/mixed.txt" 2>&1 \
  | grep -c 'splitting and retrying'
#   → "splitting and retrying" 가 여러 번 나타납니다.
#     이것이 요청 폭증의 원인입니다.

# 1. 프로듀서를 롤백한다. max.request.size 를 원래대로 돌려 클라이언트에서
#    즉시 실패하게 만듭니다. 브로커까지 가서 폭증하는 것보다 낫습니다.
kubectl rollout undo deploy/catalog-publisher

# 2. 애플리케이션의 자체 재시도 큐를 비운다.
#    RecordTooLargeException 은 재시도 불가 예외입니다. 재시도하면 안 됩니다.

# 3. 큰 레코드를 만드는 기능을 끈다 (피처 플래그).
#    이것이 진짜 즉시 조치입니다.

# 4. 영향 범위를 좁힌다 — 같은 프로듀서 인스턴스를 쓰는 다른 토픽을 분리한다.
#    최소한 대형 페이로드 토픽 전용 프로듀서를 별도로 만듭니다.

// 변경 전 — 상세 HTML 을 인라인으로 담았습니다 (4.7MB)
ProductChanged before = ProductChanged.newBuilder()
    .setProductId(id)
    .setDescriptionHtml(hugeHtml)          // ← 이것이 문제
    .setOptionMatrix(hugeMatrix)
    .build();

// 변경 후 — 본문은 스토리지에 올리고 참조만 담습니다 (2KB)
String bodyKey = objectStore.put("product-body/" + id + "/" + revision, hugeHtml);
ProductChanged after = ProductChanged.newBuilder()
    .setProductId(id)
    .setRevision(revision)
    .setBodyRef(bodyKey)                   // ← 참조
    .setBodySha256(sha256(hugeHtml))       // ← 무결성 확인용
    .setBodySize(hugeHtml.length())
    .build();

producer.send(record, (metadata, e) -> {
    if (e == null) return;

    if (e instanceof RecordTooLargeException) {
        // 재시도 불가. 재시도 큐에 넣으면 무한히 쌓입니다.
        metrics.counter("publish.too_large").increment();
        dlq.send(record.key(), summarize(record));   // 본문 없이 요약만
        alerts.page("record too large: " + record.topic() + " key=" + record.key());
        return;
    }
    if (e instanceof RetriableException) {
        // delivery.timeout.ms 안에서 클라이언트가 이미 재시도했습니다.
        // 여기까지 왔다면 그 예산이 끝난 것입니다.
        outbox.save(record);
        return;
    }
    log.error("publish failed topic={} key={}", record.topic(), record.key(), e);
});

private static final int SOFT_LIMIT = 900 * 1024;   // 한계의 90% 를 경보선으로

byte[] payload = serializer.serialize(topic, event);
if (payload.length > SOFT_LIMIT) {
    metrics.gauge("publish.payload_bytes", payload.length);
    // 한계에 도달하기 전에 claim-check 경로로 전환합니다
    return publishWithClaimCheck(event);
}
producer.send(new ProducerRecord<>(topic, event.getKey(), payload), callback);