Spring Kafka入門｜Producer・Consumerの完成コード

Spring Boot で Kafka を使うには、spring-boot-starter-kafka を入れ、送る側は KafkaTemplate、受け取る側は @KafkaListener を書きます。 接続先やシリアライザ（オブジェクトとバイト列の変換）は spring.kafka.* で設定し、Boot が KafkaTemplate とリスナーの入れ物（コンテナ）を自動で作ります。

@Service
public class OrderProducer {
    private final KafkaTemplate<String, OrderEvent> template;

    public OrderProducer(KafkaTemplate<String, OrderEvent> template) {
        this.template = template;
    }

    public void publish(OrderEvent event) {
        template.send("orders", event.customerId(), event);   // キーが同じなら同じパーティションへ
    }
}

@Component
public class OrderListener {
    @KafkaListener(topics = "orders", groupId = "order-service")
    public void onMessage(OrderEvent event) {
        // 処理。例外を投げると DefaultErrorHandler が再試行する
    }
}

つまずきやすい点は4つです。

確認環境：Spring Boot 4.1.1（Spring for Apache Kafka 4.1.1、Kafka クライアント 4.2.1 は Boot の依存管理の版）。spring.kafka.* の既定値は Boot v4.1.1 の KafkaProperties、Kafka クライアントの既定値は Apache Kafka 4.2.1 の ConsumerConfig・ProducerConfig、エラー処理・JSON・テストの既定値は Spring for Apache Kafka v4.1.1 のソースとリファレンスで確かめました。再試行の間隔は Spring Framework 7.0.9 の FixedBackOff・ExponentialBackOff と Spring Kafka の ExponentialBackOffWithMaxRetries をソースからビルドして、OpenJDK 21.0.10（Linux）で出力しました。キーとパーティションの対応は Kafka 4.2.1 の Utils.murmur2 と同じ計算で求め、Kafka 互換のブローカー Tansu 0.6.0 に kafka-python 3.0.11・confluent-kafka 2.16.0（librdkafka 2.16.0）で送って、クライアントが選んだパーティションとコンシューマーグループの割り当てを記録しました（確認日 2026年10月13日）。

Kafka の用語（最低限）

Kafka 本体の起動とトピックの作成は、関連記事「Kafka入門｜KRaftでトピック作成から消費まで最短手順」で扱っています。Kafka 4.0 から ZooKeeper は無くなり、KRaft だけで動きます。

依存（Boot 4）

// build.gradle.kts（Spring Boot 4.1.1）
dependencies {
    implementation("org.springframework.boot:spring-boot-starter-kafka")
    implementation("org.springframework.boot:spring-boot-starter-jackson")    // JSON で送るなら
    testImplementation("org.springframework.boot:spring-boot-starter-kafka-test")
}

Boot 3 までは org.springframework.kafka:spring-kafka を直接書いていましたが、Boot 4 では spring-boot-starter-kafka があります（v4.1.1 の starter/spring-boot-starter-kafka）。

application.yml（完成形）

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JacksonJsonSerializer
      acks: all
      properties:
        linger.ms: 10
    consumer:
      group-id: order-service
      auto-offset-reset: earliest
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      max-poll-records: 100
      properties:
        spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JacksonJsonDeserializer
        spring.json.trusted.packages: "com.example.order.event"
        spring.json.value.default.type: com.example.order.event.OrderEvent
    listener:
      ack-mode: record
      concurrency: 3

spring.kafka.* の設定と既定値

Boot v4.1.1 の KafkaProperties では、多くの項目が null（書かなければ Kafka クライアントの既定値）です。両方を1つの表にしました。

共通・Producer

properties.xxx は spring.kafka.producer.properties.xxx の略です。Boot に専用の項目が無い設定は、ここに Kafka の名前のまま書きます。

Consumer

enable.auto.commit は、Spring Kafka 4.1.1 の KafkaMessageListenerContainer が「設定に無ければ false にする」と決めています。コミットは Kafka に任せず、Spring のコンテナが下の ack-mode に従って行います。

Listener（コンテナ）

Admin・再試行トピック

Producer：KafkaTemplate

トピックを作る

@Configuration
public class KafkaTopicConfig {

    @Bean
    NewTopic orders() {
        return TopicBuilder.name("orders").partitions(3).replicas(1).build();
    }

