Kafka Producer: acks、retry、idempotenceの正しい使い方

1. 概要

producerがKafkaにメッセージを送信する際、答える必要のある3つの質問があります。送信完了と見なすのはいつか、エラーが発生した場合にどのようにリトライするか、そしてリトライはメッセージの重複を引き起こさないかということです。この記事では、acks、retries、enable.idempotenceについて説明します。これは、producerの信頼性と正確性を決定する3つの設定です。最後に、本番環境に対応したproducerの設定方法を学び、データ損失または重複を引き起こす一般的なエラーを回避できるようになります。

2. 必要な依存関係

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.7.0</version>
</dependency>

この記事の例では、Spring Kafkaを使わず、純粋なKafka client（KafkaProducer）を使用して、各設定がどのように影響するかを明確に理解します。

3. acks: producerが送信成功と見なす時点

acksはproducerがメッセージ送信完了と見なすまでに何個のbrokerから確認を待つかを決定します。

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");

• acks=0: producerは確認を待ちません。最速ですが、受信直後にbrokerがダウンするとメッセージが失われます。失われたことさえ知られません。
• acks=1: leader partitionがローカルログに書き込みACKを返すのを待つだけです。leaderがfollowerの同期前にクラッシュするとメッセージが失われます。
• acks=all（または-1）: ISR（in-sync replica set）内のすべてのreplicaから確認を待ちます。最も安全ですが、レイテンシが高くなります。

注意: acks=allがbrokerのmin.insync.replicas >= 2と一緒に設定されている場合のみ、本当に安全です。min.insync.replicas=1の場合、acks=allも1つのreplicaの確認だけで十分です。acks=1との安全性に差がありません。

3.1. 対応するbroker設定

# server.properties または topic config
min.insync.replicas=2

replication factor = 3およびmin.insync.replicas=2の場合、システムは1つのbrokerが故障した場合でも書き込み可能で、同時にACKの前に少なくとも2つのコピーにデータが存在することを保証します。

4. retries: 一時的なエラーの処理

producerはNotLeaderForPartitionExceptionや一時的なネットワーク接続の喪失などの一時的なエラーが発生する場合があります。retriesにより、producerは手動でコードを処理することなく自動的に再送できます。

props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

