Spring Boot: Apache Kafka

This content was generated with the assistance of AI and should be verified against the official documentation before being relied on in production; see the section bibliography for the reference material consulted while preparing these pages.

This page assumes the baseline covered on Messaging with Apache Kafka — spring.kafka. connection properties, the auto-configured KafkaTemplate and transactional producers, @KafkaListener with manual/batch acknowledgment, DefaultErrorHandler with blocking retry and a DeadLetterPublishingRecoverer, and the auto-configured ConsumerFactory/ProducerFactory with JsonSerializer/JsonDeserializer — and does not repeat any of it. What follows is everything else Spring for Apache Kafka 4.1 adds on top of that baseline: declarative topic provisioning; request/reply and multi-cluster templates; per-listener acknowledgment mode; *non-blocking retry (as opposed to the blocking DefaultErrorHandler retry already covered); rebalance listeners and manual seeking; pausing and resuming containers; the Spring wrapper around Kafka’s share consumers (KIP-932); KafkaTransactionManager-driven exactly-once; Kafka Streams via @EnableKafkaStreams; the Jackson-based serializers that replace JsonSerializer/JsonDeserializer; and opting a listener container into the new KIP-848 consumer group protocol. Examples target Spring Boot 4.1.x, Spring for Apache Kafka 4.1.x, and Apache Kafka 4.3.x, running on Java 25 (Java 21 is the current LTS both target versions still support). For the protocol-level mechanics behind several of these features — share groups, transactions, and Kafka Streams topologies — see Kafka Share Groups, Kafka Reliability and Exactly-Once Semantics, and Kafka Streams respectively; this page only covers the Spring API that wraps them.

Topic provisioning

An application that needs its topics to exist before the first producer send or consumer poll declares them as NewTopic beans; Spring Boot auto-configures a KafkaAdmin bean from spring.kafka.bootstrap-servers that finds every NewTopic bean in the context at startup and creates (or reconciles the partition/replica count of) whichever ones do not already match:

@Configuration
public class OrderTopicsConfig {

    @Bean
    public NewTopic ordersPlaced() {
        return TopicBuilder.name("orders.placed")
                .partitions(6)
                .replicas(3)
                .config(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(Duration.ofDays(7).toMillis()))
                .build();
    }

    @Bean
    public NewTopic ordersPlacedDlt() {
        // .compact() switches the topic to log compaction instead of the default delete cleanup policy
        return TopicBuilder.name("orders.placed.DLT")
                .partitions(6)
                .replicas(3)
                .build();
    }
}

TopicBuilder is a thin fluent front end over NewTopic; declaring several topics on one bean is possible with KafkaAdmin.NewTopics, and KafkaAdmin itself exposes createOrModifyTopics(NewTopic…​), describeTopics(String…​), and deleteTopics(String…​) for provisioning done imperatively at runtime rather than declaratively at startup. KafkaAdmin here plays the same startup-provisioning role Spring Boot’s own KafkaAdmin auto-configuration already plays for the connection itself — no extra property is needed beyond having NewTopic beans in the context. See Spring for Apache Kafka reference — Configuring Topics.

Request/reply and multi-destination templates

ReplyingKafkaTemplate

KafkaTemplate (covered on the baseline page) is fire-and-forget: send() resolves once the broker acknowledges the write, not once anything downstream has acted on it. ReplyingKafkaTemplate builds a synchronous request/reply exchange on top of that same asynchronous transport, correlating each request with its reply via a KafkaHeaders.CORRELATION_ID header:

@Configuration
public class OrderPricingRequestReplyConfig {

    @Bean
    public ReplyingKafkaTemplate<String, PriceRequest, PriceReply> replyingTemplate(
            ProducerFactory<String, PriceRequest> producerFactory,
            ConcurrentKafkaListenerContainerFactory<String, PriceReply> containerFactory) {

        ConcurrentMessageListenerContainer<String, PriceReply> repliesContainer =
                containerFactory.createContainer("pricing.replies");
        repliesContainer.getContainerProperties().setGroupId("pricing-replies");
        // the template itself starts/stops this container -- it must not auto-start on its own
        repliesContainer.setAutoStartup(false);
        return new ReplyingKafkaTemplate<>(producerFactory, repliesContainer);
    }
}
@Service
public class OrderPricingClient {

