KafkaとのOutboxパターン：データベース書き込みでイベントを失わない

1. 概要

通常、1つのサービスは2つの連続した操作を行う必要があります：データをデータベースに保存し、イベントをKafkaに発行して他のサービスに通知することです。これら2つのシステムは共通のトランザクションを持ちません。一方の操作が成功し、もう一方が失敗した場合、システムは一貫性を失います。その結果、注文は保存されていますが、誰もOrderPlacedイベントを受け取りません。

本記事ではdual writeの問題を説明し、Java 21、Spring Boot 3.x、Spring Kafkaを使用してOutbox patternを実装します。本記事を読み終えた後、outboxテーブルからKafkaにイベントを送信するメッセージRelayと、テストおよびべき等なコンシューマーを実装できるようになります。

2. Dual writeの問題

以下の「素朴な」実装方法を見てください：

@Transactional
public Order placeOrder(String customerId, BigDecimal amount) {
    Order order = orderRepository.save(new Order(customerId, amount));
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
    return order;
}

このコードには少なくとも2つのエラーシナリオがあります：

Kafkaの送信は成功したが、トランザクションがロールバックされる。 例えば、コミット中のエラーまたは制約違反。存在しない注文の「ゴースト」イベントが外部に発行されている。
トランザクションがコミットされたが、Kafkaの送信が失敗する。 ブローカーにアクセスできないか、コミット直後にアプリケーションがクラッシュする可能性があります。注文はデータベース内にありますが、イベントは消失します。

2つのコマンドの順序を変更しても根本的な問題は解決されません。2つの異なるシステム上の2つの操作は、分散トランザクションなしではアトミックにできません。ほとんどのチームは複雑性とパフォーマンスの理由からXA/2PCを使用したくないのです。

3. Outbox patternの考え方

Outbox patternは「データベース書き込み+Kafka送信」を「データベース書き込み+データベース書き込み」に変えます：

同じ1つのトランザクション内で、サービスはOrderを保存し、outbox_eventテーブルに1行を挿入します。
message relayと呼ばれる独立したプロセスがoutboxテーブルから未送信の行を読み取り、それらをKafkaに送信してから、送信済みとしてマークします。

ステップ1は1つのデータベースのみに関係するため、完全にアトミックです：注文があれば、必ずイベントがあります。ステップ2は何度でも再試行できます。その代わりに、at-least-onceセマンティクスを得ます：イベントは重複して送信される可能性がありますが、失われることはありません。

4. Spring Bootでの実装

4.1. 依存関係

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

次に、安全なプロデューサー設定です：

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.acks=all
spring.kafka.producer.properties.enable.idempotence=true

acks=allは、すべてのレプリカが書き込んだ後にブローカーが確認応答することを強制します。enable.idempotence=trueは、プロデューサーがネットワーク層での再試行時に重複を生成しないようにします。

4.2. Outboxテーブルとエンティティ

CREATE TABLE outbox_event (
    id             UUID PRIMARY KEY,
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id   VARCHAR(100) NOT NULL,
    event_type     VARCHAR(100) NOT NULL,
    payload        TEXT         NOT NULL,
    created_at     TIMESTAMP    NOT NULL,
    published_at   TIMESTAMP
);
CREATE INDEX idx_outbox_unpublished ON outbox_event (created_at) WHERE published_at IS NULL;

部分的なインデックスには未送信の行のみが含まれます。これにより、テーブルに数百万の古い行があっても、relayのクエリは常に高速です。

@Entity
@Table(name = "outbox_event")
public class OutboxEvent {

    @Id
    private UUID id;
    private String aggregateType;
    private String aggregateId;
    private String eventType;
    @Column(columnDefinition = "text")
    private String payload;
    private Instant createdAt;
    private Instant publishedAt;

    protected OutboxEvent() {
    }

