Outbox ve Idempotent Consumer — Olay Kaybolmasın, İki Kez de İşlenmesin

1
CREATE TABLE outbox_events (
2
    id             UUID         PRIMARY KEY,
3
    aggregate_type VARCHAR(50)  NOT NULL,
4
    aggregate_id   VARCHAR(50)  NOT NULL,      -- becomes the Kafka key: per-account ordering
5
    event_type     VARCHAR(100) NOT NULL,
6
    payload        JSONB        NOT NULL,
7
    created_at     TIMESTAMPTZ  NOT NULL DEFAULT now(),
8
    published_at   TIMESTAMPTZ
9
);
10
CREATE INDEX outbox_unpublished ON outbox_events (created_at) WHERE published_at IS NULL;
11

12
-- On the consumer side (another service, another database)
13
CREATE TABLE processed_events (
14
    event_id     UUID        PRIMARY KEY,
15
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
16
);

1
@Service
2
class TransferService {
3

4
    private final Ledger ledger;
5
    private final OutboxRepository outbox;
6
    private final ObjectMapper json;
7

8
    TransferService(Ledger ledger, OutboxRepository outbox, ObjectMapper json) {
9
        this.ledger = ledger;
10
        this.outbox = outbox;
11
        this.json = json;
12
    }
13

14
    @Transactional
15
    public long transfer(TransferCommand cmd) {
16
        Transfer transfer = ledger.post(cmd);    // debit and credit postings
17
        // Same transaction, same database: the money moves and the event is recorded,
18
        // or neither happens. No kafka.send() here — that would be the dual write
19
        // this pattern removes: money moved, but no one else ever hears about it.
20
        outbox.save(OutboxEvent.of("Account", cmd.fromIban(), "TransferCompleted",
21
                json.valueToTree(TransferCompleted.from(transfer))));
22
        return transfer.getId();
23
    }
24
}

1
interface OutboxRepository extends JpaRepository<OutboxEvent, UUID> {
2

3
    // Each pod locks its own batch; rows locked by another pod are skipped, not waited on.
4
    @Query(value = """
5
            SELECT * FROM outbox_events
6
            WHERE published_at IS NULL
7
            ORDER BY created_at
8
            LIMIT :batch
9
            FOR UPDATE SKIP LOCKED
10
            """, nativeQuery = true)
11
    List<OutboxEvent> lockNextBatch(int batch);
12
}

1
@Component
2
class OutboxRelay {
3

4
    private final OutboxRepository outbox;
5
    private final KafkaTemplate<String, String> kafka;
6

7
    OutboxRelay(OutboxRepository outbox, KafkaTemplate<String, String> kafka) {
8
        this.outbox = outbox;
9
        this.kafka = kafka;
10
    }
11

12
    @Scheduled(fixedDelay = 500)
13
    @Transactional
14
    public void relay() {
15
        for (OutboxEvent event : outbox.lockNextBatch(100)) {
16
            var record = new ProducerRecord<>("transfers", event.aggregateId(), event.payload().toString());
17
            record.headers().add("event-id", event.id().toString().getBytes(StandardCharsets.UTF_8));
18
            record.headers().add("event-type", event.eventType().getBytes(StandardCharsets.UTF_8));
19
            // Wait for the broker's ack before marking. A crash between send and commit
20
            // means the event is sent again: at-least-once, so consumers must be idempotent.
21
            kafka.send(record).join();
22
            event.markPublished(Instant.now());
23
        }
24
    }
25
}

1
@Component
2
// The limits service tracks how much of the customer's daily transfer limit is used.
3
class TransferCompletedConsumer {
4

5
    private final JdbcTemplate jdbc;
6
    private final DailyLimits limits;
7

8
    TransferCompletedConsumer(JdbcTemplate jdbc, DailyLimits limits) {
9
        this.jdbc = jdbc;
10
        this.limits = limits;
11
    }
12

13
    @KafkaListener(topics = "transfers", groupId = "limits")
14
    @Transactional
15
    public void on(@Payload TransferCompleted event, @Header("event-id") String eventId) {
16
        int firstTime = jdbc.update(
17
                "INSERT INTO processed_events (event_id) VALUES (?::uuid) ON CONFLICT DO NOTHING", eventId);
18
        if (firstTime == 0) return;                   // duplicate delivery: already counted
19

20
        // Same transaction as the marker: if the update fails, the marker rolls back too
21
        // and the redelivered event is processed again.
22
        limits.addUsage(event.customerId(), event.valueDate(), event.amount());
23
    }
24
}