Kafka Java APIの実装設計｜Producerのacks・べき等性とConsumerのコミット方式を実装視点で解説

KafkaのJavaクライアントは、依存を1つ足して20行ほど書けば動きます。つまずくのはその先で、「送ったはずのメッセージが消えた」「同じ注文が二重に処理された」「消費側が数分ごとに止まる」といった症状が本番で出てから設定の意味を調べ直すことになりがちです。この記事では、コードの写経ではなく既定値が何を保証していて、どの設定を触ると保証が崩れるかという順序でJavaクライアントの実装を組み立てます。Apache Kafka本体の構成要素と役割は別記事で扱っているため、ここではアプリケーション側のコードに絞りました。

本文で示す設定値と挙動は、2026年8月31日にApache Kafka公式ドキュメントの4.3系設定一覧と公式Javadocで実測したものです。同日時点でサポート対象のリリースは 4.3.1（2026年6月25日公開）・4.2.1（2026年5月30日公開）・4.1.2（2026年3月17日公開）の3系統で、Maven Centralの kafka-clients も最新が 4.3.1 でした。

まとめ｜Kafka Java APIで先に決める送達保証とコミット境界

先に結論を置きます。4.x系のJavaクライアントは、送信側の既定が安全側、消費側の既定が危険側という非対称です。プロデューサーは acks=all・enable.idempotence=true が既定で、何も設定しなければ重複も欠落も起きにくい状態から始まります。一方コンシューマーは enable.auto.commit=true が既定で、処理が終わる前にオフセットが進む余地を持ったまま動きます。

したがって実装で最初に決めるのは2点です。プロデューサー側は既定を崩す変更をするかどうか、コンシューマー側はコミットをどこで打つか。前者は「スループットのために保証を落とす」判断であり、後者は「欠落を許すか重複を許すか」の判断です。両方とも設定1行で挙動が変わるため、レビューで見落とすと本番で初めて表面化します。

例外処理は、クライアントが自動で再試行する範囲とアプリケーションが持つ範囲を分ける設計です。プロデューサーは delivery.timeout.ms（既定120000ミリ秒）まで自力で粘るため、その上に独自のリトライを重ねると多重管理になります。以下、依存の置き方から順に追っていきます。

Kafka Java APIの依存はkafka-clients単体で足りるかを先に決める

KafkaのJavaクライアントは org.apache.kafka:kafka-clients という単一の依存に、プロデューサー・コンシューマー・管理用のAdminClientがまとめて入っています。ストリーム処理のDSLを使う場合だけ kafka-streams が別途必要になります。2026年8月31日時点でMaven Centralの最新版は 4.3.1 でした。

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

クライアントとブローカーの版は完全一致させる必要がありません。Kafkaはバイナリプロトコルの版を接続時に折衝する仕組みを持っているため、新しいクライアントで古いブローカーへ接続する構成も動きます。ただし新版クライアントでしか使えない設定を古いブローカーに投げると実行時に弾かれるため、クライアントを先に上げてブローカーを後から上げる順序にしておくと切り戻しが楽になります。

KafkaProducerとKafkaConsumerではスレッドの扱いが逆になる

実装の骨格に効く違いはひとつです。4.3系のJavadocでは、KafkaProducer について「The producer is thread safe and sharing a single producer instance across threads will generally be faster than having multiple instances」と明記されています。つまりプロデューサーはアプリケーション全体で1インスタンスを共有するのが推奨で、リクエストごとに生成するとバッチが効かず接続だけが増えます。

逆に KafkaConsumer は「The consumer is not thread-safe」との記載です。1スレッドにつき1インスタンスが原則で、複数スレッドから同じインスタンスの poll() を呼ぶと例外になります。並列度を上げたい場合は、コンシューマーを複数プロセス・複数スレッドに分けてパーティション数まで増やすか、poll したレコードだけをワーカースレッドへ渡す構成にします。

この2つを取り違えた実装は、負荷をかけるまで表面化しません。プロデューサーをリクエストごとに作る実装は流量が小さいうちは動いてしまい、コンシューマーを共有する実装は単一スレッドで動かしている間は問題が出ないためです。

Producerの既定値は4.x系で安全側に寄っている前提から入る

プロデューサーの設定は数十ありますが、送達保証に効くのは次の値です。いずれも2026年8月31日時点の4.3系ドキュメントで確認した既定値です。