    public OutboxEvent(String aggregateType, String aggregateId, String eventType, String payload) {
        this.id = UUID.randomUUID();
        this.aggregateType = aggregateType;
        this.aggregateId = aggregateId;
        this.eventType = eventType;
        this.payload = payload;
        this.createdAt = Instant.now();
    }

    public void markPublished(Instant at) {
        this.publishedAt = at;
    }

    public UUID getId() { return id; }
    public String getAggregateId() { return aggregateId; }
    public String getEventType() { return eventType; }
    public String getPayload() { return payload; }
    public Instant getPublishedAt() { return publishedAt; }
}

idはevent idとしても機能します。コンシューマーはこれを使用して重複を除外します（セクション6）。aggregateIdはKafkaメッセージのキーになります。

4.3. 同じトランザクション内でOrderとイベントを書き込む

public record OrderPlacedEvent(Long orderId, String customerId, BigDecimal amount) {
}

@Service
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;
    private final ObjectMapper objectMapper;

    public OrderService(OrderRepository orderRepository,
                        OutboxEventRepository outboxRepository,
                        ObjectMapper objectMapper) {
        this.orderRepository = orderRepository;
        this.outboxRepository = outboxRepository;
        this.objectMapper = objectMapper;
    }

    @Transactional
    public Order placeOrder(String customerId, BigDecimal amount) {
        Order order = orderRepository.save(new Order(customerId, amount));

        var event = new OrderPlacedEvent(order.getId(), customerId, amount);
        outboxRepository.save(new OutboxEvent(
                "Order", order.getId().toString(), "OrderPlaced", toJson(event)));

        return order;
    }

    private String toJson(Object value) {
        try {
            return objectMapper.writeValueAsString(value);
        } catch (JsonProcessingException e) {
            throw new IllegalStateException("イベントをシリアライズできません", e);
        }
    }
}

重要なポイントは@Transactionalにあります。2つのsave呼び出しは一緒にコミットするか、一緒にロールバックします。OrderServiceはKafkaについて知らなくなったため、トランザクション内にネットワーク呼び出しがありません。

4.4. KafkaにイベントをプッシュするMessage relay

まず、行ロックのあるクエリを持つリポジトリを作成します：

public interface OutboxEventRepository extends JpaRepository<OutboxEvent, UUID> {

    @Query(value = """
            SELECT * FROM outbox_event
            WHERE published_at IS NULL
            ORDER BY created_at
            LIMIT :batchSize
            FOR UPDATE SKIP LOCKED
            """, nativeQuery = true)
    List<OutboxEvent> findBatchToPublish(@Param("batchSize") int batchSize);
}

FOR UPDATE SKIP LOCKED（PostgreSQL、MySQL 8+）により、複数のアプリケーションインスタンスを同時に実行できます。各インスタンスは独自のバッチをロックし、他のインスタンスが処理中の行をスキップします。これにより、2つのインスタンスが同じイベントを並行して送信することはありません。

次に、relayです：

@Component
public class OutboxRelay {

