Spring for Apache Kafka

You can synchronize Kafka producer-only transactions with other transaction managers (e.g., DataSourceTransactionManager for databases).

Default Behavior (DB-first): When using @Transactional on a method that performs both DB and Kafka operations, the interceptor starts the transaction. The KafkaTemplate synchronizes with the existing transaction manager. When the method exits, the database transaction commits first, followed by the Kafka transaction.

Kafka-first Behavior: If you require the Kafka transaction to commit before the database transaction, use nested @Transactional methods:

1. The outer method should be configured to use the DataSourceTransactionManager.
2. The inner method should be configured to use the KafkaTransactionManager.

Error Handling Note: Starting with versions 2.5.17, 2.6.12, 2.7.9, and 2.8.0, if the synchronized Kafka transaction fails to commit after the primary transaction (e.g., the DB) has already committed, the exception is thrown to the caller. You should implement remedial actions to compensate for the already-committed primary transaction.

Spring provides specialized exception handlers that follow standard recovery strategies (DLQ routing, custom ConsumerRecordRecoverer, or failing the stream).

Available Exception Handlers

• RecoveringDeserializationExceptionHandler: Handles deserialization errors (since v2.3).
• RecoveringProcessingExceptionHandler: Handles errors during stream processing (since v4.1).
• RecoveringProductionExceptionHandler: Handles errors during record production or serialization (since v4.1).

Recovery Priority Order

1. If a KafkaStreamsDeadLetterDestinationResolver is defined: Resume and forward to the resolved topic-partition using native Kafka Streams DLQ.
2. If errors.dead.letter.queue.topic.name is set: Resume and forward to that specific topic using native Kafka Streams DLQ.
3. If a ConsumerRecordRecoverer is defined: Invoke it and resume without producing DLQ records (e.g., using DeadLetterPublishingRecoverer).
4. Fail the stream without producing dead-letter records.

You can control what happens if the DLT processing method itself fails using the dltProcessingFailureStrategy property (via @RetryableTopic) or the .doNotRetryOnDltFailure() method (via RetryTopicConfigurationBuilder).

Available strategies:

• DltStrategy.ALWAYS_RETRY_ON_ERROR (Default): The record is forwarded back to the DLT topic so it doesn't block other DLT records. Note: Since version 2.8.3, this will NOT retry if the record causes a fatal exception (e.g., DeserializationException).
• DltStrategy.FAIL_ON_ERROR: The consumer ends execution without forwarding the message.

Fatal Exceptions (which prevent automatic retry in ALWAYS_RETRY_ON_ERROR mode):

• DeserializationException
• MessageConversionException
• ConversionException
• MethodArgumentResolutionException
• NoSuchMethodException
• ClassCastException

You can manage this list using the DestinationTopicResolver bean.

@RetryableTopic(dltProcessingFailureStrategy =
            DltStrategy.FAIL_ON_ERROR)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}

Starting in version 2.8.4, Spring Kafka allows you to use both blocking and non-blocking retries in conjunction. This is useful for handling different types of errors differently: for example, retrying transient errors (like DatabaseAccessException) locally via blocking retries before moving the record to a non-blocking retry topic.

Configuration Logic

• Blocking Retries (Allowlist): You explicitly define which exceptions should trigger blocking retries using configureBlockingRetries.
• Non-Blocking Fatal Exceptions (Denylist): You define which exceptions are considered "fatal" for non-blocking retries (meaning they should skip retry topics and go straight to the DLT) by overriding manageNonBlockingFatalExceptions.

Behavior Matrix

When configured together, the behavior for an exception depends on its classification in both systems:

@Override
protected void configureBlockingRetries(BlockingRetriesConfigurer blockingRetries) {
    blockingRetries
            .retryOn(ShouldRetryOnlyBlockingException.class, ShouldRetryViaBothException.class)
            .backOff(new FixedBackOff(50, 3));
}

@Override
protected void manageNonBlockingFatalExceptions(List<Class<? extends Throwable>> nonBlockingFatalExceptions) {
    nonBlockingFatalExceptions.add(ShouldSkipBothRetriesException.class);
}

When using a seek-after-handling error handler (like DefaultErrorHandler) with an asynchronous listener, the listener receives one extra dispatch-loop delivery per failing record on the same partition compared to a blocking listener.

Why this happens: The dispatch loop invokes the listener on every polled record before any asynchronous failure callback fires. In a burst of "poison-pill" records, the first poll invokes every record once before any retry state is registered on the partition. Subsequent retry cycles for the head record will then skip the records queued behind it.

Invocation Count Comparison: For N always-failing records on a single partition with maxAttempts = n:

• Async/Suspend Listener: Receives N * (n + 2) - 1 total invocations.
• Blocking Listener: Receives N * (n + 1) total invocations.