acks=all は、リーダーがISR（同期済みレプリカ）全体への複製を確認してから応答を返す設定です。enable.idempotence=true はプロデューサーIDとシーケンス番号でブローカー側が重複を検出する仕組みで、再送による二重書き込みを防ぎます。この2つが既定で入っているため、4.x系では何も書かなければ「欠落しにくく重複もしにくい」状態から始まります。

べき等性を有効にしたまま変更できる設定と保証が壊れる設定の境界

べき等性の前提条件は次のとおりです。公式ドキュメントの記述どおり、max.in.flight.requests.per.connection が5以下、retries が0より大きい、acks が all であることが揃って初めて有効です。逆に言えば、スループットを上げようとして acks=1 に落とす、あるいは並列送信を増やそうとして in-flight を6以上にすると、べき等性ごと外れます。競合する設定を明示した場合は起動時に設定エラーになるため、気付かずに保証だけ失う事故は起きにくくなっていますが、既定値の意味を知らずに古い記事の設定例を持ち込むと保証を落とす方向に働きます。

触ってよいのはバッチ関連です。linger.ms は既定5ミリ秒で、これを20〜50ミリ秒程度まで伸ばすと1リクエストあたりのレコード数が増え、圧縮も効きやすくなります。compression.type は既定 none なので、ネットワーク帯域が先に詰まる構成では lz4 や zstd を明示的に指定する価値があります。どちらも送達保証には影響しません。

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.LINGER_MS_CONFIG, 20);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

try (Producer<String, String> producer = new KafkaProducer<>(props)) {
    producer.send(new ProducerRecord<>("orders", orderId, payload), (metadata, ex) -> {
        if (ex != null) {
            log.error("send failed: topic={} key={}", "orders", orderId, ex);
        }
    });
}

send()は非同期なので戻り値の扱いでスループットが決まる

send() はレコードをバッファへ積んで即座に戻る非同期動作です。返り値の Future に対して get() を呼ぶと1件ごとに応答を待つ同期送信になり、バッチが効かなくなるためスループットが桁で落ちます。それでも同期にしたい場面は、送信の成否をその場の業務トランザクションの成否と揃えたい場合に限られます。

非同期のままにする場合、失敗はコールバックでしか受け取れません。コールバックの中で例外を握り潰す実装が最も危険で、送信失敗がログにも残らず「なぜかデータが欠けている」状態になります。上のコード例のように、失敗時はキーを含めて記録し、後から追跡できる形にしておきます。

もうひとつ落とし穴があります。プロセス終了時に close() か flush() を通らないと、バッファに残ったレコードは送信されずに消えます。try-with-resources で囲むか、シャットダウンフックで明示的に閉じる実装にしてください。

Consumerはコミット位置の設計が実装の成否そのものを決める

消費側の既定値は、送信側と違って「そのままでよい」とは言えません。4.3系で確認した主要な既定値は次のとおりです。

enable.auto.commit=true のままだと、オフセットは poll() の呼び出しに合わせて定期的に前進する仕組みです。前回のpollで受け取ったレコードを処理し終える前にプロセスが落ちれば、そのレコードは処理されないまま消費済み扱いになります。欠落を許せる用途なら問題になりませんが、注文や決済のように取りこぼしが業務事故になる処理では、まずこの設定を false にするところから始めます。

commitSyncとcommitAsyncは定常時と終了時で使い分ける

手動コミットには2つの選択肢があります。commitSync() はブローカーの応答を待ち、失敗すれば例外を投げます。commitAsync() は待たずに次のpollへ進めるためスループットは高いものの、失敗しても自動では再試行しません。

ここで commitAsync() の失敗を単純にリトライする実装は避けます。後続のコミットが先に成功していた場合、古いオフセットで上書きしてすでに処理済みのレコードを再び読み直すためです。実務でよく使われるのは、定常時は commitAsync() で回し、シャットダウンやリバランス直前だけ commitSync() で確実に打つ組み合わせです。

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);

try (Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(List.of("orders"));
    while (running) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            handle(record);
        }
        consumer.commitAsync();
    }
    consumer.commitSync();
}

この形は少なくとも1回の配信（at-least-once）方式です。処理が終わってからコミットするため欠落はしませんが、コミット直前に落ちれば同じレコードを再処理します。つまり消費側の処理を冪等に作ることが前提条件です。注文IDを一意キーにして挿入時に衝突を無視する、といった設計はデータパイプラインの冪等性から考える設計判断で扱った考え方がそのまま使えます。