• retries: 最大リトライ回数。Kafka 2.1以降は非常に高い値（またはInteger.MAX_VALUE）に設定し、回数をカウントする代わりにdelivery.timeout.ms`を使用して待機時間を制御することが推奨されます。
• retry.backoff.ms: リトライ間の待機時間。brokerへのスパムを防ぎます。
• delivery.timeout.ms: send()の呼び出しから結果の受信（成功または最終エラー）までの最大総時間。request.timeout.ms + linger.msより大きい必要があります。

問題: 注意しないと、リトライはメッセージの重複送信または順序の乱れを引き起こす可能性があります。

4.1. リトライがなぜメッセージの重複を引き起こすのか

1. producerはメッセージAを含むbatchを送信
2. brokerはAに正常に書き込み、ACKを送信
3. ネットワーク上でACKが失われる（ネットワークタイムアウト）
4. producerはACKを受信しない -> エラーと見なす -> リトライ
5. brokerはリクエストを再度受信 -> Aを2回目に書き込む
=> topicにメッセージAの2つのコピーがある

これは古典的な状況です。brokerは正常に書き込みましたが、producerはそれを知らないため再送信します。後のセクションのidempotenceがこの問題を正しく解決します。

4.2. リトライがなぜ順序を乱すのか

max.in.flight.requests.per.connection > 1の場合、producerは複数のbatchを並行して送信できます。batch 1がエラーを起こしてリトライ中の場合、batch 2がbatch 1より先にbrokerに書き込まれる可能性があります。

max.in.flight.requests.per.connection=5

この設定でidempotenceなしの場合、リトライが発生するとpartition内のメッセージの順序は保証されません。

5. Idempotence: リトライによる重複の解決

enable.idempotence=trueにより、producerはリトライ時の重複書き込みを自動的に防ぐことができます。Kafkaが各batchにProducer ID（PID）とsequence numberを割り当てるおかげです。

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

idempotenceを有効にすると、Kafkaは自動的に以下の値を設定します（自分で設定する必要はありませんが、知っておくべきです）:

acks = all
retries = Integer.MAX_VALUE (または非常に大きい値)
max.in.flight.requests.per.connection <= 5

5.1. メカニズムの動作

各producerセッションは、brokerから一意のPIDが割り当てられます。1つのpartition内に送信される各メッセージには、増加するsequence numberがあります。brokerは各PID+partitionの最後に書き込まれたsequence numberを保存します。

producerはpartition 0に対してseq=5のbatchを送信
broker: 「最後に保存したsequenceは4で、5が次です -> 書き込み、seq=5を保存」
producerは（リトライ原因で）seq=5のbatchを再送信
broker: 「seq=5はすでに書き込んだ -> スキップして、常にACK成功を返す」

このように、リトライは重複レコードを作成しません。brokerはsequence numberに基づいてduplicateを自動的に検出して削除するため、producerやconsumerが自分で処理する必要はありません。

注意: idempotenceは単一のproducerセッション内での重複がないことと単一のpartition内を保証するだけです。producerが再起動する場合（新しいPIDが割り当てられます）、古いPIDに基づくdedup機構は適用されなくなります。複数のpartition/topic全体でexactly-onceが必要な場合、transaction（transactional.id）が必要です。これはシンプルなidempotenceの範囲を超えています。

5.2. 本番環境対応のproducer設定例

public ProducerFactory() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // idempotenceを有効化 -> 自動的にacks=all、高リトライ、in-flight <= 5を設定
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

    this.producer = new KafkaProducer<>(props);
}

public void sendOrderEvent(String orderId, String payload) {
    ProducerRecord<String, String> record =
            new ProducerRecord<>("order-events", orderId, payload);

    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            log.error("注文 {} の送信失敗", orderId, exception);
        } else {
            log.info("注文 {} がpartition {} offset {} に書き込まれました",
                    orderId, metadata.partition(), metadata.offset());
        }
    });
}

• ENABLE_IDEMPOTENCE_CONFIG = trueが最も重要な行です。それが自動的にacks=allをもたらします。
• send()内のコールバックは実際の結果を示します。exception != nullの場合、delivery.timeout.msの再試行を使い果たしてもエラーが発生したことを意味します。この時点で、ビジネスロジックを処理する必要があります（ログ、アラート、エラーテーブルへの保存）。

6. テストでの動作確認

dedup部分を説明するため、applicationレベルでsequence numberメカニズムをシミュレートする単純なMessageDeduplicatorを実装します（idempotenceを有効にできない場合、例えばレガシーシステムの場合に使用）:

public class MessageDeduplicator {

    private final Map<String, Long> lastSeenSequence = new ConcurrentHashMap<>();

    public boolean shouldWrite(String partitionKey, long sequence) {
        Long last = lastSeenSequence.get(partitionKey);
        if (last != null && sequence <= last) {
            return false; // duplicate, スキップ
        }
        lastSeenSequence.put(partitionKey, sequence);
        return true;
    }
}

class MessageDeduplicatorTest {

    @Test
    void shouldSkipDuplicateSequence() {
        MessageDeduplicator deduplicator = new MessageDeduplicator();

        assertThat(deduplicator.shouldWrite("order-1", 5)).isTrue();
        assertThat(deduplicator.shouldWrite("order-1", 5)).isFalse(); // リトライで重複するseq
        assertThat(deduplicator.shouldWrite("order-1", 6)).isTrue();
    }

    @Test
    void shouldAllowDifferentPartitionKeysIndependently() {
        MessageDeduplicator deduplicator = new MessageDeduplicator();

        assertThat(deduplicator.shouldWrite("order-1", 1)).isTrue();
        assertThat(deduplicator.shouldWrite("order-2", 1)).isTrue();
    }
}

最初のテストは、同じsequenceが2回目に送信されたときにスキップされることを確認します。これはproducer idempotenceの本質に合致しています。2番目のテストは、dedupが1つのkey/partition内にのみ適用され、他のkeyに影響しないことを確認します。

7. ニーズに応じた設定サマリーテーブル

ニーズ	acks	enable.idempotence	リスク
重要でないログ、メトリック	0	false	エラー時にメッセージが失われるが、スループットが最大
速度と安全性のバランス	1	false	followerの同期前にleaderがクラッシュするとメッセージが失われる
重要なビジネスデータ（注文、支払い）	all	true	レイテンシが高くなるが、メッセージの損失なし、リトライによる重複なし

8. よくあるエラー

• acks=allを設定しているのにmin.insync.replicasを忘れている: 安全だと思っていますが、brokerは1つのISRのACKだけで十分です。acks=1と違いはありません。
• Kafka clientのデフォルトリトライと並行して独自のリトライロジックを記述している: 両方のリトライレイヤーが機能するため、メッセージの重複が2倍になる可能性があります。
• idempotenceを有効にしているのにmax.in.flight.requests.per.connection > 5を設定している: Kafkaはproducer初期化時にこの設定を拒否します（ConfigException）。idempotenceがサポートしている制限を超えています。
• idempotenceを全体的なexactly-onceと混同している: idempotenceはproducer-to-brokerのみを保護します。consumerがメッセージを処理した後、非idempotentな副作用（メール送信、金額控除など）を呼び出す場合、consumer側で重複処理が発生する可能性があります。

注意: Kafkaバージョンが0.11より古い場合は、enable.idempotence=trueを有効にしないでください。この機能はまだbrokerでサポートされておらず、producer初期化時にエラーが発生します。

9. 結論

acks、retries、enable.idempotenceは、信頼できるproducerが必要な場合、切り離せない3つの要素です。acks=allは送信成功と見なす前にデータが十分にレプリケートされることを保証します。retriesは一時的なエラーを克服するのに役立ちます。idempotenceはリトライが重複レコードを作成しないことを保証します。重要なビジネスデータの場合は、デフォルトで3つをすべて一緒に使用してください。レイテンシのコストは、後でデータエラーまたは重複を処理するコストよりはるかに小さいです。