    private final ReplyingKafkaTemplate<String, PriceRequest, PriceReply> replyingTemplate;

    public OrderPricingClient(ReplyingKafkaTemplate<String, PriceRequest, PriceReply> replyingTemplate) {
        this.replyingTemplate = replyingTemplate;
    }

    public PriceReply priceOrder(String orderId, PriceRequest request) throws Exception {
        ProducerRecord<String, PriceRequest> record = new ProducerRecord<>("pricing.requests", orderId, request);
        RequestReplyFuture<String, PriceRequest, PriceReply> future =
                replyingTemplate.sendAndReceive(record, Duration.ofSeconds(5));
        future.getSendFuture().get(5, TimeUnit.SECONDS); // fail fast if the request itself never sent
        return future.get(5, TimeUnit.SECONDS).value();
    }
}

The pricing side replies by returning a value from a @KafkaListener annotated @SendTo, which Spring routes back to the header-provided reply topic and correlation ID automatically:

@Component
public class OrderPricingServer {

    @KafkaListener(id = "pricing-server", topics = "pricing.requests")
    @SendTo
    public PriceReply price(PriceRequest request) {
        return pricingEngine.price(request);
    }
}

RoutingKafkaTemplate

A single application that must publish to topics served by different clusters, or that need different value serializers depending on the destination, routes through one RoutingKafkaTemplate instead of injecting several KafkaTemplate beans and choosing between them by hand. It dispatches each send() to the first matching ProducerFactory in an ordered map of topic-name regular expressions:

@Bean
public RoutingKafkaTemplate routingTemplate(
        GenericApplicationContext context,
        ProducerFactory<Object, Object> jsonProducerFactory) {

    Map<String, Object> rawConfigs = new HashMap<>(jsonProducerFactory.getConfigurationProperties());
    rawConfigs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
    DefaultKafkaProducerFactory<Object, Object> rawProducerFactory =
            new DefaultKafkaProducerFactory<>(rawConfigs);
    context.registerBean("rawEventsProducerFactory", DefaultKafkaProducerFactory.class, () -> rawProducerFactory);

    // ordered map: more specific patterns first, a catch-all default last
    Map<Pattern, ProducerFactory<Object, Object>> routes = new LinkedHashMap<>();
    routes.put(Pattern.compile("raw\\..+"), rawProducerFactory);
    routes.put(Pattern.compile(".+"), jsonProducerFactory);
    return new RoutingKafkaTemplate(routes);
}

A RoutingKafkaTemplate does not support transactions, execute(), flush(), or metrics(), because those operations need a single, known producer/topic at configuration time, which a router by definition does not have until the first send(). See Spring for Apache Kafka reference — Sending Messages.

Per-listener acknowledgment mode

The baseline page sets acknowledgment mode globally, via spring.kafka.listener.ack-mode on the container factory. Since Spring for Apache Kafka 4.1, the ackMode attribute on @KafkaListener itself overrides the container factory’s default for that one listener, so a single factory can back listeners with different acknowledgment needs instead of one factory bean per AckMode:

@Component
public class OrderEventListeners {

    // container factory default ack mode applies here (e.g. BATCH, from application.yml)
    @KafkaListener(id = "orders-placed", topics = "orders.placed")
    public void onOrderPlaced(OrderPlacedEvent event) {
        orderProcessor.process(event);
    }

    // this one listener needs to control its own commit point, without a dedicated factory bean
    @KafkaListener(id = "orders-refunded", topics = "orders.refunded", ackMode = "MANUAL")
    public void onOrderRefunded(OrderRefundedEvent event, Acknowledgment acknowledgment) {
        refundProcessor.process(event);
        acknowledgment.acknowledge();
    }
}

ackMode accepts the same names as ContainerProperties.AckMode (RECORD, BATCH, MANUAL, MANUAL_IMMEDIATE, and so on) and also accepts a ${…​} placeholder or #{…​} SpEL expression. See Spring for Apache Kafka reference — @KafkaListener Annotation.

Non-blocking retry with @RetryableTopic

The baseline page’s DefaultErrorHandler retries in place, blocking that partition’s poll loop between attempts — fine for a fast, transient failure, but it stalls every other record behind the failing one for the whole backoff period. @RetryableTopic instead retries by re-publishing the failed record to a dedicated retry topic with its own consumer, so the original partition keeps moving while the retry waits in the background:

@Component
public class OrderEventListener {

    @RetryableTopic(
            attempts = "4",
            backoff = @Backoff(delay = 1000, multiplier = 2.0, maxDelay = 15000),
            autoCreateTopics = "true",
            exclude = IllegalArgumentException.class)
    @KafkaListener(topics = "orders.placed", groupId = "orders-service")
    public void onOrderPlaced(OrderPlacedEvent event) {
        orderProcessor.process(event);
    }

    @DltHandler
    public void onOrderPlacedDlt(OrderPlacedEvent event, @Header(KafkaHeaders.EXCEPTION_MESSAGE) String reason) {
        log.error("order {} exhausted retries, sent to DLT: {}", event.orderId(), reason);
    }
}

With the default topicSuffixingStrategy, each retry attempt gets its own physically separate topic — orders.placed-retry-0, orders.placed-retry-1, and so on — and a final failure lands on orders.placed.DLT, the same .DLT suffix convention DeadLetterPublishingRecoverer uses on the baseline page. exclude (and its counterpart include) keep exceptions that are never worth retrying — typically deserialization or validation failures — from being retried at all, mirroring addNotRetryableExceptions on DefaultErrorHandler. RetryTopicConfigurationBuilder is the equivalent programmatic, @Bean-based configuration for cases — such as applying the same retry policy across many listeners — where annotating every method individually would be repetitive. Non-blocking retry is not compatible with batch listeners or with container-managed transactions. See Spring for Apache Kafka reference — Non-Blocking Retries.

Rebalance listeners and seeking

ConsumerAwareRebalanceListener

A partition rebalance — a new consumer joining the group, an existing one leaving, a @RetryableTopic topic being provisioned — is usually invisible to application code. When it must not be, ConsumerAwareRebalanceListener hooks the moments before/after Spring commits any pending offsets on revocation and the moment new partitions are assigned:

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderPlacedEvent> ordersContainerFactory(
        ConsumerFactory<String, OrderPlacedEvent> consumerFactory) {

    ConcurrentKafkaListenerContainerFactory<String, OrderPlacedEvent> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.getContainerProperties().setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {

        @Override
        public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
            partitions.forEach(tp -> offsetTracker.store(tp, consumer.position(tp)));
        }

        @Override
        public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
            partitions.forEach(tp -> offsetTracker.restore(tp).ifPresent(offset -> consumer.seek(tp, offset)));
        }
    });
    return factory;
}

onPartitionsRevokedBeforeCommit runs before Spring commits pending offsets (the place to flush any pending Acknowledgment), onPartitionsRevokedAfterCommit runs once that commit has happened (the place to persist state to an external store), and onPartitionsLost (since Spring for Apache Kafka 2.4) fires instead of the revoked callbacks when the broker considers the assignment already gone. See Spring for Apache Kafka reference — Rebalance Listeners.

ConsumerSeekAware

A listener bean that needs to control its own starting offset — replaying from the beginning on a fresh deployment, or seeking to a timestamp — implements ConsumerSeekAware and drives the ConsumerSeekCallback it is handed:

@Component
public class OrderAuditListener extends AbstractConsumerSeekAware {

    @KafkaListener(id = "orders-audit", topics = "orders.placed")
    public void onOrderPlaced(OrderPlacedEvent event) {
        auditLog.record(event);
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        // replay the last hour on every (re)assignment, instead of resuming from the committed offset
        assignments.keySet().forEach(tp ->
                callback.seekToTimestamp(tp.topic(), tp.partition(),
                        System.currentTimeMillis() - Duration.ofHours(1).toMillis()));
    }
}

AbstractConsumerSeekAware (extended above) tracks the per-partition callback bookkeeping automatically; a class implementing the raw ConsumerSeekAware interface directly must store the ConsumerSeekCallback itself (typically per-thread) via registerSeekCallback and clean it up in onPartitionsRevoked. ConsumerSeekCallback also exposes seek(topic, partition, offset), seekToBeginning/seekToEnd, and seekRelative for offsets relative to the current position. See Spring for Apache Kafka reference — Seeking to a Specific Offset.

Pausing and resuming containers

A container backed by KafkaListenerEndpointRegistry can be paused and resumed by its listener ID, for example to shed load while a downstream dependency is degraded:

@Service
public class DownstreamHealthGate {

    private final KafkaListenerEndpointRegistry registry;

    public DownstreamHealthGate(KafkaListenerEndpointRegistry registry) {
        this.registry = registry;
    }

    public void onDownstreamUnhealthy() {
        registry.getListenerContainer("orders-placed").pause();
    }

    public void onDownstreamHealthy() {
        registry.getListenerContainer("orders-placed").resume();
    }
}

pause() takes effect just before the container’s next poll() — it keeps polling with an empty subscription to avoid triggering a rebalance, it simply stops returning records — and resume() takes effect right after the poll that observes it; isPauseRequested()/isConsumerPaused() distinguish "asked to pause" from "actually paused." ConsumerPausedEvent/ConsumerResumedEvent are published so other beans can react to the transition via @EventListener instead of polling the container’s own state. See Spring for Apache Kafka reference — Pausing and Resuming Listener Containers.

Share consumers: the Spring wrapper around Kafka queues

Kafka Share Groups covers the protocol underneath this section in full — record-level acquisition locks, the ACCEPT/RELEASE/REJECT/RENEW outcomes, and delivery-count poison-message handling. Spring for Apache Kafka 4.1 (backed by Kafka 4.2+ share-group support) wraps that protocol behind the same @KafkaListener programming model already used for consumer groups, via a parallel set of share-consumer-specific factories.

ShareConsumerFactory/DefaultShareConsumerFactory play the same role ConsumerFactory plays on the baseline page, and ShareKafkaListenerContainerFactory is the share-consumer equivalent of ConcurrentKafkaListenerContainerFactory:

@Configuration
public class OrderTaskShareConsumerConfig {

    @Bean
    public ShareConsumerFactory<String, OrderTaskEvent> orderTaskShareConsumerFactory(KafkaProperties properties) {
        Map<String, Object> configs = properties.buildConsumerProperties(null);
        return new DefaultShareConsumerFactory<>(configs);
    }

    @Bean
    public ShareKafkaListenerContainerFactory<String, OrderTaskEvent> shareKafkaListenerContainerFactory(
            ShareConsumerFactory<String, OrderTaskEvent> orderTaskShareConsumerFactory) {

        ShareKafkaListenerContainerFactory<String, OrderTaskEvent> factory =
                new ShareKafkaListenerContainerFactory<>(orderTaskShareConsumerFactory);
        factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.MANUAL);
        // a permanently-broken record is REJECTed instead of RELEASEd back for redelivery
        factory.setShareConsumerRecordRecoverer((record, ex) ->
                ex instanceof TransientTaskException ? AcknowledgeType.RELEASE : AcknowledgeType.REJECT);
        factory.setConcurrency(5);
        return factory;
    }
}
@Component
public class OrderTaskShareListener {

    @KafkaListener(
            id = "order-task-workers",
            topics = "orders.tasks",
            groupId = "order-task-workers",
            containerFactory = "shareKafkaListenerContainerFactory")
    public void process(ConsumerRecord<String, OrderTaskEvent> record, ShareAcknowledgment acknowledgment) {
        orderTaskProcessor.process(record.value());
        acknowledgment.acknowledge(); // ACCEPT
    }
}

ContainerProperties.ShareAckMode mirrors the protocol-level acknowledgement modes, plus a container-managed option ShareAcknowledgment never appears in:

ShareAckMode Behavior

EXPLICIT (default)

The container itself acknowledges ACCEPT on a successful listener invocation, and consults the configured ShareConsumerRecordRecoverer (REJECT by default) when the listener throws. The listener method never receives a ShareAcknowledgment argument.

MANUAL

The listener drives every outcome itself through an injected ShareAcknowledgment; subsequent polls on that consumer thread block until every record from the previous poll has been settled.

IMPLICIT

The broker auto-accepts every acquired record regardless of processing outcome, matching the protocol’s own share.acknowledgement.mode=implicit. No acknowledgment API is available to the listener at all.

ShareAcknowledgment exposes the same four outcomes as the plain-Java ShareConsumer example on Kafka Share Groups — acknowledge() (ACCEPT), release() (RELEASE), reject() (REJECT), and renew() (RENEW, extending the acquisition lock for a record still being worked on) — and is only ever non-null in MANUAL mode. ShareConsumerRecordRecoverer is a functional interface, (record, exception) → AcknowledgeType, that settles a listener exception in EXPLICIT mode without the listener seeing an acknowledgment object at all; ShareConsumerRecordRecoverer.REJECTING and .RELEASING are ready-made implementations for the two common policies.