    @Bean
    NewTopic ordersDlt() {
        // DLT は元のトピックと同じかそれ以上のパーティション数にする（下の「DLT」を参照）
        return TopicBuilder.name("orders-dlt").partitions(3).replicas(1).build();
    }
}

Boot の KafkaAdmin が、起動時に NewTopic の Bean を見て、無いトピックを作ります（spring.kafka.admin.auto-create、既定 true）。本番でトピックを別の手順で管理しているなら、auto-create: false にします。

送るコード（完成形）

package com.example.order;

import com.example.order.event.OrderEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

@Service
public class OrderProducer {

    private static final Logger log = LoggerFactory.getLogger(OrderProducer.class);
    private final KafkaTemplate<String, OrderEvent> template;

    public OrderProducer(KafkaTemplate<String, OrderEvent> template) {
        this.template = template;
    }

    /** 送りっぱなしにせず、結果をログに出す（非同期） */
    public CompletableFuture<SendResult<String, OrderEvent>> publish(OrderEvent event) {
        return template.send("orders", event.customerId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("送信失敗 orderId={}", event.orderId(), ex);
                    } else {
                        var md = result.getRecordMetadata();
                        log.info("送信 orderId={} partition={} offset={}", event.orderId(), md.partition(), md.offset());
                    }
                });
    }

    /** 送れたことを確かめてから戻る（同期） */
    public void publishAndWait(OrderEvent event) throws Exception {
        template.send("orders", event.customerId(), event).get(10, TimeUnit.SECONDS);
    }
}

package com.example.order.event;

import java.math.BigDecimal;
import java.time.Instant;

public record OrderEvent(String orderId, String customerId, BigDecimal amount, Instant orderedAt) {}

send は CompletableFuture<SendResult> を返します（Spring Kafka 3.0 で ListenableFuture から変わりました）。send を呼んだだけでは、まだ届いていません。 バッファにためて linger.ms（既定 5ms）待ってから、まとめて送ります。届いたかどうかは whenComplete か get で確かめます。

キーとパーティション

キーがあるとき、Kafka の Java クライアントは「キーの murmur2 ハッシュ ÷ パーティション数の余り」でパーティションを決めます（Kafka 4.2.1 の BuiltInPartitioner）。Utils.murmur2 と同じ計算で求めた結果です。

パーティションを増やすと、同じキーでも入る先が変わります。 customer-1 の増やす前のメッセージはパーティション 0、増やした後のメッセージは 3 に入るので、増やした直後はキーごとの順番が保証されません。パーティション数は最初に余裕を持って決めます。

同じキーを、別のクライアントから3パーティションのトピックに送って、選ばれたパーティションを記録しました。

librdkafka を使うクライアント（confluent-kafka、Confluent の Go・.NET のクライアントなど）は、既定のパーティションの決め方が Java と違います。 Java と Python など複数の言語から同じトピックに送り、キーごとの順番をそろえたいときは、librdkafka 側に partitioner=murmur2_random を設定します。

キーを null にすると、Java クライアントは一定量を送るごとにパーティションを切り替えながら送ります（Kafka 4.2.1 の BuiltInPartitioner）。順番が必要なメッセージには、注文 ID や顧客 ID などのキーを付けます。

Consumer：@KafkaListener

受け取るコード（完成形）

package com.example.order;

import com.example.order.event.OrderEvent;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class OrderListener {

    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);
    private final OrderService orderService;

    public OrderListener(OrderService orderService) {
        this.orderService = orderService;
    }

    // 値だけ受け取り、ヘッダーから場所を取る
    @KafkaListener(id = "order-listener", topics = "orders", groupId = "order-service", concurrency = "3")
    public void onOrder(@Payload OrderEvent event,
                        @Header(KafkaHeaders.RECEIVED_KEY) String key,
                        @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                        @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("受信 key={} partition={} offset={} orderId={}", key, partition, offset, event.orderId());
        orderService.accept(event);   // 例外を投げると DefaultErrorHandler へ
    }

    // ConsumerRecord ごと受け取る書き方
    @KafkaListener(topics = "orders-audit", groupId = "audit")
    public void onAudit(ConsumerRecord<String, String> record) {
        log.info("{}-{}@{} {}", record.topic(), record.partition(), record.offset(), record.value());
    }
}

concurrency とパーティション

concurrency = "3" は、コンシューマーを3つ動かすという意味です。1つのパーティションはグループの中の1つのコンシューマーにしか割り当てられないので、パーティション数より多くしても、余ったコンシューマーは何もしません。

