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

1. 概要

一つのメッセージをKafkaに送信することはsend()の一行で済みます。難しい部分は付属する質問にあります：メッセージが本当に保存されたかどうか、プロデューサーが再試行するときにメッセージが重複して書き込まれないか、毎秒数万のメッセージを送信する際にブローカーを詰まらせないようにするかです。

この記事は3つの構成パラメータグループを介してこれら3つの質問に答えます：acks、enable.idempotence、およびバッチング（batch.size、linger.ms、compression.type）。この記事を読んだ後、安全で高速なプロデューサーを設定する方法、Kafkaクライアントだけを使用してもSpring Boot 3.xを使用してもわかります。また、テスト方法も知ることができます。

2. セットアップ

Javaだけを使用する場合、kafka-clientsが必要です：

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

Spring Boot 3.xを使用する場合、spring-kafkaを使用してください。バージョンはSpring Bootによって管理され、kafka-clientsが含まれています：

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

記事全体を通じて、注文イベントを送信します：

public record Order(String id, String customerId, BigDecimal total) {

    public String toJson() {
        return """
            {"id":"%s","customerId":"%s","total":%s}""".formatted(id, customerId, total);
    }
}

3. acks: メッセージがいつ送信されたと見なされるか？

acksは、プロデューサーが要求が成功したと見なされる前に、ブローカーからどのくらいの確認を待つかを決定します。

値	意味	リスク
acks=0	何も待たない	メッセージが失われ、まったく気づかない
acks=1	リーダーが書き込みを待つ	リーダーがフォロワーがコピーする前に故障するとメッセージが失われる
acks=all (-1)	ISR内のすべてのレプリカが書き込みを待つ	わずかに高いレイテンシ

クライアントだけを使用した設定例：

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

try (var producer = new KafkaProducer<String, String>(props)) {
    var order = new Order("o-1001", "c-42", new BigDecimal("250000"));
    RecordMetadata meta = producer
        .send(new ProducerRecord<>("orders", order.customerId(), order.toJson()))
        .get();
    System.out.printf("partition=%d offset=%d%n", meta.partition(), meta.offset());
}

.get()を呼び出すと送信が同期になります。ここではこれを行って結果を印刷するためだけです。acks=allの場合、.get()は、ISR内のすべてのレプリカが書き込みを終了した後にのみ返ります：

partition=2 offset=17

キーとしてcustomerIdを使用しているため、同じ顧客のすべての注文が同じパーティションに入り、正しい順序を保ちます。

3.1. min.insync.replicasがない場合、acks=allでは不十分です

acks=allは「現在ISR内のすべてのレプリカ」を意味します。ISRが縮小してリーダーだけになったと仮定します。この場合、acks=allはacks=1と変わりません。トピックまたはブローカーで追加の設定が必要です：

kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders \
  --partitions 6 --replication-factor 3 \
  --config min.insync.replicas=2

replication-factor=3とmin.insync.replicas=2の場合、各書き込みには最低2つのコピーが必要です。ISRが2未満に低下すると、プロデューサーは、単一のコピーのみを暗黙的に書き込む代わりに、NotEnoughReplicasExceptionを受け取ります。

注： min.insync.replicasをreplication-factorと等しく設定しないでください。その場合、1つのブローカーの再起動で、トピックがすべての書き込みを拒否します。

4. Idempotence: 重複をせずに再試行する

acks=allは再試行を伴います。次のシナリオを考えてみてください：ブローカーはメッセージを書き込みましたが、応答は途中で失われました。プロデューサーは送信が失敗したと考えて再度送信し、メッセージは2回書き込まれます。

冪等性プロデューサーはこのシナリオを正確に解決します。enable.idempotence=trueが有効な場合：

• ブローカーはプロデューサーに**プロデューサーID（PID）**を割り当てます。
• パーティションに送信される各バッチは、増加するシーケンス番号を持ちます。
• ブローカーは、既に表示されているシーケンスを持つバッチを削除し、シーケンスをスキップするバッチを拒否します。したがって、複数のリクエストが処理中にある場合でも、順序が保たれます。

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

各行の意味：

• ENABLE_IDEMPOTENCE_CONFIGはPIDとシーケンス番号のメカニズムを有効にします。
• 冪等性はacks=allを強制します。明示的にacks=1とenable.idempotence=trueを設定すると、プロデューサーは初期化時にConfigExceptionをスローします。
• max.in.flight.requests.per.connectionは≤ 5である必要があります。この制限内で、Kafkaはパーティション内の順序を保証します。
• retriesを非常に大きく設定します。実際の再試行時間はdelivery.timeout.ms（デフォルト2分）によって制限されます。

