The SeekToCurrentErrorHandler can now be configured to commit the offset of a recovered record when the container is configured with AckMode.MANUAL_IMMEDIATE (since 2.2.4). KafkaHeaders.DLT_ORIGINAL_OFFSET: The original offset. The processor does not change the key or value; it simply adds headers. To close existing Consumers, call stop() (and then start()) on the KafkaListenerEndpointRegistry and/or stop() and start() on any other listener container beans. Starting or stopping the registry will start or stop all the registered containers. If the first method is used, or the replyTimeout argument is null, the templates defaultReplyTimeout property is used (5 seconds by default). Starting with version 2.9.6, if you want to retain just the last value of these headers, use the configureDeadLetterPublishingContainerFactory() method shown above to set the factorys retainAllRetryHeaderValues property to false. There is an additional property returnPartialOnTimeout (default false). Any queue can be turned into a priority one using client-provided optional arguments (but, unlike other features that use optional arguments, not policies). For record listeners, when the AckMode is any manual value, offsets for already acknowledged records are committed. Starting with version 2.2, you can now use @KafkaListener as a meta annotation. To use the template, you can configure a producer factory and provide it in the templates constructor. The aggregate of partitions currently assigned to this containers child KafkaMessageListenerContainer s (explicitly or not). With the batch converter, the converted headers are available in the KafkaHeaders.BATCH_CONVERTED_HEADERS as a List> where the map in a position of the list corresponds to the data position in the payload. Then, if you are using the DeadLetterPublishingRecoverer to publish a failed record, the processor will send the recovered records offset in the original topic/partition to the transaction. So, before running tests with an embedded Kafka on random ports, we can set spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers as a system property - and the EmbeddedKafkaBroker will use it to expose its broker addresses. When using Spring Boot, (and you havent used start.spring.io to create your project), omit the version and Boot will automatically bring in the correct version that is compatible with your Boot version: Spring for Apache Kafka is designed to be used in a Spring Application Context. In a previous tutorial, we learned how to work with Spring and If you are using a DefaultMessageHandlerMethodFactory, set this resolver as the last custom resolver; the factory will ensure that this resolver will be used before the standard PayloadMethodArgumentResolver, which has no knowledge of KafkaNull payloads. The ConsumerSeekAware has new methods allowing you to perform seeks relative to the beginning, end, or current position and to seek to the first offset greater than or equal to a time stamp. ToStringSerializer.ADD_TYPE_INFO_HEADERS (default true): You can set it to false to disable this feature on the ToStringSerializer (sets the addTypeInfo property). With mode V2, it is not necessary to have a producer for each group.id/topic/partition because consumer metadata is sent along with the offsets to the transaction and the broker can determine if the producer is fenced using that information instead. Instead, they are registered with an infrastructure bean of type KafkaListenerEndpointRegistry. Headers of type MimeType and MediaType are now mapped as simple strings in the RecordHeader value. This might be useful if you want to create several containers with similar properties or you wish to use some externally configured factory, such as the one provided by Spring Boot auto-configuration. You can configure the Charset used to convert String to/from byte[] with the default being UTF-8. The next example combines @KafkaListener and @EventListener into a single class. Here, we have 4 listeners in two groups, g1 and g2. "my-topic" "my-topic-retry-0", "my-topic-retry-1", , "my-topic-dlt". It does not apply if the container is configured to listen to a topic pattern (regex). The ContainerProperties provides an idleBetweenPolls option to let the main loop in the listener container to sleep between KafkaConsumer.poll() calls. The releaseStrategy is now a BiConsumer. You can now modify @KafkaListener annotations during application initialization. You should save a reference to the callback. See Testing Applications for more information. The container commits any pending offset commits before calling the error handler. test-log and user-log. The following example shows how to do so: You can specify each partition in the partitions or partitionOffsets attribute but not both. Spring AOT native hints are provided to assist in developing native images for Spring applications that use Spring for Apache Kafka, including hints for AVRO generated classes used in @KafkaListener s. Some examples can be seen in the spring-aot-smoke-tests GitHub repository. If this is true, the initial offsets (positive or negative) are relative to the current position for this consumer. Starting with version 2.6.7, in addition to detecting DeserializationException s, the template will call the replyErrorChecker function, if provided. You can later start the DLT handler via the KafkaListenerEndpointRegistry. Starting with version 2.0, the @KafkaListener annotation has a new attribute: errorHandler. With EOSMode.V2 (aka BETA), the only supported mode, it is no longer necessary to use the same transactional.id, even for consumer-initiated transactions; in fact, it must be unique on each instance the same as for producer-initiated transactions. Also available in version 2.8.8 or later. True if a consumer pause has been requested. If you wish to use a different error handling strategy for record and batch listeners, the CommonMixedErrorHandler is provided allowing the configuration of a specific error handler for each listener type. As you can see, you have to define several infrastructure beans when not using Spring Boot. For example, a spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers entry (for testing in Spring Boot application) can be added into a junit-platform.properties file in the testing classpath. JsonDeserializer.KEY_TYPE_METHOD (default empty): See Using Methods to Determine Types. The following test case configuration snippet illustrates how to use this feature: You can provide a listener container with a KafkaAwareTransactionManager instance. To enable population of this header, set the container property deliveryAttemptHeader to true. Apache Kafka provides a mechanism to add interceptors to producers and consumers. Should the DLT processing fail, there are two possible behaviors available: ALWAYS_RETRY_ON_ERROR and FAIL_ON_ERROR. The default behavior is to suffix with the delay values, except for fixed delay configurations with multiple topics, in which case the topics are suffixed with the topics index. See Using ReplyingKafkaTemplate for more information. This section explores some of those techniques. This deserializer delegates to a real deserializer (key or value). You can use this method, for example, for setting initial offsets for the partitions, by calling the callback. Starting with version 2.0, a KafkaJaasLoginModuleInitializer class has been added to assist with Kerberos configuration. The default error handler is now the SeekToCurrentErrorHandler for record listeners and RecoveringBatchErrorHandler for batch listeners. ShouldSkipBothRetriesException.class would never be retried in any way and would go straight to the DLT if the first processing attempt failed. In the latter the consumer ends the execution without forwarding the message. The following listing shows the constructors signature: It also has a concurrency property. Type timer. See Pausing and Resuming Listener Containers for more information. For a parent container, the source and container properties are identical. Starting with version 2.3.5, the predicate is also called after a timeout (if returnPartialOnTimeout is true). In previous versions of Spring for Apache Kafka, the transactional.id was generated differently for transactions started by a listener container with a record-based listener, to support fencing zombies, which is not necessary any more, with EOSMode.V2 being the only option starting with 3.0. This header can be used in various places, such as a RecordInterceptor, RecordFilterStrategy and in the listener code itself. See Using KafkaMessageListenerContainer for more information. You should understand that the application listener gets events for all containers, so you may need to check the listener ID if you want to take specific action based on which container is idle. The listener containers created for @KafkaListener annotations are not beans in the application context. Aside from the logs, there was no indication that there was a problem. This map is also used to map raw incoming byte[] headers to String using the charset if, and only if, the boolean in the map value is true. On the consumer side, you can configure a JsonMessageConverter; it can handle ConsumerRecord values of type byte[], Bytes and String so should be used in conjunction with a ByteArrayDeserializer, BytesDeserializer or StringDeserializer. This class takes an implementation of RecordFilterStrategy in which you implement the filter method to signal that a message is a duplicate and should be discarded. You can now annotate @KafkaListener methods (and classes and @KafkaHandler methods) with @SendTo. Batch listeners can optionally receive the complete ConsumerRecords object instead of a List. The DefaultAfterRollbackProcessor and SeekToCurrentErrorHandler can now recover (skip) records that keep failing, and, by default, does so after 10 failures. See also Null Payloads and Log Compaction of 'Tombstone' Records. Conditional Delegating Error Handlers, D.5.9. The timeout passed into Consumer.poll() in milliseconds. The following example shows how to use it: Starting with version 2.2.4, you can also use the @EmbeddedKafka annotation to specify the Kafka ports property. Starting with versions 2.9.8, 3.0.6, you can provide a function in ContainerProperties micrometerTagsProvider; the function receives the ConsumerRecord and returns tags which can be based on that record, and merged with any static tags in micrometerTags. NO_OFFSET - there is no offset for a partition and the auto.offset.reset policy is none. Previously, you could pause a consumer within a ConsumerAwareMessageListener and resume it by listening for a ListenerContainerIdleEvent, which provides access to the Consumer object. Use the second method if you need to provide type information for the return type, to assist the message converter. AnnotationEnhancer is a BiFunction, AnnotatedElement, Map and must return a map of attributes. See Using KafkaMessageListenerContainer and Listener Container Properties for more information. See Aggregating Multiple Replies for more information. The following example uses both @KafkaListener and @EventListener: Note that you can obtain the current positions when idle is detected by implementing ConsumerSeekAware in your listener. When receiving a raw ConsumerRecord the integer is in a byte[4]. For another technique to send different types to different topics, see Using RoutingKafkaTemplate. This is similar functionality to throwing an exception when the container is configured with a DefaultErrorHandler. The preceding example uses the following configuration: When using an ErrorHandlingDeserializer with a batch listener, you must check for the deserialization exceptions in message headers. You can also receive null values for other reasons, such as a Deserializer that might return null when it cannot deserialize a value. KafkaHeaders.DLT_KEY_EXCEPTION_MESSAGE: The Exception message (key deserialization errors only). If the recoverer throws an exception, or the thread is interrupted during its sleep, the batch of records will be redelivered on the next poll. Since version 2.7.3, Spring for Apache Kafka provides the ExponentialBackOffWithMaxRetries which is a subclass that receives the maxRetries property and automatically calculates the maxElapsedTime, which is a little more convenient. The recoverer requires a KafkaTemplate, which is used to send the record. Accessing Topics' Information at Runtime, 4.2.12. The KafkaStreamBrancher has been introduced for better end-user experience when conditional branches are built on top of KStream instance. See Message Headers for more information. To replace any BatchErrorHandler implementation, you should implement handleBatch() The following code shows how to do so: The following examples show how to validate: Starting with version 2.5.11, validation now works on payloads for @KafkaHandler methods in a class-level listener. Starting with version 2.7.3, a new component ContainerGroupSequencer has been introduced. All you need is to declare a KafkaStreamsConfiguration bean named defaultKafkaStreamsConfig. The JsonDeserializer now provides TypeReference-based constructors for better handling of target generic container types. For convenience, the framework also provides an ABSwitchCluster which supports two sets of bootstrap servers; one of which is active at any time. On the outbound side, by default, all MessageHeaders are mapped, except id, timestamp, and the headers that map to ConsumerRecord properties. If you wish this condition to be considered fatal, set the admins fatalIfBrokerNotAvailable property to true. When using a batch listener, version 2.4.2 introduced an alternative mechanism to deal with failures while processing a batch; the BatchToRecordAdapter. You can also provide a custom implementation of Spring Retrys SleepingBackOffPolicy interface: You can set the global timeout for the retrying process. Key Features of Apache Kafka What is Spring Boot? When you use Log Compaction, you can send and receive messages with null payloads to identify the deletion of a key. You can disable this by setting the addTypeInfo property to false. See Forwarding Listener Results using @SendTo for more information about sending replies. Starting with version 2.4.3, you can set the templates allowNonTransactional property to true. Producer and Consumer Factory Changes, D.7.14. Normally, when a KafkaTemplate is transactional (configured with a transaction-capable producer factory), transactions are required. Producer Interceptor Managed in Spring, 4.1.17. If the listener throws an exception, the transaction is rolled back and the consumer is repositioned so that the rolled-back record(s) can be retrieved on the next poll. The corresponding @KafkaListener s for this example are shown in Annotation Properties. See After-rollback Processor. As discussed in @KafkaListener Annotation, a ConcurrentKafkaListenerContainerFactory is used to create containers for annotated methods. Starting with version 2.2.5, the DefaultAfterRollbackProcessor can be invoked in a new transaction (started after the failed transaction rolls back). Events related to consumer authentication and authorization. You can get a reference to the bean from the application context, such as auto-wiring, to manage its registered containers. The following example shows how to add a ReplyHeadersConfigurer: You can also add more headers if you wish. An Implementation for native Micrometer metrics is provided. The ContainerProperties class has been moved from org.springframework.kafka.listener.config to org.springframework.kafka.listener. This can be used to add headers within the stream processing; the header values are SpEL expressions; the root object of the expression evaluation has 3 properties: record - the org.apache.kafka.streams.processor.api.Record (key, value, timestamp, headers), context - the ProcessorContext, allowing access to the current record metadata. The Acknowledgment has the following method: This method gives the listener control over when offsets are committed. 3. The following example creates beans that use this method: Note that, for this to work, the method signature for the conversion target must be a container object with a single generic parameter type, such as the following: Note that you can still access the batch headers. When you use this API, the DefaultKafkaProducerFactory and DefaultKafkaConsumerFactory also provide properties (through constructors or setter methods) to inject custom Serializer and Deserializer instances into the target Producer or Consumer. The 0.11.0.0 client introduced support for headers in messages. With @KafkaListener, this value is obtained from the info attribute. In addition, a FilteringBatchMessageListenerAdapter is provided, for when you use a batch message listener. For five TopicPartitionOffset instances, two containers get two partitions, and the third gets one. When this is set to true, instead of completing the future with a KafkaReplyTimeoutException, a partial result completes the future normally (as long as at least one reply record has been received). You can modify this behavior by setting the monitorInterval (default 30 seconds) and noPollThreshold (default 3.0) properties in the ContainerProperties when configuring the listener container. The @KafkaListener annotation provides a mechanism for simple POJO listeners. In combination with the global retryable topics fatal exceptions classification, you can configure the framework for any behavior youd like, such as having some exceptions trigger both blocking and non-blocking retries, trigger only one kind or the other, or go straight to the DLT without retries of any kind. Starting with version 2.7, it has additional methods which are called after the listener exits (normally, or by throwing an exception). This might be used, for example, to access the consumer metrics in the interceptor. Default. Four constants in KafkaHeaders that were deprecated in 2.9.x have now been removed. Question - 1 : As per Spring-Kafka documentation, there are 2 ways to implement Kafka-Consumer; "You can receive messages by configuring a Notice that the send methods return a CompletableFuture. The bootstrapping of Non-Blocking Retries infrastructure beans has changed in this release to avoid some timing problems that occurred in some application regarding application initialization. This allows a Kafka streams topology to interact with a spring-messaging component, such as a Spring Integration flow. See noPollThreshold and pollTimeout. The container property restartAfterAuthException has been added. This allows the same container factory to be used for both record and batch listeners. Further, you can explicitly configure the groupId on the annotation. Error handlers and after rollback processors that extend FailedRecordProcessor can now be configured with one or more RetryListener s to receive information about retry and recovery progress. The execute method provides direct access to the underlying Producer. Starting with version 2.7, you can declare multiple NewTopic s in a single KafkaAdmin.NewTopics bean definition: By default, if the broker is not available, a message is logged, but the context continues to load. KafkaListenerObservation.DefaultKafkaListenerObservationConvention, KafkaTemplateObservation.DefaultKafkaTemplateObservationConvention. a LinkedHashMap) because it is traversed in order; you should add more specific patterns at the beginning. Starting with version 2.1.1, you can now set the client.id property for consumers created by the annotation. Also since version 2.7 ConsumerPartitionPausedEvent and ConsumerPartitionResumedEvent instances are published with the container as the source property and the TopicPartition instance. these methods now require an ObjectProvider parameter. Previously, the value was populated but the key DeserializationException remained in the headers. See String serialization for more information. The ChainedKafkaTransactionManager is now deprecated, since version 2.7; see the javadocs for its super class ChainedTransactionManager for more information. Version 2.1.3 introduced the ChainedKafkaTransactionManager. Starting with version 2.2.5, you can specify that certain string-valued headers should not be mapped using JSON, but to/from a raw byte[]. Events related to consumer authentication and authorization failures are now published by the container. Please submit GitHub issues and/or pull requests for additional entries in that chapter. The attribute values can contain SpEL and/or property placeholders; the enhancer is called before any resolution is performed. With the concurrent container, timers are created for each thread and the, Starting with version 2.5.8, you can now configure the. For convenience, starting with version 2.3, the framework also provides a StringOrBytesSerializer which can serialize all three value types so it can be used with any of the message converters. See the Kafka documentation for more information. As with most annotation properties, you can use SpEL expressions; for an example of how to generate a large list of partitions, see Manually Assigning All Partitions. KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN: The Exception cause class name, if present (since version 2.8). If you have set a linger.ms, you may wish to invoke flush() before waiting or, for convenience, the template has a constructor with an autoFlush parameter that causes the template to flush() on each send. Spring for Apache Kafka also provides JsonSerializer and JsonDeserializer implementations that are based on the Listeners can be configured to receive the entire batch of messages returned by the consumer.poll() operation, rather than one at a time. You can also extend them to implement some particular configuration logic in the configure(Map configs, boolean isKey) method. With AssertJ, the final part looks like the following code: The kafka-clients library provides MockConsumer and MockProducer classes for testing purposes. After that, the same semantics as BATCH are applied. There is a 30-second default maximum delay for the. Starting with version 3.0, you can set a different concurrency for the retry containers (either on the annotation, or in RetryConfigurationBuilder). KafkaHeaders.DLT_KEY_EXCEPTION_FQCN: The Exception class name (key deserialization errors only). These listeners can be used, for example, to create and bind a Micrometer KafkaClientMetrics instance when a new client is created (and close it when the client is closed). The listener container will defer the out-of-order commits until the missing acknowledgments are received. The map can be configured using a constructor, or via properties (a comma delimited list of pattern:serializer). An object of type FailedDeserializationInfo, which contains all the contextual information is provided to the function. Anything returned by the error handler is ignored. JsonDeserializer.KEY_DEFAULT_TYPE: Fallback type for deserialization of keys if no header information is present. They are now simple strings for interoperability.