3パーティションのトピックに、同じグループのコンシューマーを1つずつ増やしたときの割り当てです（kafka-python 3.0.11、割り当て方式 range）。

c4 は割り当てが無く、c2 が抜けたときに代わりに入りました。concurrency × アプリの台数が、パーティション数を超えないように決めます。

JSON で送る・受け取る（信頼パッケージ）

Spring Kafka 4 の JSON クラス

Spring Kafka 4.0 で、Jackson 3 用のクラスが加わりました。Boot 4 は Jackson 3 を使うので、こちらを使います。

信頼パッケージ（trusted packages）

JacksonJsonSerializer は、送るときにヘッダー __TypeId__ にクラス名（com.example.order.event.OrderEvent）を入れます。受け取る側の JacksonJsonDeserializer は、そのクラス名でオブジェクトを作ります。どんなクラスでも作れてしまうと危ないので、作ってよいパッケージ（信頼パッケージ）を決めています。 既定は java.util と java.lang だけです（v4.1.1 の DefaultJacksonJavaTypeMapper）。

設定していないと、受け取りのときに次の例外になります（文は v4.1.1 のソース）。

The class 'com.example.order.event.OrderEvent' is not in the trusted packages: [java.util, java.lang]. If you believe this class is safe to deserialize, please provide its name. If the serialization is only done by a trusted source, you can also enable trust all (*).

*（すべて信頼）は、送る側を自分たちが管理しているときだけにします。送る側と受け取る側でパッケージ名が違う（別のサービス）ときは、spring.json.type.mapping を両方に書くか、use.type.headers: false と value.default.type で受け取るクラスを固定します。

読めないメッセージで止まらないようにする

JSON が壊れたメッセージが1件あると、Kafka クライアントの poll が次の例外を投げ、同じメッセージを何度読んでも先に進めません（文は Kafka 4.2.1 の CompletedFetch）。

Error deserializing VALUE for partition orders-0 at offset 5. If needed, please seek past the record to continue consumption.

ErrorHandlingDeserializer で包むと、変換の失敗を DeserializationException としてリスナーの手前で受け取れます。上の application.yml の value-deserializer と spring.deserializer.value.delegate.class がその設定です。DeserializationException は再試行しても直らない例外として扱われ（下の表）、すぐに DLT へ送るかログに出して次へ進みます。

エラー処理：DefaultErrorHandler・再試行・DLT

既定の動き

リスナーで例外が出ると、Spring Kafka の DefaultErrorHandler が受け取ります。既定の再試行は SeekUtils.DEFAULT_BACK_OFF = FixedBackOff(0, 9) です。Spring Framework 7.0.9 の FixedBackOff をソースからビルドして、間隔を出力しました。

回数が尽きると、既定ではエラーのログを出してそのメッセージを捨て（コミットして）次へ進みます。ログの形は v4.1.1 の FailedRecordTracker で次のとおりです（orders-0@5 は「トピック-パーティション@オフセット」）。

Backoff FixedBackOffExecution[interval=0, currentAttempts=10, maxAttempts=9] exhausted for orders-0@5

再試行の間はコンシューマーがそのパーティションで止まります。 間隔を長くするほど、その間ほかのメッセージも待たされます。数秒より長く待ちたいときは、下の「再試行トピック」を使います。

再試行しない例外

次の例外は、既定で再試行せずにすぐ回復処理（ログ・DLT）へ回ります（v4.1.1 の ExceptionClassifier.defaultFatalExceptionsList）。

自分の例外を足すときは addNotRetryableExceptions(…) を使います。

DLT（デッドレタートピック）へ送る（完成形）

package com.example.order;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries;

@Configuration
public class KafkaErrorConfig {

    @Bean
    DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        // 失敗したメッセージを "<元のトピック>-dlt" の同じパーティションへ送る
        var recoverer = new DeadLetterPublishingRecoverer(template);

        var backOff = new ExponentialBackOffWithMaxRetries(4);   // 再試行4回
        backOff.setInitialInterval(1_000);
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(5_000);

        var handler = new DefaultErrorHandler(recoverer, backOff);
        handler.addNotRetryableExceptions(InvalidOrderException.class);   // 業務上の誤りは再試行しない
        return handler;
    }
}

Boot 4.1.1 の KafkaAnnotationDrivenConfiguration は、CommonErrorHandler の Bean が1つあれば、自動で作るリスナーのコンテナに設定します。Bean を定義するだけで効きます。

