Spring KafkaでのKafkaリトライとデッドレターコピック

1. 概要

Kafkaコンシューマーは確実にメッセージ処理エラーに遭遇します。一時的なエラーもあります。例えば、決済ゲートウェイのタイムアウト。恒久的なエラーもあります。例えば、データ形式が正しくない。適切に処理しないと、2つの状況のいずれかに直面します。メッセージの損失か、コンシューマーが成功することのないメッセージに永遠に引っかかることです。

このガイドでは、Spring Kafkaでのリトライ設定について、DefaultErrorHandlerを使用したブロッキングと@RetryableTopicを使用した非ブロッキングの2つの方法を説明します。リトライ回数を超えてもエラーが続いたメッセージは、デッドレターコピック(DLT)に送信されます。このガイドを読み終えると、どの方法をいつ選択するかが分かり、動作を検証するテストが書けるようになります。

2. セットアップ

Spring Kafkaの依存関係を追加します。Spring Boot 3.xはバージョンを自動的に管理するため、バージョンを明示的に指定する必要はありません。

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

application.propertiesでの最小限の設定は以下の通りです。

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=order-service
spring.kafka.consumer.auto-offset-reset=earliest

Boot はデフォルトでStringSerializerとStringDeserializerを使用します。この例を簡潔にするため、メッセージの値はString形式の注文コードです。

3. エラーが発生する可能性のあるConsumer

これは決済処理をするサービスです。形式が正しくない注文コードは恒久的なエラーです。決済ゲートウェイ呼び出しエラーは一時的なエラーです。

public class PaymentGatewayException extends RuntimeException {
    public PaymentGatewayException(String message) {
        super(message);
    }
}

@Service
public class PaymentService {

    public void charge(String orderId) {
        if (orderId == null || !orderId.startsWith("ORD-")) {
            throw new IllegalArgumentException("無効な注文コード: " + orderId);
        }
        // 決済ゲートウェイを呼び出します。タイムアウト時にPaymentGatewayExceptionを投げることがあります。
    }
}

リスナーはサービスを呼び出すだけです。

@Component
public class OrderListener {

    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);
    private final PaymentService paymentService;

    public OrderListener(PaymentService paymentService) {
        this.paymentService = paymentService;
    }

    @KafkaListener(topics = "orders")
    public void onOrder(String orderId) {
        log.info("注文{}を処理しています", orderId);
        paymentService.charge(orderId);
    }
}

何も設定しないと、Spring KafkaはDefaultErrorHandlerとFixedBackOff(0, 9)を使用します。エラーメッセージは連続して9回再試行され、合計10回試行されます。その後、ハンドラーはログに記録するだけでスキップします。メッセージは失われたと見なされます。これが設定し直す必要がある理由です。

4. DefaultErrorHandlerによるブロッキングリトライ

4.1. BackoffとDLTの設定

リスナーが例外をスローすると、DefaultErrorHandlerはエラーメッセージのオフセットをシークバックして、次のポーリング時に再度読み込みます。リトライ回数を超えると、ハンドラーはConsumerRecordRecovererを呼び出します。ここで、recovererはDeadLetterPublishingRecovererです。

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        var recoverer = new DeadLetterPublishingRecoverer(template);

        var backOff = new ExponentialBackOffWithMaxRetries(3);
        backOff.setInitialInterval(1_000L);
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(10_000L);

        var handler = new DefaultErrorHandler(recoverer, backOff);
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

各部分の説明は以下の通りです。

• DeadLetterPublishingRecoverer(template)はエラーメッセージを<トピック名>.DLTに送信し、元のメッセージと同じパーティションに送信します。ordersトピックの場合、DLTはorders.DLTです。
• ExponentialBackOffWithMaxRetries(3)は最大3回まで再試行します。待機時間は順に1秒、2秒、4秒で、10秒を超えません。
• addNotRetryableExceptionsは再試行が必要ない例外を宣言します。IllegalArgumentExceptionはDLTに直接進みます。何回再試行しても、データが正しくなければ正しくないままです。

Spring Bootは自動的にbean CommonErrorHandlerをデフォルトコンテナファクトリーに追加します。ConcurrentKafkaListenerContainerFactoryを自分で作成する必要はありません。

ORD-1を送信し、決済ゲートウェイが継続的にタイムアウトする場合、ログは以下のようになります。

10:00:00.010 INFO  注文ORD-1を処理しています
10:00:01.020 INFO  注文ORD-1を処理しています
10:00:03.030 INFO  注文ORD-1を処理しています
10:00:07.040 INFO  注文ORD-1を処理しています
10:00:07.060 ERROR Backoff ... exhausted for orders-0@42

合計4回の処理があります。1回目と3回の再試行です。その後、メッセージはorders.DLTに存在します。

4.2. DLTのヘッダー

DLT内のメッセージは、キーと値をそのまま保持します。recovererは調査用にヘッダーを追加します。