pollの取得件数と処理間隔を誤るとリバランスで消費が止まる条件

コンシューマーは max.poll.interval.ms（既定300000ミリ秒＝5分）以内に次の poll() を呼ばないと、グループから離脱したとみなされてリバランスが発生します。1件あたりの処理に外部API呼び出しやDB更新が入る構成では、既定の max.poll.records=500 が重すぎることがあります。500件×0.6秒で5分を超えるためです。

症状は分かりやすく、一定周期で同じレコードが再処理され、消費が前に進まないという形です。対処は件数を下げる方向が第一手で、上のコード例のように100件程度へ落とすと余裕ができます。処理そのものが重い場合は、pollしたレコードをワーカースレッドへ渡し、poll ループ自体は止めない構成へ切り替えます。

auto.offset.reset=latest という既定も検証時の混乱を招く要因です。新しいコンシューマーグループで起動すると過去のレコードを読まないため、「送っているのに何も届かない」と見える状態です。過去分から読みたい場合は earliest を明示します。実際のラグがどう動いているかはKafka監視の指標設計をまとめた記事の考え方で数値として確認できます。

例外はクライアントが再試行する範囲とアプリ側の切り分けで決める

Javaクライアントの例外は、再試行して直るものと直らないものにはっきり分かれます。RetriableException を継承する例外（リーダー切り替え中や一時的なタイムアウト）は、プロデューサーが delivery.timeout.ms の範囲内で自動的に再送します。既定は120000ミリ秒なので、2分間はクライアントが自力で粘っていると考えてよいわけです。

この上にアプリケーション側の再試行を重ねると、実際の待ち時間が想定の何倍にもなります。呼び出し元にタイムアウトがある同期処理では、クライアントがまだ粘っている最中に上位が諦め、後から送信が成功して二重登録になることもあります。プロデューサーの送信に自前のリトライを足すなら、delivery.timeout.ms を業務側の許容時間まで縮めてからにするのが筋の通った順序です。再試行の間隔設計そのものはリトライパターンの指数バックオフとリトライ予算で整理しています。

修正で回復しない例外は再送せず隔離に回す判断基準を先に決める

一方、RecordTooLargeException（メッセージがブローカーの上限を超えた）、SerializationException（シリアライズに失敗した）、認可エラーは、何度送っても結果が変わりません。これらを再送ループに載せると、同じレコードで延々と失敗し続けます。送信前の検証で弾くか、送れなかったレコードを別の場所へ退避して業務を先に進めます。

消費側でも同じ切り分けが必要です。特定のレコードで処理が失敗し続けると、そのパーティションのオフセットが進まず、後続のレコードがすべて滞留します。順序保証と引き換えに詰まるという構造であり、「1件の不正データで全体を止める」か「不正データを別トピックへ逃がして流し続ける」かは業務要件で決めるほかありません。決済のように順序が意味を持つ処理なら止める側、ログ集約のように欠けても後追いできる処理なら逃がす側が妥当です。逃がす場合は、失敗レコードを退避する専用トピックを用意し、そこに溜まった件数を監視項目に入れておきます。

コンシューマーグループの新プロトコルは移行判断を分けて考える

4.x系では、コンシューマーグループの調整方式に2つの選択肢があります。group.protocol の有効値は classic と consumer で、2026年8月31日時点の4.3系では既定は classic のままです。新しい consumer プロトコルは、パーティション割り当てをブローカー側の調整役が担当し、リバランス時に全員が一斉に止まる挙動を避ける設計になっています。

移行の判断材料は2つです。第一に、ブローカー側が対応した構成であること。第二に、group.remote.assignor のようなサーバー側の割り当て設定は group.protocol=consumer のときだけ効くため、クライアント設定だけを書き換えても意味を持たない点です。既定が classic である以上、リバランス時の停止が実測で問題になっている構成に限って移行を検討するのが妥当で、新規構築だからという理由で先に切り替える必要はありません。ブローカー側の世代についてはKRaft構成への移行手順をまとめた記事を合わせて確認してください。

Java APIを直に書く条件とConnectへ寄せる条件を分ける

ここまでの設定と例外処理を自前で持つ判断は、常に正解ではありません。同じ「Kafkaへ入れる・Kafkaから出す」でも、コードを書かない選択肢があるためです。

コミット境界を業務処理と揃える必要がある場合はJava APIを直に書く