DLT のパーティション数が元より少ないと、同じ番号のパーティションが無いので送れません。 DLT は元と同じかそれ以上のパーティション数で作ります。Spring Kafka 3.3 で、名前の付け方が -dlt にそろえられました（それ以前の DeadLetterPublishingRecoverer は .DLT）。3.2 以前から上げたアプリは、送り先のトピック名を確かめます。

DLT のメッセージは、別のリスナーで受けて記録・調査します。

@KafkaListener(topics = "orders-dlt", groupId = "order-dlt")
public void onDlt(ConsumerRecord<String, byte[]> record,
                  @Header(KafkaHeaders.DLT_EXCEPTION_MESSAGE) String reason) {
    log.warn("DLT {}@{} reason={}", record.partition(), record.offset(), reason);
}

再試行トピック（@RetryableTopic）

DefaultErrorHandler の再試行はそのパーティションを止めます。止めずに、時間をおいて再試行したいときは @RetryableTopic を使います。失敗したメッセージを別の再試行用トピックに移し、時間が来たら読み直します。

@RetryableTopic(attempts = "4", backOff = @BackOff(delay = 1000, multiplier = 2.0, maxDelay = 10000))
@KafkaListener(topics = "payments", groupId = "payment-service")
public void onPayment(PaymentEvent event) {
    paymentService.charge(event);
}

@DltHandler
public void onPaymentDlt(PaymentEvent event, @Header(KafkaHeaders.EXCEPTION_MESSAGE) String reason) {
    log.error("決済の再試行が尽きた {} {}", event.paymentId(), reason);
}

再試行の考え方（再試行してよい処理・よくない処理）は、関連記事「Spring Retry・Resilience4jで再試行｜@Retryable設定」で扱っています。

手動コミット

ack-mode の一覧

コード（完成形）

spring:
  kafka:
    listener:
      ack-mode: manual_immediate

@KafkaListener(topics = "orders", groupId = "order-service")
public void onOrder(OrderEvent event, Acknowledgment ack) {
    try {
        orderService.accept(event);
        ack.acknowledge();                          // DB への保存が終わってからコミット
    } catch (TemporaryException e) {
        ack.nack(Duration.ofSeconds(5));             // コミットせず、5秒後にこのメッセージから読み直す
    }
}

Spring Kafka 4.1 からは、@KafkaListener(ackMode = "MANUAL_IMMEDIATE") のようにリスナーごとに変えられます。

Kafka の配信は「少なくとも1回」です。 処理のあとコミットの前にアプリが落ちると、同じメッセージがもう一度届きます。DB に保存する処理は、注文 ID を一意キーにするなど、2回届いても結果が変わらない（冪等な）作りにします。

テスト：EmbeddedKafka

package com.example.order;

import com.example.order.event.OrderEvent;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.bean.override.mockito.MockitoBean;

import java.math.BigDecimal;
import java.time.Instant;

import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.verify;
import static org.mockito.ArgumentMatchers.argThat;

@SpringBootTest(properties = "spring.kafka.consumer.auto-offset-reset=earliest")
@EmbeddedKafka(partitions = 3, topics = {"orders", "orders-dlt"})
class OrderFlowTest {

    @Autowired
    OrderProducer producer;

    @MockitoBean
    OrderService orderService;

    @Test
    void 送ったメッセージをリスナーが受け取る() throws Exception {
        var event = new OrderEvent("o-1", "customer-1", new BigDecimal("3980"), Instant.now());
        producer.publishAndWait(event);

        verify(orderService, timeout(10_000)).accept(argThat(e -> e.orderId().equals("o-1")));
    }
}

bootstrapServersProperty の既定が spring.kafka.bootstrap-servers なので、Boot のテストでは、起動したテスト用ブローカーの接続先が自動で入ります。 Spring Kafka 4.0 で ZooKeeper の部品（EmbeddedKafkaZKBroker・zookeeperPort など）と JUnit 4 の EmbeddedKafkaRule は削除され、KRaft の EmbeddedKafkaKraftBroker だけになりました。

テスト用のコンシューマーで直接読むときは KafkaTestUtils を使います。

Map<String, Object> props = KafkaTestUtils.consumerProps(broker, "test-group", false);
try (Consumer<String, String> c = new DefaultKafkaConsumerFactory<String, String>(props).createConsumer()) {
    broker.consumeFromAnEmbeddedTopic(c, "orders-audit");
    ConsumerRecord<String, String> r = KafkaTestUtils.getSingleRecord(c, "orders-audit");
}