kafka_dlt-original-topic      = orders
kafka_dlt-original-partition  = 0
kafka_dlt-original-offset     = 42
kafka_dlt-exception-fqcn      = org.springframework.kafka.listener.ListenerExecutionFailedException
kafka_dlt-exception-cause-fqcn = com.example.PaymentGatewayException
kafka_dlt-exception-message   = Listener method ... threw exception
kafka_dlt-exception-stacktrace = ...

元々の例外はkafka_dlt-exception-cause-fqcnに記載されています。コードでヘッダーを読む必要がある場合は、KafkaHeadersの定数を使用します。例えばKafkaHeaders.DLT_ORIGINAL_TOPICです。

注意: recovererはデフォルトで同じパーティションに送信するため、orders.DLTはordersと同じ数のパーティションが必要です。そうでない場合、公開は失敗します。bean NewTopicを使用して同じ数のパーティションでDLTを作成することができます。別の方法として、new TopicPartition(r.topic() + ".DLT", -1)を返すdestination resolverを渡して、Kafkaに自動的にパーティションを選択させることもできます。

4.3. ブロッキングリトライが適切な場合

ブロッキングリトライはメッセージの順序を保持します。後のメッセージは、前のメッセージが成功するか既にDLTに移動してからのみ処理されます。これは順序が重要な場合に正しい選択です。例えば、同じ注文のOrderCreated → OrderPaid → OrderShippedという一連のイベント。

代償として、待機中にパーティーション全体がブロックされます。Spring Kafka 2.9以降、ハンドラーはbackoffを待機する際にmax.poll.interval.msを超えません。これはseekとpauseメカニズムのおかげです。それでも、backoffを合計数分に設定すると、lagが急増します。

5. @RetryableTopicによる非ブロッキングリトライ

5.1. アノテーションによる設定

非ブロッキングリトライでは、エラーメッセージは別のリトライトピックに送信されます。これにより、コンシューマーは次のメッセージをすぐに処理できます。別のコンシューマーが指定されたbackoffの後にリトライトピックを読みます。

@Component
public class OrderRetryListener {

    private static final Logger log = LoggerFactory.getLogger(OrderRetryListener.class);
    private final PaymentService paymentService;

    public OrderRetryListener(PaymentService paymentService) {
        this.paymentService = paymentService;
    }

    @RetryableTopic(
            attempts = "4",
            backoff = @Backoff(delay = 1_000, multiplier = 2.0, maxDelay = 10_000),
            exclude = IllegalArgumentException.class)
    @KafkaListener(topics = "payments")
    public void onPayment(String orderId,
                          @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        log.info("{} から{}を処理しています", orderId, topic);
        paymentService.charge(orderId);
    }

    @DltHandler
    public void onDlt(String orderId,
                      @Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) {
        log.error("注文{}がDLTに移動しました: {}", orderId, error);
    }
}

主な点は以下の通りです。

• attempts = "4"は合計4回の処理を意味します。元のトピックで1回、リトライトピックで3回です。値はStringで、"${app.retry.attempts}"などのプレースホルダーをサポートします。
• @Backoffはorg.springframework.retry.annotationパッケージから取得されます。Spring Retryは既にspring-kafkaに含まれています。
• excludeは直接DLTに移動する例外を列挙します。これはaddNotRetryableExceptionsと同等です。
• @DltHandlerは同じクラス内に配置されます。ここはログ記録、アラート送信、手動処理用のテーブルへの保存などを行う場所です。

5.2. 作成されるトピック

段階的なbackoffを使用すると、デフォルトではリトライごとに個別のトピックが作成され、遅延に応じてサフィックスが付きます。

payments
payments-retry-1000
payments-retry-2000
payments-retry-4000
payments-dlt

Spring Kafkaは起動時にこれらのトピックを自動的に作成します(autoCreateTopics = "true"はデフォルトです)。本番環境では、ブローカー上の自動作成を無効にすることが一般的です。その場合、正しい名前とパーティション数でトピックを事前に作成する必要があります。遅延の代わりにインデックス番号を使用したい場合は、topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUEを設定します。

サフィックスに注意してください。@RetryableTopicは-dltを使用し、DeadLetterPublishingRecovererは.DLTを使用します。モニタリング設定時にこの点を覚えておいてください。

5.3. 非ブロッキングのトレードオフ

非ブロッキングリトライは順序を破壊します。ORD-1がエラーになってリトライトピックに移動し、ORD-2が即座に成功する場合を想定してください。この場合、ORD-2はORD-1より前に処理されます。ビジネスロジックがイベントの順序に依存する場合、この方法は使用しないでください。

さらに、トピック数が増加し、コンシューマー数も多くなります。代わりに、1つの「厄介な」メッセージがパーティーション全体を遅延させることはありません。

6. DLTフローのテスト