Mitigation Strategies: If your listener performs non-idempotent work (e.g., outbound HTTP calls or non-transactional DB writes) and you want to avoid these extra deliveries:

1. Use @RetryableTopic (non-blocking retry): Failing records are routed to a separate retry topic, allowing the main partition to advance without extra dispatch-loop deliveries.
2. Make the listener idempotent: This aligns with Kafka's default at-least-once delivery model.

Decision Guide:

• Use Seek-based handlers (Blocking): For events requiring strict per-partition ordering (e.g., state transitions keyed by entity ID).
• Use @RetryableTopic (Async): For events that do not require ordering (e.g., search index updates, notifications, downstream enrichment).

When using non-blocking retries with Spring Kafka, message processing and backing off are managed by the consumer thread. Consequently, delay precision is provided on a best-effort basis.

Key Guarantees

• No Early Processing: It is guaranteed that a message will never be processed before its scheduled due time.
• Single Partition Precision: For consumers handling only a single partition, message processing should occur approximately at the exact due time in most scenarios.

Factors Affecting Precision

Several factors can cause delays to be longer than expected:

• Processing Overlap: If processing a previous message takes longer than the back-off period of the next message, the next message's delay will be increased.
• Short Delays: For delays of approximately 1 second or less, maintenance tasks (such as committing offsets) may delay execution.
• Multi-Partition Consumers: Precision is reduced if a retry topic's consumer handles multiple partitions, as the system relies on waking up the consumer from polling and utilizing full pollTimeouts to make timing adjustments.

Listener containers created via @KafkaListener are not standard beans in the application context. Instead, they are managed by the KafkaListenerEndpointRegistry.

Key lifecycle behaviors:

• Auto-startup: The registry automatically starts containers where autoStartup is set to true. This property on the @KafkaListener annotation overrides the default setting in the container factory.
• Registry Management: Starting or stopping the KafkaListenerEndpointRegistry will start or stop all containers it manages.
• Late Registration: Endpoints registered after the application context has refreshed (e.g., prototype beans created later) will start immediately regardless of autoStartup, to comply with the SmartLifecycle contract.
• Version 2.8.7+ Note: You can set the registry's alwaysStartAfterRefresh property to false to ensure the container's autoStartup property is respected even for late-registered endpoints.

To manage a container programmatically, autowire the KafkaListenerEndpointRegistry and use the container's id.

@KafkaListener(id = "myContainer", topics = "myTopic", autoStartup = "false")
public void listen(...) { ... }

@Autowired
private KafkaListenerEndpointRegistry registry;

// Start the specific container manually
this.registry.getListenerContainer("myContainer").start();

Starting with version 3.2, Spring Boot applications use the spring.application.name property as a default prefix for auto-generated client IDs. This improves observability when troubleshooting or applying quotas on the Kafka server side.

Prefix Behavior:

• Consumer (without consumer group): Prefixed with application name.
• Consumer (with consumer group): The group ID is used; the application name prefix is not applied to the client ID in the same way (the group ID remains the primary identifier).
• Producer: Prefixed with application name.
• Admin: Prefixed with application name.

Example Mapping (for spring.application.name=myapp):

Spring introduces StreamsBuilderFactoryBean to manage the lifecycle of Kafka Streams within the Spring application context. It implements SmartLifecycle and exposes a StreamsBuilder singleton as a bean.

Key Behaviors

• Lifecycle: It manages an internal KafkaStreams instance. Unlike the native API where a closed KafkaStreams instance cannot be restarted, StreamsBuilderFactoryBean can be safely stopped and restarted; a new KafkaStreams instance is created on each start().
• Auto-startup: If autoStartup = true (default), you must declare your KStream instances on the StreamsBuilder before the application context is refreshed. You can do this by defining KStream as a regular bean.
• Manual Control: To control the lifecycle manually (e.g., stopping/starting based on a condition), reference the factory bean directly using the & prefix (e.g., @Qualifier("&myKStreamBuilder")).
• Customization: You can delegate options like KafkaStreams.StateListener, Thread.UncaughtExceptionHandler, and StateRestoreListener to the internal instance, or use a KafkaStreamsCustomizer for deeper configuration.

@Bean
public FactoryBean<StreamsBuilder> myKStreamBuilder(KafkaStreamsConfiguration streamsConfig) {
    return new StreamsBuilderFactoryBean(streamsConfig);
}

// To declare a KStream that starts with the context:
@Bean
public KStream<?, ?> kStream(StreamsBuilder kStreamBuilder) {
    KStream<Integer, String> stream = kStreamBuilder.stream(STREAMING_TOPIC1);
    // Fluent KStream API
    return stream;
}