コンテナ（本物の Kafka）でテストする方法は、関連記事「Testcontainersの使い方｜Spring BootでPostgreSQL」で扱っています（Kafka のコンテナも同じ考え方で使えます）。

よくあるエラーと対処

エラー文は Kafka 4.2.1（AbstractCoordinator・CommitFailedException・NetworkClient・KafkaProducer）と Spring Kafka 4.1.1 のソースにある文です（… は値が入る部分）。

よくある質問

Q. KafkaTemplate の型は <String, OrderEvent> と <Object, Object> のどちらで受け取りますか。 A. Boot 4.1.1 の KafkaAutoConfiguration が自動で作るのは、KafkaTemplate<?, ?> 型の Bean 1つです。ジェネリクスは使う側で KafkaTemplate<String, OrderEvent> と書いて受け取れます。実際に送れる型は、spring.kafka.producer.*-serializer の設定で決まります。

Q. 自分で ConsumerFactory の Bean を作ると、application.yml の設定はどうなりますか。 A. Boot の KafkaAutoConfiguration は @ConditionalOnMissingBean(ConsumerFactory.class) なので、自分で作ると Boot の ConsumerFactory は作られず、spring.kafka.consumer.* は自動では使われません。自分で作るときは、KafkaProperties#buildConsumerProperties() で設定を取り込みます。

Q. 1件ずつではなくまとめて受け取れますか。 A. spring.kafka.listener.type: batch か @KafkaListener(batch = "true") にすると、List<OrderEvent> で受け取れます。エラーのときは BatchListenerFailedException で失敗した位置を伝えると、その位置から再試行されます。

Q. メトリクスは取れますか。 A. Boot のリファレンスでは、自動で作る ConsumerFactory・ProducerFactory に Micrometer のリスナーが登録され、Kafka クライアントのメトリクスが /actuator/prometheus などに出ると説明されています。コンシューマーの遅れ（ラグ）は kafka_consumer_fetch_manager_records_lag_max などで見られます。

まとめ

• Boot 4 の依存は spring-boot-starter-kafka（Spring Kafka 4.1.1・Kafka クライアント 4.2.1）
• 送るのは KafkaTemplate.send(トピック, キー, 値)、受け取るのは @KafkaListener。send は CompletableFuture で結果を確かめる
• 同じキーは同じパーティション。パーティションを増やすと入る先が変わる。librdkafka は既定の決め方が Java と違う
• JSON は Boot 4 なら JacksonJsonSerializer・JacksonJsonDeserializer。受け取り側に spring.json.trusted.packages
• 読めないメッセージ対策に ErrorHandlingDeserializer
• 既定の再試行は 合計10回・待ちなしのあと捨てる。DefaultErrorHandler の Bean で回数・間隔・DLT（-dlt、同じパーティション番号）を決める
• コミットは Spring のコンテナが行う（enable.auto.commit は書かなければ false）。手動なら manual_immediate と acknowledge()
• テストは @EmbeddedKafka。接続先は spring.kafka.bootstrap-servers に自動で入る

参考・出典

確認日はいずれも 2026年10月13日です。

• Spring for Apache Kafka Reference（4.1）：https://docs.spring.io/spring-kafka/reference/
• Spring for Apache Kafka「What's new?」：https://docs.spring.io/spring-kafka/reference/whats-new.html
• Spring for Apache Kafka v4.1.1 ソース（DefaultErrorHandler・SeekUtils・DeadLetterPublishingRecoverer・ExceptionClassifier・DefaultJacksonJavaTypeMapper・ContainerProperties・EmbeddedKafka）：https://github.com/spring-projects/spring-kafka/tree/v4.1.1
• Spring Boot Reference「Apache Kafka Support」：https://docs.spring.io/spring-boot/reference/messaging/kafka.html
• Spring Boot v4.1.1 ソース（KafkaProperties・KafkaAutoConfiguration・KafkaAnnotationDrivenConfiguration）：https://github.com/spring-projects/spring-boot/tree/v4.1.1
• Apache Kafka 4.2.1 ソース（ConsumerConfig・ProducerConfig・Utils.murmur2・BuiltInPartitioner）：https://github.com/apache/kafka/tree/4.2.1
• Apache Kafka Documentation「Configuration」：https://kafka.apache.org/documentation/#configuration
• librdkafka「CONFIGURATION.md」（partitioner）：https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。