Javaクライアントを直に使う価値が出るのは、オフセットのコミット位置を業務処理の完了と厳密に揃えたい場合です。DBへの書き込みが成功した後にだけコミットする、複数レコードをまとめて1トランザクションで処理する、といった制御は、コミットのタイミングを自分で握っていなければ組めません。既存のJavaアプリケーションへ組み込む場合や、外部依存を増やせない事情がある場合も直接実装が向きます。

この判断は、そもそもKafkaを選ぶかどうかの判断と地続きです。個別メッセージの確実な配送と再試行を主眼に置くなら別の選択肢もあり、その比較はRabbitMQとKafkaの設計思想の違いで整理しています。

外部システムとの入出力はコネクタへ寄せると実装と運用の総量が減る

逆に、DBやオブジェクトストレージとKafkaの間でデータを流すだけなら、Javaクライアントを書く理由はほぼありません。既存のコネクタを設定で動かすほうが、オフセット管理も再起動時の再開も枠組み側が持ってくれるためです。この領域はKafka Connectの構成要素とコネクタ運用の担当範囲になります。集計や結合を伴う変換処理も、生のクライアントで状態を持つより kafka-streams に寄せたほうが実装量が減ります。

判断の順序としては、コネクタで足りるかを先に確認し、足りない部分だけをJavaクライアントで書く形が無駄になりません。ストリーム基盤を含めたデータ処理の全体設計から相談したい場合は、データ分析基盤構築・MLOps構築支援で構成の検討からご依頼いただけます。

よくある質問

kafka-clientsのバージョンはブローカーと合わせる必要がありますか？

厳密な一致は不要です。接続時にプロトコル版を折衝する仕組みがあるため、クライアントが新しくブローカーが古い組み合わせでも動作します。ただし新しい設定値を古いブローカーへ投げると実行時に弾かれるので、動作確認は実際に接続する版の組み合わせで行ってください。

acks=1にすればスループットは上がりますか？

応答待ちは短くなりますが、2026年8月31日時点の4.3系では acks の既定が all であり、これを 1 に落とすとべき等性の前提条件から外れます。耐久性と重複排除の両方を手放す変更になるため、スループットが実測で不足していることを確認したうえで、圧縮やバッチ設定を先に試す順序をおすすめします。

自動コミットのままでも問題ない場面はありますか？

欠落しても業務影響がない用途、たとえば統計目的のログ集約や、後から全件を再集計できる処理なら実用になります。逆に注文・決済・在庫のように取りこぼしが事故になる処理では、手動コミットへ切り替えて処理完了後にコミットする形にしてください。

1つのコンシューマーを複数スレッドで共有できますか？

できません。4.3系のJavadocに「The consumer is not thread-safe」と明記されており、複数スレッドから同じインスタンスを操作すると例外になります。並列度を上げる場合は、スレッドごとにインスタンスを分けるか、pollしたレコードだけをワーカーへ渡してループ本体は単一スレッドに保つ構成にします。

Exactly-onceはJavaクライアントだけで実現できますか？

Kafka内部で完結する処理なら、トランザクションAPIと消費側の isolation.level=read_committed の組み合わせで実現できます。既定は read_uncommitted なので明示的な指定が必要です。ただし外部DBへの書き込みを含む処理は、Kafkaのトランザクション範囲外になるため、消費側の処理を冪等に作る方針のほうが現実的です。

関連記事

• Apache Kafkaが選ばれる理由とは？特徴とメリットを徹底解説：クライアントが接続する先のブローカーとトピックの構造を確認できます。
• Kafka Connectとは？構成要素とコネクタ運用を実装視点で解説【2026年版】：コードを書かずに入出力を組む選択肢を比較できます。
• Kafka監視の設計｜ラグ・ISR・ディスクの指標選定とJMX取得経路を実装視点で解説：コンシューマーラグを数値として追う設計を確認できます。
• Kafka KRaftとは？ZooKeeper廃止後の構成と移行手順を実装視点で解説：接続先クラスタ側の世代と移行の前提を押さえられます。
• リトライパターンとは？指数バックオフ・ジッター・リトライ予算の実装判断を解説：再送間隔と諦める条件の決め方を確認できます。
• Confluent Kafkaとは？PlatformとCloudの構成差・ライセンス境界・費用分担を実装視点で解説：商用ディストリビューションを使う場合の前提を確認できます。