Kafka 3.0から、acks=allとenable.idempotence=trueがすでにデフォルトになっています。それでも、明示的に宣言する必要があります。そうしないと、誰かがacks=1を設定すると、冪等性は暗黙的に無効になります。

4.1. 冪等性ができないこと

これは誤解されることが多い部分です：

• これは単一のプロデューサーセッション内、各パーティションの重複のみを防ぎます。アプリケーションが再起動すると、プロデューサーは新しいPIDを受け取ります。再起動後に再度送信されたメッセージは、やはり重複して書き込まれます。
• これは、コードによってsend()が2回呼び出される場合の重複を防ぎません。たとえば、catchで自分で再試行する場合。
• 複数のパーティション間のアトミック書き込みを防ぎます。

上記のケースはトランザクション（transactional.id）またはコンシューマーがorderIdによって重複除外を行うことが必要です。

5. バッチング: スループットのためのレイテンシのトレードオフ

プロデューサーはメッセージを1つずつ送信しません。send()は、レコードをメモリ内のバッファに配置するだけで、各パーティションによって整理されます。バックグラウンドのスレッド（sender）はレコードをバッチにグループ化して送信します。3つの主なパラメータ：

パラメータ	デフォルト	役割
batch.size	16384バイト	パーティションの1つのバッチの最大サイズ
linger.ms	0（Kafka 4.0から5）	バッチを充填するのを待つ時間
compression.type	none	バッチ全体を圧縮：gzip、snappy、lz4、zstd
buffer.memory	32 MB	待機中のすべてのバッチの総メモリ

バッチは、batch.sizeが満杯または**linger.msが終了**いずれかの条件が最初に達成されたときに送信されます。linger.ms=0の場合、バッチはほぼ常にすぐに送信されるため、通常わずかなレコードしか含みません。

props.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

この設定は最大約10 msの追加のレイテンシを受け入れます。交換に、各要求は64KBの圧縮されたデータを持ちます。注文イベントのような反復的なJSONデータの場合、lz4またはzstdは通常、ネットワーク帯域幅とブローカー上のディスク容量を大幅に削減します。バッチあたりの圧縮は、個別のメッセージの圧縮よりもはるかに効果的です。バッチが大きいほど、利用する反復するシーケンスが増えるからです。

プロデューサーのメトリクスを介して確認することができます：

producer.metrics().forEach((name, metric) -> {
    if (name.name().equals("batch-size-avg") || name.name().equals("records-per-request-avg")) {
        System.out.println(name.name() + " = " + metric.metricValue());
    }
});

linger.ms=0で10,000の注文を連続して送信してから、linger.ms=10でもう一度繰り返すと、records-per-request-avgが明らかに増加します：

# linger.ms=0
batch-size-avg = 1830.4
records-per-request-avg = 21.7
# linger.ms=10
batch-size-avg = 41022.9
records-per-request-avg = 486.2

具体的な数値はマシンと負荷に依存しますが、傾向は常にそうです。

注： バッファが満杯になると（コンシューマーが遅い場合は関係ありません。ここではブローカーが遅いかネットワークが過負荷です）、send()はブロック最大max.block.ms（デフォルト60秒）してからTimeoutExceptionをスローします。ここでの可能性を計算せずに、HTTPリクエスト処理のスレッド内でsend()を呼び出さないでください。

注： delivery.timeout.msはlinger.ms + request.timeout.ms以上である必要があります。そうでない場合、プロデューサーは初期化を拒否します。

6. Spring Boot 3.xでの設定

Spring Bootは一般的なパラメータの独自のプロパティを持っています。他のパラメータはpropertiesを通じて渡されます：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      batch-size: 64KB
      compression-type: lz4
      buffer-memory: 64MB
      properties:
        enable.idempotence: true
        linger.ms: 10
        max.in.flight.requests.per.connection: 5
        delivery.timeout.ms: 120000

パブリッシャーはKafkaTemplateを使用します。Spring Kafka 3.0から、send()はCompletableFutureを返します：

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, String>> publish(Order order) {
        return kafkaTemplate.send("orders", order.customerId(), order.toJson())
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("注文{}の送信に失敗しました", order.id(), ex);
                } else {
                    var meta = result.getRecordMetadata();
                    log.info("注文{} -> パーティション{} オフセット{}",
                        order.id(), meta.partition(), meta.offset());
                }
            });
    }
}

3つのポイントに注意してください：