以下のテストは@EmbeddedKafkaを使用します。形式が正しくない注文コードを送信し、そのメッセージがorders.DLTに表示されることを確認し、再試行されないことを検証します。

@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(partitions = 1, topics = {"orders", "orders.DLT"})
class OrderDltTest {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private EmbeddedKafkaBroker broker;

    @Test
    void givenInvalidOrderId_whenConsumed_thenSentToDltWithoutRetry() {
        kafkaTemplate.send("orders", "order-key", "INVALID-99");

        Map<String, Object> props = KafkaTestUtils.consumerProps("dlt-test", "false", broker);
        try (Consumer<String, String> consumer = new DefaultKafkaConsumerFactory<>(
                props, new StringDeserializer(), new StringDeserializer()).createConsumer()) {

            broker.consumeFromAnEmbeddedTopic(consumer, "orders.DLT");
            ConsumerRecord<String, String> record =
                    KafkaTestUtils.getSingleRecord(consumer, "orders.DLT", Duration.ofSeconds(10));

            assertThat(record.value()).isEqualTo("INVALID-99");
            assertThat(record.key()).isEqualTo("order-key");

            String cause = new String(record.headers()
                    .lastHeader(KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN).value());
            assertThat(cause).isEqualTo(IllegalArgumentException.class.getName());

            String originalTopic = new String(record.headers()
                    .lastHeader(KafkaHeaders.DLT_ORIGINAL_TOPIC).value());
            assertThat(originalTopic).isEqualTo("orders");
        }
    }
}

テストは以下のように動作します。

• ${spring.embedded.kafka.brokers}は、アプリケーションを埋め込みブローカーに向けます。
• KafkaTestUtils.consumerPropsはauto.offset.reset=earliestでコンシューマー設定を作成するため、テストはコンシューマーが購読する前に公開されたメッセージを見落とすことはありません。
• IllegalArgumentExceptionは再試行不可のため、メッセージはほぼすぐにDLTに到達します。テストはbackoff チェーン全体を待つ必要がありません。
• アサーションは3つのことを検証します。キーと値が保持されている、ヘッダーが元の例外を記録している、そしてヘッダーが正しいソーストピックを記録しています。

リトライフローをテストしたい場合は、backoff値をプロパティーに出します。テストでは、テストを高速に実行するために待機時間を数十ミリ秒に設定します。

7. よくあるエラー

• 例外を分類しない。 NullPointerExceptionやvalidation エラーのリトライは時間の無駄です。どの例外が再試行されるべきか、どの例外がDLTに直接行くべきかを明確に宣言します。DeserializationExceptionのようなdeserialize エラーはデフォルトで再試行不可です。しかし、ハンドラーがこのエラーを受け取るには、deserializerをErrorHandlingDeserializerでラップする必要があります。
• KafkaTemplateのSerializer が一致しない。 Recovererは既にdeserialize されたvalueを再公開します。コンシューマーがJSONをオブジェクトに読み込みながら、template がStringSerializerを使用する場合、DLTへの送信は失敗します。templateに合ったserializerを使用します。deserialize エラーのメッセージの場合、valueはbyte[]なので、ByteArraySerializerが必要です。
• リスナーがidempotent でない。 リトライは同じメッセージが複数回処理される可能性があることを意味します。既にchargeが実行されてからタイムアウトする場合、リトライは再度charge を実行します。idempotencyキーを使用して重複排除します。例えば、処理済みのorderIdを保存します。
• 誰もDLTを読まない。 DLTはゴミ箱ではありません。DLT にメッセージが新たに到着したときに警告する必要があり、エラーが修正された後にリプレイするプロセスが必要です。
• ブロッキングリトライでのbackoff時間が長すぎる。 パーティーション全体がその時間ブロックされます。合計待機時間が数十秒を超える場合は、非ブロッキングへの切り替えを検討します。

注意: キーに基づいて厳密な順序が必要なフローに@RetryableTopicを使用しないでください。逆に、スループットが高いトピックに長いbackoff を伴うブロッキングリトライを使用しないでください。1つのエラーメッセージにより、lagが数万増加する可能性があります。

8. 結論

このガイドでは、Spring Kafkaでメッセージエラーを処理する2つの方法を説明しました。DefaultErrorHandlerとDeadLetterPublishingRecovererを組み合わせたブロッキングリトライは、順序を保持しますがパーティションをブロックします。@RetryableTopicと@DltHandlerによる非ブロッキングリトライは、コンシューマーを遅延させませんが、順序を牺牲にします。

どちらの方法を選択しても、4つのことを行うべきです。再試行可能な例外と再試行不可の例外を分類し、リスナーがidempotent であることを確認し、適切なパーティション数でDLTを作成し、DLTを定期的に監視します。6番目のセクションのような@EmbeddedKafkaを使用したテストは、本番環境に移行する前に設定が正しく機能することを確認するのに役立ちます。