As of Spring for Apache Kafka 4.1, share consumer support has real limitations beyond what the underlying protocol already lacks (covered on Kafka Share Groups): no message converters, no batch listeners, no partition assignment or topic-pattern subscription (both excluded by the share-group protocol itself), no manual offset seeking, and error handling goes exclusively through ShareConsumerRecordRecoverer rather than the CommonErrorHandler hierarchy used elsewhere on this page. See Spring for Apache Kafka reference — Kafka Queues (Share Consumers).

Exactly-once with KafkaTransactionManager

The baseline page shows transaction-id-prefix switching KafkaTemplate into transactional mode for a producer calling @Transactional directly; Kafka Reliability and Exactly-Once Semantics covers the underlying initTransactions()/beginTransaction()/ sendOffsetsToTransaction()/commitTransaction() protocol in full. The consume-transform-produce case — a @KafkaListener that both reads and writes Kafka inside the same transaction — is driven by a KafkaTransactionManager the container uses, rather than by the listener method calling those APIs itself:

@Bean
public KafkaTransactionManager<Object, Object> kafkaTransactionManager(ProducerFactory<Object, Object> producerFactory) {
    return new KafkaTransactionManager<>(producerFactory);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderPlacedEvent> ordersContainerFactory(
        ConsumerFactory<String, OrderPlacedEvent> consumerFactory,
        KafkaTransactionManager<Object, Object> kafkaTransactionManager) {

    ConcurrentKafkaListenerContainerFactory<String, OrderPlacedEvent> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.getContainerProperties().setTransactionManager(kafkaTransactionManager);
    return factory;
}
@Component
public class OrderEnrichmentListener {

    private final KafkaTemplate<String, OrderEnrichedEvent> kafkaTemplate;

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

    @KafkaListener(topics = "orders.placed", groupId = "order-enrichment")
    public void onOrderPlaced(OrderPlacedEvent event) {
        // the container already opened a Kafka transaction before this method ran;
        // this send and the eventual offset commit join that same transaction automatically
        kafkaTemplate.send("orders.enriched", event.orderId(), enrichmentService.enrich(event));
    }
}

With KafkaTemplate’s `ProducerFactory set to the same one backing kafkaTransactionManager, the container begins a transaction before invoking the listener, folds the consumed offset into it via sendOffsetsToTransaction() on success, and commits; an exception aborts the transaction and repositions the consumer to redeliver the rolled-back record(s) on the next poll — the read-process-write sequence is exactly-once, while the read and process steps individually remain at-least-once. KafkaTemplate uses KafkaAwareTransactionManager more broadly than just KafkaTransactionManager, but KafkaTransactionManager is the concrete implementation used here. ChainedKafkaTransactionManager is deprecated; combining a Kafka transaction with a separate resource (e.g. a database) now means letting the container’s KafkaTransactionManager own the Kafka side and adding a plain @Transactional on the listener method for the other resource’s own transaction manager. A downstream consumer that must only ever see committed output still needs isolation-level: read_committed, exactly as on the baseline page:

spring:
  kafka:
    producer:
      transaction-id-prefix: order-enrichment-tx-
    consumer:
      isolation-level: read_committed

See Spring for Apache Kafka reference — Transactions and Spring for Apache Kafka reference — Exactly Once Semantics.

Kafka Streams with @EnableKafkaStreams

Kafka Streams covers the topology/DSL concepts (KStream, KTable, windowing, state stores) this section does not repeat. @EnableKafkaStreams on a @Configuration class wires Spring’s own lifecycle around the native KafkaStreams client: it declares a StreamsBuilderFactoryBean (bean name defaultKafkaStreamsBuilder) that starts and stops the underlying KafkaStreams instance along with the Spring ApplicationContext, and makes a StreamsBuilder available for injection so topology can be built as @Bean methods evaluated once, at context refresh, before the streams instance starts:

@Configuration
@EnableKafkaStreams
public class OrderStreamsConfig {

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kafkaStreamsConfiguration(KafkaProperties properties) {
        return new KafkaStreamsConfiguration(properties.buildStreamsProperties(null));
    }

    @Bean
    public KStream<String, OrderPlacedEvent> ordersEnrichedStream(StreamsBuilder streamsBuilder) {
        KStream<String, OrderPlacedEvent> stream = streamsBuilder.stream("orders.placed");
        stream.mapValues(enrichmentService::enrich).to("orders.enriched");
        return stream;
    }
}
spring:
  kafka:
    streams:
      application-id: order-enrichment-streams
      bootstrap-servers: localhost:9092
      properties:
        errors.dead.letter.queue.topic.name: orders.enriched.DLT

Deserialization, processing, and production failures inside the topology each get their own recovering exception handler since Spring for Apache Kafka 4.1 (Kafka Streams DLQ support, KIP-1034): RecoveringDeserializationExceptionHandler, RecoveringProcessingExceptionHandler, and RecoveringProductionExceptionHandler, configured via the matching StreamsConfig handler-class property, route the offending record to a DeadLetterPublishingRecoverer (the same recoverer class the baseline page uses for @KafkaListener) or to a custom KafkaStreamsDeadLetterDestinationResolver when different failure types need different destinations:

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kafkaStreamsConfiguration(KafkaProperties properties,
        KafkaTemplate<Object, Object> kafkaTemplate) {

    Map<String, Object> props = properties.buildStreamsProperties(null);
    props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
            RecoveringDeserializationExceptionHandler.class);
    props.put(RecoveringDeserializationExceptionHandler.RECOVERER,
            new DeadLetterPublishingRecoverer(kafkaTemplate,
                    (record, ex) -> new TopicPartition("orders.enriched.DLT", -1)));
    return new KafkaStreamsConfiguration(props);
}

See Spring for Apache Kafka reference — Kafka Streams Support.

Jackson serializers and the new consumer group protocol

The baseline page’s spring.json.trusted.packages/spring.json.value.default.type examples target JsonSerializer/JsonDeserializer. Spring for Apache Kafka’s Jackson 3-based replacements, JacksonJsonSerializer/JacksonJsonDeserializer (same package, org.springframework.kafka.support.serializer), use the equivalent static configuration constants instead of raw property strings, and are the pair a new 4.1 application should reach for:

spring:
  kafka:
    producer:
      value-serializer: org.springframework.kafka.support.serializer.JacksonJsonSerializer
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.JacksonJsonDeserializer
      properties:
        spring.json.trusted.packages: "com.example.orders.events"
        spring.json.value.default.type: com.example.orders.events.OrderPlacedEvent
        # opt this consumer group into the KIP-848 broker-side group protocol instead of the classic one
        group.protocol: consumer

group.protocol: consumer (passed through under spring.kafka.consumer.properties, exactly like any other raw ConsumerConfig entry the baseline page already shows) switches the consumer group onto Kafka’s newer, broker-coordinated rebalance protocol instead of the classic client-side one; under it, ConsumerAwareRebalanceListener.onPartitionsAssigned may fire more than once per rebalance, as partitions arrive incrementally rather than all at once — code relying on a single call per rebalance (the seek example above included) should tolerate being invoked repeatedly with smaller assignment sets rather than assuming completeness on the first call. See Spring for Apache Kafka reference — Serialization, Deserialization, and Message Conversion.

New spring.kafka.* properties introduced on this page

Property Purpose

spring.kafka.consumer.isolation-level

read_committed restricts a consumer to records from committed transactions only; required alongside KafkaTransactionManager/transaction-id-prefix for end-to-end exactly-once.

spring.kafka.consumer.properties.group.protocol

consumer opts a listener container into the KIP-848 broker-side consumer group protocol instead of the classic client-side one.

spring.kafka.streams.application-id

The Kafka Streams application.id, used for the topology’s internal consumer group, changelog topics, and state store naming.

spring.kafka.streams.bootstrap-servers

Overrides spring.kafka.bootstrap-servers for the Kafka Streams client specifically, when the streams application talks to a different cluster than the rest of the application.

spring.kafka.streams.properties.errors.dead.letter.queue.topic.name

The default DLQ topic Kafka Streams' native exception handlers (KIP-1034) publish to when a KafkaStreamsDeadLetterDestinationResolver does not resolve a more specific destination.

References