RF=3인데 브로커 1대 죽자 유실됐다

acks=all
enable.idempotence=true
max.in.flight.requests.per.connection=5
delivery.timeout.ms=120000
compression.type=lz4

$ kafka-configs.sh --bootstrap-server kafka-1:9092 \
    --entity-type topics --entity-name member-points --describe

Dynamic configs for topic member-points are:
  retention.ms=1209600000 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:retention.ms=1209600000}

# min.insync.replicas 가 없습니다 → 기본값 1 이 적용됩니다.

INFO [Partition member-points-3 broker=1] Shrinking ISR from 1,2 to 1. Leader: (highWatermark: 41288052, endOffset: 41288231). Out of sync replicas: (brokerId: 2, endOffset: 41287004, lastCaughtUpTimeMs: 1761789871442).

WARN  [Producer clientId=points-writer-1] Got error produce response with correlation id 88213 on topic-partition member-points-3, retrying (2147483646 attempts left). Error: NOT_ENOUGH_REPLICAS

# 재시도가 delivery.timeout.ms 를 넘기면 최종 실패로 콜백에 전달됩니다
org.apache.kafka.common.errors.NotEnoughReplicasException: The size of the current ISR : 1 is insufficient to satisfy the min.isr requirement of 2 for partition member-points-3, live replica(s) broker.id are : [1]

x-kafka-env: &kenv
  KAFKA_PROCESS_ROLES: 'broker,controller'
  KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
  KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093'
  KAFKA_LISTENERS: 'PLAINTEXT://:19092,CONTROLLER://:9093,PLAINTEXT_HOST://:9092'
  KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
  KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
  CLUSTER_ID: '4L6g3nShT-eMCtK--X86sw'
  KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
  KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
  KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
  KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
  KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'

services:
  kafka-1:
    image: apache/kafka:4.3.1
    hostname: kafka-1
    container_name: kafka-1
    ports: ['29092:9092']
    environment:
      <<: *kenv
      KAFKA_NODE_ID: 1
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-1:19092,PLAINTEXT_HOST://localhost:29092'
  kafka-2:
    image: apache/kafka:4.3.1
    hostname: kafka-2
    container_name: kafka-2
    ports: ['39092:9092']
    environment:
      <<: *kenv
      KAFKA_NODE_ID: 2
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-2:19092,PLAINTEXT_HOST://localhost:39092'
  kafka-3:
    image: apache/kafka:4.3.1
    hostname: kafka-3
    container_name: kafka-3
    ports: ['49092:9092']
    environment:
      <<: *kenv
      KAFKA_NODE_ID: 3
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-3:19092,PLAINTEXT_HOST://localhost:49092'

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

# 1. min ISR 을 다르게 준 토픽 두 개를 만든다
docker exec -it kafka-1 $K/kafka-topics.sh --create --topic minisr-1 \
  --partitions 1 --replication-factor 3 --bootstrap-server localhost:9092
docker exec -it kafka-1 $K/kafka-topics.sh --create --topic minisr-2 \
  --partitions 1 --replication-factor 3 --config min.insync.replicas=2 \
  --bootstrap-server localhost:9092

# 2. 브로커 두 대를 내려 ISR 을 리더 하나로 만든다
docker stop kafka-2 kafka-3
docker exec -it kafka-1 $K/kafka-topics.sh --describe --bootstrap-server localhost:9092 \
  --topic minisr-1 --topic minisr-2
#   → 둘 다 Isr: 1

# 3. acks=all 로 minisr-1 (min ISR=1) 에 쓴다 → 성공합니다. 이게 사고의 원인입니다.
echo "danger" | docker exec -i kafka-1 $K/kafka-console-producer.sh \
  --topic minisr-1 --bootstrap-server localhost:9092 \
  --producer-property acks=all
#   → 아무 오류 없이 끝납니다.

# 4. 같은 acks=all 로 minisr-2 (min ISR=2) 에 쓴다 → 거부됩니다.
echo "safe" | docker exec -i kafka-1 $K/kafka-console-producer.sh \
  --topic minisr-2 --bootstrap-server localhost:9092 \
  --producer-property acks=all --producer-property delivery.timeout.ms=10000
#   → NOT_ENOUGH_REPLICAS 로 재시도한 뒤 타임아웃으로 실패합니다:
#     org.apache.kafka.common.errors.NotEnoughReplicasException: The size of the current ISR : 1
#     is insufficient to satisfy the min.isr requirement of 2 for partition minisr-2-0, ...

# 5. acks=1 로 minisr-2 에 쓴다 → 성공합니다.
#    min.insync.replicas 는 쓰기 거부 측면에서 acks=all 과만 상호작용합니다.
echo "acks-1-bypass" | docker exec -i kafka-1 $K/kafka-console-producer.sh \
  --topic minisr-2 --bootstrap-server localhost:9092 \
  --producer-property acks=1
#   → 성공. 이것이 "min ISR 만 설정해도 안 되는" 이유입니다.

# 6. 브로커를 복구하고 각 토픽의 데이터를 확인한다
docker start kafka-2 kafka-3
sleep 20
docker exec -it kafka-1 $K/kafka-console-consumer.sh --topic minisr-2 \
  --from-beginning --timeout-ms 8000 --bootstrap-server localhost:9092
#   → "safe" 는 없고 "acks-1-bypass" 만 있습니다.

# 1. 개별 토픽에 적용
kafka-configs.sh --bootstrap-server kafka-1:9092 \
  --entity-type topics --entity-name member-points \
  --alter --add-config min.insync.replicas=2

# 2. RF 가 2 이하인 토픽에 2를 걸면 즉시 쓰기가 멈춥니다. 먼저 RF 를 확인하세요.
kafka-topics.sh --bootstrap-server kafka-1:9092 --describe \
  | awk '/^Topic:/ {print $1, $2, $6, $8}'

# 3. min ISR 이 없는 토픽 목록을 뽑는다 (점검용)
for t in $(kafka-topics.sh --bootstrap-server kafka-1:9092 --list | grep -v '^__'); do
  v=$(kafka-configs.sh --bootstrap-server kafka-1:9092 \
        --entity-type topics --entity-name "$t" --describe --all \
        | grep -o 'min.insync.replicas=[0-9]*' | head -1)
  echo "$t  $v"
done

# 4. 클러스터 기본값도 올린다 (신규 토픽에 적용)
kafka-configs.sh --bootstrap-server kafka-1:9092 \
  --entity-type brokers --entity-default \
  --alter --add-config min.insync.replicas=2

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        return;
    }
    if (exception instanceof RetriableException) {
        // delivery.timeout.ms 안에서는 클라이언트가 이미 재시도했습니다.
        // 여기까지 왔다면 그 시간이 끝난 것입니다.
        metrics.counter("kafka.produce.retriable_exhausted").increment();
    }
    // 유실을 만들지 않으려면 로컬에 보존하고 나중에 재전송해야 합니다.
    outbox.save(record);                 // 예: DB outbox 테이블
    log.error("produce failed, saved to outbox: topic={} key={}",
              record.topic(), record.key(), exception);
});