재처리했더니 결제가 두 번 됐다

application.id=payment-settlement
processing.guarantee=exactly_once_v2
# Streams 가 내부적으로 트랜잭션을 쓰고, 컨슈머는 read_committed 로 동작합니다

public void process(Record<String, PaymentApproved> record) {
    PaymentApproved approved = record.value();

    // ★ 외부 호출. Kafka 트랜잭션 바깥입니다.
    PgResult result = pgClient.confirmPayment(
        approved.getOrderId(),
        approved.getAmount());          // 멱등 키 없음

    context().forward(record.withValue(toSettled(result)));
}

$ kafka-console-consumer.sh --bootstrap-server kafka-1:9092 \
    --topic payment-settled --from-beginning --timeout-ms 60000 \
    --isolation-level read_committed \
    --property print.partition=true --property print.offset=true \
    --property print.key=true | grep ORD-77412

Partition:5	Offset:1180224	ORD-77412	{"orderId":"ORD-77412","pgApprovalNo":"A-8812043","amount":34900}
Partition:5	Offset:1204881	ORD-77412	{"orderId":"ORD-77412","pgApprovalNo":"A-9014772","amount":34900}

producer.initTransactions();          // transactional.id 등록, 이전 인스턴스의 트랜잭션 abort

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

    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, String> r : records) {
            // ① 출력 토픽 쓰기 — 트랜잭션 안
            producer.send(new ProducerRecord<>("output", r.key(), transform(r.value())));
        }
        // ② 입력 오프셋 커밋 — 트랜잭션 안
        producer.sendOffsetsToTransaction(offsetsOf(records), consumer.groupMetadata());
        producer.commitTransaction();     // ①과 ②가 함께 성공하거나 함께 실패
    } catch (AbortableException e) {
        producer.abortTransaction();
        // ★ 중요: abort 후에는 컨슈머 위치를 명시적으로 되돌려야 합니다.
        //   컨슈머는 자동으로 rewind 하지 않습니다.
        resetConsumerPosition(consumer, records);
    }
}

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.nio.file.*;
import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class SideEffectConsumer {
    public static void main(String[] args) throws Exception {
        Path ledger = Path.of("external-ledger.txt");   // "외부 시스템" 역할

        Properties p = new Properties();
        p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        p.put(ConsumerConfig.GROUP_ID_CONFIG, "settler");
        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");
        p.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

        try (KafkaConsumer<String, String> c = new KafkaConsumer<>(p)) {
            c.subscribe(List.of("payment-approved"));
            while (true) {
                ConsumerRecords<String, String> records = c.poll(Duration.ofSeconds(1));
                for (ConsumerRecord<String, String> r : records) {
                    // ★ 멱등성 없는 외부 부작용
                    Files.writeString(ledger, "CHARGE " + r.key() + "\n",
                        StandardOpenOption.CREATE, StandardOpenOption.APPEND);
                }
                if (!records.isEmpty()) c.commitSync();
            }
        }
    }
}

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

# 1. 토픽 생성
docker exec -it broker $K/kafka-topics.sh --create --topic payment-approved \
  --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092

# 2. 결제 승인 이벤트 5건 (키 = 주문 ID)
for i in 1 2 3 4 5; do
  echo "ORD-$i:{\"orderId\":\"ORD-$i\",\"amount\":10000}" | \
  docker exec -i broker $K/kafka-console-producer.sh --topic payment-approved \
    --bootstrap-server localhost:9092 \
    --property parse.key=true --property key.separator=:
done

# 3. 컨슈머를 돌려 전부 처리한다
java -cp kafka-clients-4.3.1.jar:. SideEffectConsumer &
sleep 10 && kill %1
wc -l external-ledger.txt        # → 5

# 4. 오프셋을 되돌린다 (그룹이 Empty 여야 합니다)
docker exec -it broker $K/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group settler --topic payment-approved --reset-offsets --to-earliest --execute

# 5. 다시 돌린다 → 외부 원장에 같은 청구가 또 들어갑니다
java -cp kafka-clients-4.3.1.jar:. SideEffectConsumer &
sleep 10 && kill %1
wc -l external-ledger.txt        # → 10  ★ 이중 청구
sort external-ledger.txt | uniq -c
#   2 CHARGE ORD-1
#   2 CHARGE ORD-2
#   ...

# 1. 외부 시스템에서 "처리 완료된 키 목록"을 먼저 확보한다
psql -c "\copy (SELECT order_id FROM pg_confirmations
                WHERE created_at BETWEEN '2026-07-24 04:00' AND '2026-07-24 08:00')
         TO 'processed.csv' CSV"

# 2. 전체 구간을 되돌리는 대신, 별도 그룹으로 재처리 전용 컨슈머를 띄운다.
#    그 컨슈머가 processed.csv 를 읽어 이미 처리된 키를 건너뛰게 만듭니다.
kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
  --group settler-replay-20260724 --topic payment-approved \
  --reset-offsets --to-datetime 2026-07-24T04:00:00.000 --execute

# 3. 재처리 컨슈머는 반드시 dry-run 모드를 먼저 지원해야 합니다.
#    "몇 건이 실제로 외부 호출 대상인가"를 사람이 확인한 뒤 실행합니다.

CREATE TABLE processed_events (
  event_key    TEXT PRIMARY KEY,      -- 예: 'payment-approved:ORD-77412:1'
  topic        TEXT NOT NULL,
  partition_id INT  NOT NULL,
  "offset"     BIGINT NOT NULL,
  processed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  result       JSONB
);

void handle(ConsumerRecord<String, PaymentApproved> record) {
    String eventKey = eventKeyOf(record);   // 비즈니스 식별자 기반

    // ① 선점: 이미 처리했으면 여기서 끝난다 (0 rows affected)
    int claimed = jdbc.update(
        "INSERT INTO processed_events (event_key, topic, partition_id, \"offset\") " +
        "VALUES (?,?,?,?) ON CONFLICT (event_key) DO NOTHING",
        eventKey, record.topic(), record.partition(), record.offset());

    if (claimed == 0) {
        metrics.counter("settlement.skipped_duplicate").increment();
        return;                              // ★ 외부 호출을 하지 않습니다
    }

    // ② 외부 호출
    PgResult result = pgClient.confirmPayment(...);

    // ③ 결과 기록
    jdbc.update("UPDATE processed_events SET result = ?::jsonb WHERE event_key = ?",
                toJson(result), eventKey);
}

payment-approved
   │
   ├─(Streams, exactly_once_v2)─→ payment-command   ← Kafka 안. EOS 온전히 성립
   │                                    │
   │                                    └─(전용 컨슈머 1개)─→ PG API
   │                                          · 멱등 키 사용
   │                                          · 처리 이력 테이블
   │                                          · 이 컨슈머만 검증하면 됨
   └─→ ... 다른 소비자들