    private static final Logger log = LoggerFactory.getLogger(OutboxRelay.class);

    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OutboxRelay(OutboxEventRepository outboxRepository,
                       KafkaTemplate<String, String> kafkaTemplate) {
        this.outboxRepository = outboxRepository;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(fixedDelay = 1000)
    @Transactional
    public void publishPending() {
        for (OutboxEvent event : outboxRepository.findBatchToPublish(100)) {
            try {
                kafkaTemplate.send(toRecord(event)).get(5, TimeUnit.SECONDS);
                event.markPublished(Instant.now());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            } catch (Exception e) {
                log.warn("イベント{}の送信に失敗しました。再試行します", event.getId(), e);
                break; // 順序を保つために停止し、次の実行で再試行します
            }
        }
    }

    private ProducerRecord<String, String> toRecord(OutboxEvent event) {
        var record = new ProducerRecord<>("orders", event.getAggregateId(), event.getPayload());
        record.headers()
              .add("eventId", event.getId().toString().getBytes(StandardCharsets.UTF_8))
              .add("eventType", event.getEventType().getBytes(StandardCharsets.UTF_8));
        return record;
    }
}

各部分を順番に説明します：

@Scheduled(fixedDelay = 1000)は、前回の実行完了の1秒後にrelayを実行します。設定クラスに@EnableSchedulingを追加することを忘れずに。
.get(5, TimeUnit.SECONDS)はブローカーの確認応答を待ちます。relayは、Kafkaがメッセージを受け取ったことが確認されてからのみpublishedAtをマークします。
markPublishedは管理対象のエンティティを変更します。JPAはトランザクションコミット時に自動的にUPDATEコマンドを生成するため、saveを呼び出す必要はありません。
エラーが発生した場合、relayはその行をスキップするのではなくbreakします。同じ注文のイベントは順序が変わりません。
キーはaggregateIdなので、1つのOrderのすべてのイベントは同じパーティションに入り、Kafkaで順序が保持されます。

ブローカーが一時的に停止した場合のログは次のようになります：

WARN  OutboxRelay : イベント3f6c1e0a-...の送信に失敗しました。再試行します
org.apache.kafka.common.errors.TimeoutException: Topic orders not present in metadata after 5000 ms.

ブローカーが復旧すると、次の実行がその正確なイベントから送信を再開します。

注意： アプリケーションがKafkaがメッセージを受け取った後しかしトランザクションコミット前にクラッシュした場合、outboxの行は「未送信」のままです。次の実行で再度送信されます。これがat-least-onceセマンティクスであり、コンシューマーは重複を処理する必要があります。

5. テスト

最初のテストでは、各注文が正確に1行のoutboxを生成することを確認します：

@DataJpaTest
@Import({OrderService.class, JacksonAutoConfiguration.class})
class OrderServiceTest {

    @Autowired
    OrderService orderService;

    @Autowired
    OutboxEventRepository outboxRepository;

    @Test
    void placeOrder_shouldWriteOutboxEventInSameTransaction() {
        Order order = orderService.placeOrder("customer-42", new BigDecimal("150000"));

        List<OutboxEvent> events = outboxRepository.findAll();

        assertThat(events).hasSize(1);
        OutboxEvent event = events.getFirst();
        assertThat(event.getAggregateId()).isEqualTo(order.getId().toString());
        assertThat(event.getEventType()).isEqualTo("OrderPlaced");
        assertThat(event.getPayload()).contains("\"customerId\":\"customer-42\"");
        assertThat(event.getPublishedAt()).isNull();
    }
}

リポジトリのネイティブクエリは起動時にチェックされません。したがって、H2はSKIP LOCKEDを完全にサポートしていなくても、このテストを実行できます。実際のクエリでrelayをテストするには、Testcontainersを使用してPostgreSQLを使用する必要があります。

2番目のテストでは、Kafkaが失敗したときにrelayが送信済みとしてマークされないことを確認します：

class OutboxRelayTest {

    OutboxEventRepository repository = mock(OutboxEventRepository.class);
    KafkaTemplate<String, String> kafkaTemplate = mock(KafkaTemplate.class);
    OutboxRelay relay = new OutboxRelay(repository, kafkaTemplate);

    @Test
    void shouldKeepEventUnpublished_whenKafkaFails() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.failedFuture(new RuntimeException("broker down")));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNull();
    }

    @Test
    void shouldMarkPublished_whenKafkaAcknowledges() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.completedFuture(null));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNotNull();
    }
}

これら2つのテストは、ブローカーが確認応答した場合にのみ送信済みとしてマークするという、コア動作をロックします。

6. コンシューマー側：べき等な処理

重複を受け取る可能性があるため、コンシューマーは処理済みのeventIdを覚えておく必要があります。この記録は、ビジネスロジックと同じトランザクション内に配置する必要があります：

@Component
public class InventoryListener {

    private final ProcessedEventRepository processedEvents;
    private final InventoryService inventoryService;