• コールバックはプロデューサーのsenderスレッドで実行されます。whenComplete内で負荷の高い作業をしたり、ブロッキングを呼び出したりしないでください。これにより、他のすべての送信が遅くなるためです。
• コールバックがエラーを受け取る場合、プロデューサーは**delivery.timeout.msの制限内で再試行を完了しています**。手動で再試行してsend()を再度呼び出さないでください。これは冪等性が防ぐことができない重複の源です。
• CompletableFutureを返すと、呼び出し元が待つかどうかを決定できます。

7. プロデューサーの動作をテストする

kafka-clientsはブローカーなしでテストするためのMockProducerを提供します。これを使用するには、ロジックをProducerインターフェイスを取得するクラスに抽象化します：

public class OrderSender {

    private final Producer<String, String> producer;

    public OrderSender(Producer<String, String> producer) {
        this.producer = producer;
    }

    public Future<RecordMetadata> send(Order order) {
        return producer.send(new ProducerRecord<>("orders", order.customerId(), order.toJson()));
    }

    public static Properties reliableConfig(String bootstrap) {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
        p.put(ProducerConfig.ACKS_CONFIG, "all");
        p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        p.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        p.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        p.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
        p.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        return p;
    }
}

JUnit 5とAssertJを使用してテストします：

class OrderSenderTest {

    @Test
    void whenSendingOrders_thenKeyIsCustomerId() {
        var mock = new MockProducer<>(true, new StringSerializer(), new StringSerializer());
        var sender = new OrderSender(mock);

        sender.send(new Order("o-1", "c-42", new BigDecimal("100")));
        sender.send(new Order("o-2", "c-42", new BigDecimal("200")));

        assertThat(mock.history())
            .hasSize(2)
            .extracting(ProducerRecord::key)
            .containsOnly("c-42");
    }

    @Test
    void whenBrokerFails_thenFutureCompletesExceptionally() {
        var mock = new MockProducer<>(false, new StringSerializer(), new StringSerializer());
        var sender = new OrderSender(mock);

        Future<RecordMetadata> future = sender.send(new Order("o-3", "c-7", BigDecimal.TEN));
        mock.errorNext(new NotEnoughReplicasException("ISR < min.insync.replicas"));

        assertThatThrownBy(future::get)
            .hasCauseInstanceOf(NotEnoughReplicasException.class);
    }

    @Test
    void reliableConfig_isAccepted() {
        var props = OrderSender.reliableConfig("localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        var config = new ProducerConfig(props);

        assertThat(config.getString(ProducerConfig.ACKS_CONFIG)).isEqualTo("-1");
        assertThat(config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)).isTrue();
    }
}

各テストは1つの動作をチェックします：

• 最初のテストはautoComplete=trueを使用するため、すべてのsend()は即座に成功します。history()はキーの確認を許可し、顧客による順序が保たれることを確認できます。
• 2番目のテストはautoComplete=falseを使用してからerrorNext(...)を使用してブローカーが書き込みを拒否することをシミュレートします。これは、エラー処理コードをテストする方法です。
• 最後のテストはProducerConfigを初期化して、Kafkaが構成の組み合わせを自動的に検証します。acks=1と冪等性を組み合わせるなど、無効な組み合わせは、デプロイ時に爆発する代わりにConfigExceptionでテストが失敗します。Kafkaはallを"-1"に正規化するため、アサーションは"-1"と比較します。

8. いつ緩和すべきか？

「最大安全」設定が常に正しいわけではありません：

• メトリック、ログ、クリックストリーム: 数個のレコードを失うことは許容されます。この場合、acks=1、高いlinger.ms、zstd圧縮により最良のスループットが得られます。
• 低レイテンシが必要なシステム（例えば、リアルタイム相場表示）：linger.msを小さく保つ（0～2 ms）。この場合、大きなバッチを期待しないでください。
• 支払い、注文: 常にacks=all、冪等性、min.insync.replicas=2を使用してください。読み取り－処理－書き込みが正確に1回になる必要がある場合は、トランザクションの使用を検討してください。

注： max.in.flight.requests.per.connection=1を「念のため」設定することは、冪等性が有効になっているときは不必要です。スループットを減らすだけです。

9. 結論

3つのパラメータグループは3つの異なる問題を解決します：

• acks=allとmin.insync.replicasを組み合わせることで、メッセージが失われないことを保証します。
• enable.idempotenceは再試行が重複しないことを保証し、パーティション内の順序を保ちます。その範囲は単一のプロデューサーセッションのみです。
• batch.size、linger.ms、compression.typeはスループットを決定し、わずかなレイテンシと引き換えに。

これらの値を明示的に宣言してください。デフォルトに依存しないでください。テストで検証して、デプロイ時の驚きを避けてください。プロデューサーがすでにそれを行っているため、コード内でsend()を自分で再試行しないでください。