    public InventoryListener(ProcessedEventRepository processedEvents,
                             InventoryService inventoryService) {
        this.processedEvents = processedEvents;
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = "orders", groupId = "inventory-service")
    @Transactional
    public void onOrderPlaced(@Header("eventId") String eventId, String payload) {
        if (processedEvents.existsById(eventId)) {
            return; // 既に処理済み、重複をスキップ
        }
        inventoryService.reserveStock(payload);
        processedEvents.save(new ProcessedEvent(eventId));
    }
}

processed_eventテーブルはevent_idをプライマリキーとして設定する必要があります。2つのスレッドが同じ重複を処理しようとした場合、制約により1つはブロックされます。

7. よくあるエラーと使用すべきでない場合

OrderService内でKafkaを呼び出すことで「高速化」する。 これはdual writeに戻ることになります。すべてのイベントはoutboxテーブルを経由する必要があります。
outboxテーブルのクリーンアップを忘れる。 テーブルは非常に急速に増大します。数日以上前のpublished_atを持つ行を削除するジョブを実施してください。
エラー行をスキップして続行する。 これはイベントの順序を破壊します。ペイロードが永続的に破損している場合は、調査のためにそれをFAILED状態に移し、無限に再試行しないでください。
ポーリングが頻繁すぎる。 複数のインスタンスでの100 msのサイクルはデータベースに大きな負荷をかけます。通常、0.5～2秒のサイクルで十分です。
非常に低いレイテンシーまたは非常に高いスループットが必要。 この場合、ポーリングをCDC（Change Data Capture）に置き換える必要があります。例えば、Debeziumはメッセージログ/binlogを読み取り、outboxの行をKafkaに直接プッシュします。outboxテーブルはそのままで、relayだけが変わります。

注意： イベントが重要でない場合、例えば失われても構わないメトリクスまたはユーザー行動ログの場合は、Outbox patternを使用しないでください。この場合、Kafkaに直接送信する方が簡単です。逆に、Kafkaトランザクション（transactional.id）はOutboxを置き換えることはできません：複数のKafkaトピック間のアトミック性のみを保証し、データベースは含まれません。

8. 結論

データベースとKafkaの間のdual writeには常に、イベントを失ったり重複させたりする隙間があります。Outbox patternはこの隙間を3つのステップで塞ぎます。第1に、ビジネスデータと同じトランザクション内でイベントをoutboxテーブルに書き込みます。第2に、イベントをKafkaにプッシュし、ブローカーが確認応答したときのみ送信済みとしてマークする独立したrelayを使用します。第3に、at-least-onceセマンティクスの重複を吸収するために、コンシューマーでべき等な処理を実装します。

必要なピースはすべて揃っています：部分インデックス付きスキーマ、Kafkaに依存しないOrderService、複数のインスタンスで実行できるFOR UPDATE SKIP LOCKEDを使用するrelay、コア動作のテスト、および重複を防ぐコンシューマー。システムが大きくなると、ビジネスコードを変更せずにrelayのポーリングをCDCに置き換えることができます。

KafkaとのOutboxパターン：データベース書き込みでイベントを失わない

1. 概要

2. Dual writeの問題

3. Outbox patternの考え方

4. Spring Bootでの実装

4.1. 依存関係

4.2. Outboxテーブルとエンティティ

4.3. 同じトランザクション内でOrderとイベントを書き込む

4.4. KafkaにイベントをプッシュするMessage relay

5. テスト

6. コンシューマー側：べき等な処理

7. よくあるエラーと使用すべきでない場合

8. 結論

同じトピックの記事

Kafka の Consumer group と Rebalance (Spring Boot)

Change Data CaptureとDebeziumとKafka

KafkaでのSagaパターンによる分散トランザクション：ツアー予約の例

Kafkaの監視: consumer lagと監視対象のメトリクス

Kafkaプロデューサー: acks, idempotence, batching