sendAndReceive(Message message);\n\n RequestReplyTypedMessageFuture sendAndReceive(Message message,\n ParameterizedTypeReference returnType);\n")])])]),a("p",[e._v("These will use the template’s default "),a("code",[e._v("replyTimeout")]),e._v(", there are also overloaded versions that can take a timeout in the method call.")]),e._v(" "),a("p",[e._v("Use the first method if the consumer’s "),a("code",[e._v("Deserializer")]),e._v(" or the template’s "),a("code",[e._v("MessageConverter")]),e._v(" can convert the payload without any additional information, either via configuration or type metadata in the reply message.")]),e._v(" "),a("p",[e._v("Use the second method if you need to provide type information for the return type, to assist the message converter.\nThis also allows the same template to receive different types, even if there is no type metadata in the replies, such as when the server side is not a Spring application.\nThe following is an example of the latter:")]),e._v(" "),a("p",[e._v("Example 6. Template Bean")]),e._v(" "),a("p",[e._v("Java")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Bean\nReplyingKafkaTemplate template(\n ProducerFactory pf,\n ConcurrentKafkaListenerContainerFactory factory) {\n\n ConcurrentMessageListenerContainer replyContainer =\n factory.createContainer("replies");\n replyContainer.getContainerProperties().setGroupId("request.replies");\n ReplyingKafkaTemplate template =\n new ReplyingKafkaTemplate<>(pf, replyContainer);\n template.setMessageConverter(new ByteArrayJsonMessageConverter());\n template.setDefaultTopic("requests");\n return template;\n}\n')])])]),a("p",[e._v("Kotlin")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Bean\nfun template(\n pf: ProducerFactory?,\n factory: ConcurrentKafkaListenerContainerFactory\n): ReplyingKafkaTemplate {\n val replyContainer = factory.createContainer("replies")\n replyContainer.containerProperties.groupId = "request.replies"\n val template = ReplyingKafkaTemplate(pf, replyContainer)\n template.messageConverter = ByteArrayJsonMessageConverter()\n template.defaultTopic = "requests"\n return template\n}\n')])])]),a("p",[e._v("Example 7. Using the template")]),e._v(" "),a("p",[e._v("Java")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('RequestReplyTypedMessageFuture future1 =\n template.sendAndReceive(MessageBuilder.withPayload("getAThing").build(),\n new ParameterizedTypeReference() { });\nlog.info(future1.getSendFuture().get(10, TimeUnit.SECONDS).getRecordMetadata().toString());\nThing thing = future1.get(10, TimeUnit.SECONDS).getPayload();\nlog.info(thing.toString());\n\nRequestReplyTypedMessageFuture> future2 =\n template.sendAndReceive(MessageBuilder.withPayload("getThings").build(),\n new ParameterizedTypeReference>() { });\nlog.info(future2.getSendFuture().get(10, TimeUnit.SECONDS).getRecordMetadata().toString());\nList things = future2.get(10, TimeUnit.SECONDS).getPayload();\nthings.forEach(thing1 -> log.info(thing1.toString()));\n')])])]),a("p",[e._v("Kotlin")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('val future1: RequestReplyTypedMessageFuture? =\n template.sendAndReceive(MessageBuilder.withPayload("getAThing").build(),\n object : ParameterizedTypeReference() {})\nlog.info(future1?.sendFuture?.get(10, TimeUnit.SECONDS)?.recordMetadata?.toString())\nval thing = future1?.get(10, TimeUnit.SECONDS)?.payload\nlog.info(thing.toString())\n\nval future2: RequestReplyTypedMessageFuture?>? =\n template.sendAndReceive(MessageBuilder.withPayload("getThings").build(),\n object : ParameterizedTypeReference?>() {})\nlog.info(future2?.sendFuture?.get(10, TimeUnit.SECONDS)?.recordMetadata.toString())\nval things = future2?.get(10, TimeUnit.SECONDS)?.payload\nthings?.forEach(Consumer { thing1: Thing? -> log.info(thing1.toString()) })\n')])])]),a("h5",{attrs:{id:"reply-type-message"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#reply-type-message"}},[e._v("#")]),e._v(" Reply Type Message")]),e._v(" "),a("p",[e._v("When the "),a("code",[e._v("@KafkaListener")]),e._v(" returns a "),a("code",[e._v("Message")]),e._v(", with versions before 2.5, it was necessary to populate the reply topic and correlation id headers.\nIn this example, we use the reply topic header from the request:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "requestor", topics = "request")\n@SendTo\npublic Message messageReturn(String in) {\n return MessageBuilder.withPayload(in.toUpperCase())\n .setHeader(KafkaHeaders.TOPIC, replyTo)\n .setHeader(KafkaHeaders.MESSAGE_KEY, 42)\n .setHeader(KafkaHeaders.CORRELATION_ID, correlation)\n .build();\n}\n')])])]),a("p",[e._v("This also shows how to set a key on the reply record.")]),e._v(" "),a("p",[e._v("Starting with version 2.5, the framework will detect if these headers are missing and populate them with the topic - either the topic determined from the "),a("code",[e._v("@SendTo")]),e._v(" value or the incoming "),a("code",[e._v("KafkaHeaders.REPLY_TOPIC")]),e._v(" header (if present).\nIt will also echo the incoming "),a("code",[e._v("KafkaHeaders.CORRELATION_ID")]),e._v(" and "),a("code",[e._v("KafkaHeaders.REPLY_PARTITION")]),e._v(", if present.")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "requestor", topics = "request")\n@SendTo // default REPLY_TOPIC header\npublic Message messageReturn(String in) {\n return MessageBuilder.withPayload(in.toUpperCase())\n .setHeader(KafkaHeaders.MESSAGE_KEY, 42)\n .build();\n}\n')])])]),a("h5",{attrs:{id:"aggregating-multiple-replies"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#aggregating-multiple-replies"}},[e._v("#")]),e._v(" Aggregating Multiple Replies")]),e._v(" "),a("p",[e._v("The template in "),a("a",{attrs:{href:"#replying-template"}},[e._v("Using "),a("code",[e._v("ReplyingKafkaTemplate")])]),e._v(" is strictly for a single request/reply scenario.\nFor cases where multiple receivers of a single message return a reply, you can use the "),a("code",[e._v("AggregatingReplyingKafkaTemplate")]),e._v(".\nThis is an implementation of the client-side of the "),a("a",{attrs:{href:"https://www.enterpriseintegrationpatterns.com/patterns/messaging/BroadcastAggregate.html",target:"_blank",rel:"noopener noreferrer"}},[e._v("Scatter-Gather Enterprise Integration Pattern"),a("OutboundLink")],1),e._v(".")]),e._v(" "),a("p",[e._v("Like the "),a("code",[e._v("ReplyingKafkaTemplate")]),e._v(", the "),a("code",[e._v("AggregatingReplyingKafkaTemplate")]),e._v(" constructor takes a producer factory and a listener container to receive the replies; it has a third parameter "),a("code",[e._v("BiPredicate>, Boolean> releaseStrategy")]),e._v(" which is consulted each time a reply is received; when the predicate returns "),a("code",[e._v("true")]),e._v(", the collection of "),a("code",[e._v("ConsumerRecord")]),e._v(" s is used to complete the "),a("code",[e._v("Future")]),e._v(" returned by the "),a("code",[e._v("sendAndReceive")]),e._v(" method.")]),e._v(" "),a("p",[e._v("There is an additional property "),a("code",[e._v("returnPartialOnTimeout")]),e._v(" (default false).\nWhen this is set to "),a("code",[e._v("true")]),e._v(", instead of completing the future with a "),a("code",[e._v("KafkaReplyTimeoutException")]),e._v(", a partial result completes the future normally (as long as at least one reply record has been received).")]),e._v(" "),a("p",[e._v("Starting with version 2.3.5, the predicate is also called after a timeout (if "),a("code",[e._v("returnPartialOnTimeout")]),e._v(" is "),a("code",[e._v("true")]),e._v(").\nThe first argument is the current list of records; the second is "),a("code",[e._v("true")]),e._v(" if this call is due to a timeout.\nThe predicate can modify the list of records.")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("AggregatingReplyingKafkaTemplate template =\n new AggregatingReplyingKafkaTemplate<>(producerFactory, container,\n coll -> coll.size() == releaseSize);\n...\nRequestReplyFuture>> future =\n template.sendAndReceive(record);\nfuture.getSendFuture().get(10, TimeUnit.SECONDS); // send ok\nConsumerRecord>> consumerRecord =\n future.get(30, TimeUnit.SECONDS);\n")])])]),a("p",[e._v("Notice that the return type is a "),a("code",[e._v("ConsumerRecord")]),e._v(" with a value that is a collection of "),a("code",[e._v("ConsumerRecord")]),e._v(' s.\nThe "outer" '),a("code",[e._v("ConsumerRecord")]),e._v(' is not a "real" record, it is synthesized by the template, as a holder for the actual reply records received for the request.\nWhen a normal release occurs (release strategy returns true), the topic is set to '),a("code",[e._v("aggregatedResults")]),e._v("; if "),a("code",[e._v("returnPartialOnTimeout")]),e._v(" is true, and timeout occurs (and at least one reply record has been received), the topic is set to "),a("code",[e._v("partialResultsAfterTimeout")]),e._v('.\nThe template provides constant static variables for these "topic" names:')]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('/**\n * Pseudo topic name for the "outer" {@link ConsumerRecords} that has the aggregated\n * results in its value after a normal release by the release strategy.\n */\npublic static final String AGGREGATED_RESULTS_TOPIC = "aggregatedResults";\n\n/**\n * Pseudo topic name for the "outer" {@link ConsumerRecords} that has the aggregated\n * results in its value after a timeout.\n */\npublic static final String PARTIAL_RESULTS_AFTER_TIMEOUT_TOPIC = "partialResultsAfterTimeout";\n')])])]),a("p",[e._v("The real "),a("code",[e._v("ConsumerRecord")]),e._v(" s in the "),a("code",[e._v("Collection")]),e._v(" contain the actual topic(s) from which the replies are received.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("The listener container for the replies MUST be configured with "),a("code",[e._v("AckMode.MANUAL")]),e._v(" or "),a("code",[e._v("AckMode.MANUAL_IMMEDIATE")]),e._v("; the consumer property "),a("code",[e._v("enable.auto.commit")]),e._v(" must be "),a("code",[e._v("false")]),e._v(" (the default since version 2.3)."),a("br"),e._v("To avoid any possibility of losing messages, the template only commits offsets when there are zero requests outstanding, i.e. when the last outstanding request is released by the release strategy."),a("br"),e._v("After a rebalance, it is possible for duplicate reply deliveries; these will be ignored for any in-flight requests; you may see error log messages when duplicate replies are received for already released replies.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("If you use an "),a("a",{attrs:{href:"#error-handling-deserializer"}},[a("code",[e._v("ErrorHandlingDeserializer")])]),e._v(" with this aggregating template, the framework will not automatically detect "),a("code",[e._v("DeserializationException")]),e._v(" s."),a("br"),e._v("Instead, the record (with a "),a("code",[e._v("null")]),e._v(" value) will be returned intact, with the deserialization exception(s) in headers."),a("br"),e._v("It is recommended that applications call the utility method "),a("code",[e._v("ReplyingKafkaTemplate.checkDeserialization()")]),e._v(" method to determine if a deserialization exception occurred."),a("br"),e._v("See its javadocs for more information."),a("br"),e._v("The "),a("code",[e._v("replyErrorChecker")]),e._v(" is also not called for this aggregating template; you should perform the checks on each element of the reply.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h4",{attrs:{id:"_4-1-4-receiving-messages"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-4-receiving-messages"}},[e._v("#")]),e._v(" 4.1.4. Receiving Messages")]),e._v(" "),a("p",[e._v("You can receive messages by configuring a "),a("code",[e._v("MessageListenerContainer")]),e._v(" and providing a message listener or by using the "),a("code",[e._v("@KafkaListener")]),e._v(" annotation.")]),e._v(" "),a("h5",{attrs:{id:"message-listeners"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#message-listeners"}},[e._v("#")]),e._v(" Message Listeners")]),e._v(" "),a("p",[e._v("When you use a "),a("a",{attrs:{href:"#message-listener-container"}},[e._v("message listener container")]),e._v(", you must provide a listener to receive data.\nThere are currently eight supported interfaces for message listeners.\nThe following listing shows these interfaces:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("public interface MessageListener { (1)\n\n void onMessage(ConsumerRecord data);\n\n}\n\npublic interface AcknowledgingMessageListener { (2)\n\n void onMessage(ConsumerRecord data, Acknowledgment acknowledgment);\n\n}\n\npublic interface ConsumerAwareMessageListener extends MessageListener { (3)\n\n void onMessage(ConsumerRecord data, Consumer consumer);\n\n}\n\npublic interface AcknowledgingConsumerAwareMessageListener extends MessageListener { (4)\n\n void onMessage(ConsumerRecord data, Acknowledgment acknowledgment, Consumer consumer);\n\n}\n\npublic interface BatchMessageListener { (5)\n\n void onMessage(List> data);\n\n}\n\npublic interface BatchAcknowledgingMessageListener { (6)\n\n void onMessage(List> data, Acknowledgment acknowledgment);\n\n}\n\npublic interface BatchConsumerAwareMessageListener extends BatchMessageListener { (7)\n\n void onMessage(List> data, Consumer consumer);\n\n}\n\npublic interface BatchAcknowledgingConsumerAwareMessageListener extends BatchMessageListener { (8)\n\n void onMessage(List> data, Acknowledgment acknowledgment, Consumer consumer);\n\n}\n")])])]),a("table",[a("thead",[a("tr",[a("th",[a("strong",[e._v("1")])]),e._v(" "),a("th",[e._v("Use this interface for processing individual "),a("code",[e._v("ConsumerRecord")]),e._v(" instances received from the Kafka consumer "),a("code",[e._v("poll()")]),e._v(" operation when using auto-commit or one of the container-managed "),a("a",{attrs:{href:"#committing-offsets"}},[e._v("commit methods")]),e._v(".")])])]),e._v(" "),a("tbody",[a("tr",[a("td",[a("strong",[e._v("2")])]),e._v(" "),a("td",[e._v("Use this interface for processing individual "),a("code",[e._v("ConsumerRecord")]),e._v(" instances received from the Kafka consumer "),a("code",[e._v("poll()")]),e._v(" operation when using one of the manual "),a("a",{attrs:{href:"#committing-offsets"}},[e._v("commit methods")]),e._v(".")])]),e._v(" "),a("tr",[a("td",[a("strong",[e._v("3")])]),e._v(" "),a("td",[e._v("Use this interface for processing individual "),a("code",[e._v("ConsumerRecord")]),e._v(" instances received from the Kafka consumer "),a("code",[e._v("poll()")]),e._v(" operation when using auto-commit or one of the container-managed "),a("a",{attrs:{href:"#committing-offsets"}},[e._v("commit methods")]),e._v("."),a("br"),e._v("Access to the "),a("code",[e._v("Consumer")]),e._v(" object is provided.")])]),e._v(" "),a("tr",[a("td",[a("strong",[e._v("4")])]),e._v(" "),a("td",[e._v("Use this interface for processing individual "),a("code",[e._v("ConsumerRecord")]),e._v(" instances received from the Kafka consumer "),a("code",[e._v("poll()")]),e._v(" operation when using one of the manual "),a("a",{attrs:{href:"#committing-offsets"}},[e._v("commit methods")]),e._v("."),a("br"),e._v("Access to the "),a("code",[e._v("Consumer")]),e._v(" object is provided.")])]),e._v(" "),a("tr",[a("td",[a("strong",[e._v("5")])]),e._v(" "),a("td",[e._v("Use this interface for processing all "),a("code",[e._v("ConsumerRecord")]),e._v(" instances received from the Kafka consumer "),a("code",[e._v("poll()")]),e._v(" operation when using auto-commit or one of the container-managed "),a("a",{attrs:{href:"#committing-offsets"}},[e._v("commit methods")]),e._v("."),a("code",[e._v("AckMode.RECORD")]),e._v(" is not supported when you use this interface, since the listener is given the complete batch.")])]),e._v(" "),a("tr",[a("td",[a("strong",[e._v("6")])]),e._v(" "),a("td",[e._v("Use this interface for processing all "),a("code",[e._v("ConsumerRecord")]),e._v(" instances received from the Kafka consumer "),a("code",[e._v("poll()")]),e._v(" operation when using one of the manual "),a("a",{attrs:{href:"#committing-offsets"}},[e._v("commit methods")]),e._v(".")])]),e._v(" "),a("tr",[a("td",[a("strong",[e._v("7")])]),e._v(" "),a("td",[e._v("Use this interface for processing all "),a("code",[e._v("ConsumerRecord")]),e._v(" instances received from the Kafka consumer "),a("code",[e._v("poll()")]),e._v(" operation when using auto-commit or one of the container-managed "),a("a",{attrs:{href:"#committing-offsets"}},[e._v("commit methods")]),e._v("."),a("code",[e._v("AckMode.RECORD")]),e._v(" is not supported when you use this interface, since the listener is given the complete batch."),a("br"),e._v("Access to the "),a("code",[e._v("Consumer")]),e._v(" object is provided.")])]),e._v(" "),a("tr",[a("td",[a("strong",[e._v("8")])]),e._v(" "),a("td",[e._v("Use this interface for processing all "),a("code",[e._v("ConsumerRecord")]),e._v(" instances received from the Kafka consumer "),a("code",[e._v("poll()")]),e._v(" operation when using one of the manual "),a("a",{attrs:{href:"#committing-offsets"}},[e._v("commit methods")]),e._v("."),a("br"),e._v("Access to the "),a("code",[e._v("Consumer")]),e._v(" object is provided.")])])])]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("The "),a("code",[e._v("Consumer")]),e._v(" object is not thread-safe."),a("br"),e._v("You must only invoke its methods on the thread that calls the listener.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("You should not execute any "),a("code",[e._v("Consumer")]),e._v(" methods that affect the consumer’s positions and or committed offsets in your listener; the container needs to manage such information.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h5",{attrs:{id:"message-listener-containers"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#message-listener-containers"}},[e._v("#")]),e._v(" Message Listener Containers")]),e._v(" "),a("p",[e._v("Two "),a("code",[e._v("MessageListenerContainer")]),e._v(" implementations are provided:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("KafkaMessageListenerContainer")])])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ConcurrentMessageListenerContainer")])])])]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("KafkaMessageListenerContainer")]),e._v(" receives all message from all topics or partitions on a single thread.\nThe "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(" delegates to one or more "),a("code",[e._v("KafkaMessageListenerContainer")]),e._v(" instances to provide multi-threaded consumption.")]),e._v(" "),a("p",[e._v("Starting with version 2.2.7, you can add a "),a("code",[e._v("RecordInterceptor")]),e._v(" to the listener container; it will be invoked before calling the listener allowing inspection or modification of the record.\nIf the interceptor returns null, the listener is not called.\nStarting with version 2.7, it has additional methods which are called after the listener exits (normally, or by throwing an exception).\nAlso, starting with version 2.7, there is now a "),a("code",[e._v("BatchInterceptor")]),e._v(", providing similar functionality for "),a("a",{attrs:{href:"#batch-listeners"}},[e._v("Batch Listeners")]),e._v(".\nIn addition, the "),a("code",[e._v("ConsumerAwareRecordInterceptor")]),e._v(" (and "),a("code",[e._v("BatchInterceptor")]),e._v(") provide access to the "),a("code",[e._v("Consumer")]),e._v(".\nThis might be used, for example, to access the consumer metrics in the interceptor.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("You should not execute any methods that affect the consumer’s positions and or committed offsets in these interceptors; the container needs to manage such information.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("CompositeRecordInterceptor")]),e._v(" and "),a("code",[e._v("CompositeBatchInterceptor")]),e._v(" can be used to invoke multiple interceptors.")]),e._v(" "),a("p",[e._v("By default, starting with version 2.8, when using transactions, the interceptor is invoked before the transaction has started.\nYou can set the listener container’s "),a("code",[e._v("interceptBeforeTx")]),e._v(" property to "),a("code",[e._v("false")]),e._v(" to invoke the interceptor after the transaction has started instead.")]),e._v(" "),a("p",[e._v("Starting with versions 2.3.8, 2.4.6, the "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(" now supports "),a("a",{attrs:{href:"https://kafka.apache.org/documentation/#static_membership",target:"_blank",rel:"noopener noreferrer"}},[e._v("Static Membership"),a("OutboundLink")],1),e._v(" when the concurrency is greater than one.\nThe "),a("code",[e._v("group.instance.id")]),e._v(" is suffixed with "),a("code",[e._v("-n")]),e._v(" with "),a("code",[e._v("n")]),e._v(" starting at "),a("code",[e._v("1")]),e._v(".\nThis, together with an increased "),a("code",[e._v("session.timeout.ms")]),e._v(", can be used to reduce rebalance events, for example, when application instances are restarted.")]),e._v(" "),a("h6",{attrs:{id:"using-kafkamessagelistenercontainer"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#using-kafkamessagelistenercontainer"}},[e._v("#")]),e._v(" Using "),a("code",[e._v("KafkaMessageListenerContainer")])]),e._v(" "),a("p",[e._v("The following constructor is available:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("public KafkaMessageListenerContainer(ConsumerFactory consumerFactory,\n ContainerProperties containerProperties)\n")])])]),a("p",[e._v("It receives a "),a("code",[e._v("ConsumerFactory")]),e._v(" and information about topics and partitions, as well as other configuration, in a "),a("code",[e._v("ContainerProperties")]),e._v("object."),a("code",[e._v("ContainerProperties")]),e._v(" has the following constructors:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("public ContainerProperties(TopicPartitionOffset... topicPartitions)\n\npublic ContainerProperties(String... topics)\n\npublic ContainerProperties(Pattern topicPattern)\n")])])]),a("p",[e._v("The first constructor takes an array of "),a("code",[e._v("TopicPartitionOffset")]),e._v(" arguments to explicitly instruct the container about which partitions to use (using the consumer "),a("code",[e._v("assign()")]),e._v(" method) and with an optional initial offset.\nA positive value is an absolute offset by default.\nA negative value is relative to the current last offset within a partition by default.\nA constructor for "),a("code",[e._v("TopicPartitionOffset")]),e._v(" that takes an additional "),a("code",[e._v("boolean")]),e._v(" argument is provided.\nIf this is "),a("code",[e._v("true")]),e._v(", the initial offsets (positive or negative) are relative to the current position for this consumer.\nThe offsets are applied when the container is started.\nThe second takes an array of topics, and Kafka allocates the partitions based on the "),a("code",[e._v("group.id")]),e._v(" property — distributing partitions across the group.\nThe third uses a regex "),a("code",[e._v("Pattern")]),e._v(" to select the topics.")]),e._v(" "),a("p",[e._v("To assign a "),a("code",[e._v("MessageListener")]),e._v(" to a container, you can use the "),a("code",[e._v("ContainerProps.setMessageListener")]),e._v(" method when creating the Container.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");\ncontainerProps.setMessageListener(new MessageListener() {\n ...\n});\nDefaultKafkaConsumerFactory cf =\n new DefaultKafkaConsumerFactory<>(consumerProps());\nKafkaMessageListenerContainer container =\n new KafkaMessageListenerContainer<>(cf, containerProps);\nreturn container;\n')])])]),a("p",[e._v("Note that when creating a "),a("code",[e._v("DefaultKafkaConsumerFactory")]),e._v(", using the constructor that just takes in the properties as above means that key and value "),a("code",[e._v("Deserializer")]),e._v(" classes are picked up from configuration.\nAlternatively, "),a("code",[e._v("Deserializer")]),e._v(" instances may be passed to the "),a("code",[e._v("DefaultKafkaConsumerFactory")]),e._v(" constructor for key and/or value, in which case all Consumers share the same instances.\nAnother option is to provide "),a("code",[e._v("Supplier")]),e._v(" s (starting with version 2.3) that will be used to obtain separate "),a("code",[e._v("Deserializer")]),e._v(" instances for each "),a("code",[e._v("Consumer")]),e._v(":")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("DefaultKafkaConsumerFactory cf =\n new DefaultKafkaConsumerFactory<>(consumerProps(), null, () -> new CustomValueDeserializer());\nKafkaMessageListenerContainer container =\n new KafkaMessageListenerContainer<>(cf, containerProps);\nreturn container;\n")])])]),a("p",[e._v("Refer to the "),a("a",{attrs:{href:"https://docs.spring.io/spring-kafka/api/org/springframework/kafka/listener/ContainerProperties.html",target:"_blank",rel:"noopener noreferrer"}},[e._v("Javadoc"),a("OutboundLink")],1),e._v(" for "),a("code",[e._v("ContainerProperties")]),e._v(" for more information about the various properties that you can set.")]),e._v(" "),a("p",[e._v("Since version 2.1.1, a new property called "),a("code",[e._v("logContainerConfig")]),e._v(" is available.\nWhen "),a("code",[e._v("true")]),e._v(" and "),a("code",[e._v("INFO")]),e._v(" logging is enabled each listener container writes a log message summarizing its configuration properties.")]),e._v(" "),a("p",[e._v("By default, logging of topic offset commits is performed at the "),a("code",[e._v("DEBUG")]),e._v(" logging level.\nStarting with version 2.1.2, a property in "),a("code",[e._v("ContainerProperties")]),e._v(" called "),a("code",[e._v("commitLogLevel")]),e._v(" lets you specify the log level for these messages.\nFor example, to change the log level to "),a("code",[e._v("INFO")]),e._v(", you can use "),a("code",[e._v("containerProperties.setCommitLogLevel(LogIfLevelEnabled.Level.INFO);")]),e._v(".")]),e._v(" "),a("p",[e._v("Starting with version 2.2, a new container property called "),a("code",[e._v("missingTopicsFatal")]),e._v(" has been added (default: "),a("code",[e._v("false")]),e._v(" since 2.3.4).\nThis prevents the container from starting if any of the configured topics are not present on the broker.\nIt does not apply if the container is configured to listen to a topic pattern (regex).\nPreviously, the container threads looped within the "),a("code",[e._v("consumer.poll()")]),e._v(" method waiting for the topic to appear while logging many messages.\nAside from the logs, there was no indication that there was a problem.")]),e._v(" "),a("p",[e._v("As of version 2.8, a new container property "),a("code",[e._v("authExceptionRetryInterval")]),e._v(" has been introduced.\nThis causes the container to retry fetching messages after getting any "),a("code",[e._v("AuthenticationException")]),e._v(" or "),a("code",[e._v("AuthorizationException")]),e._v(" from the "),a("code",[e._v("KafkaConsumer")]),e._v(".\nThis can happen when, for example, the configured user is denied access to read a certain topic or credentials are incorrect.\nDefining "),a("code",[e._v("authExceptionRetryInterval")]),e._v(" allows the container to recover when proper permissions are granted.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("By default, no interval is configured - authentication and authorization errors are considered fatal, which causes the container to stop.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("Starting with version 2.8, when creating the consumer factory, if you provide deserializers as objects (in the constructor or via the setters), the factory will invoke the "),a("code",[e._v("configure()")]),e._v(" method to configure them with the configuration properties.")]),e._v(" "),a("h6",{attrs:{id:"using-concurrentmessagelistenercontainer"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#using-concurrentmessagelistenercontainer"}},[e._v("#")]),e._v(" Using "),a("code",[e._v("ConcurrentMessageListenerContainer")])]),e._v(" "),a("p",[e._v("The single constructor is similar to the "),a("code",[e._v("KafkaListenerContainer")]),e._v(" constructor.\nThe following listing shows the constructor’s signature:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("public ConcurrentMessageListenerContainer(ConsumerFactory consumerFactory,\n ContainerProperties containerProperties)\n")])])]),a("p",[e._v("It also has a "),a("code",[e._v("concurrency")]),e._v(" property.\nFor example, "),a("code",[e._v("container.setConcurrency(3)")]),e._v(" creates three "),a("code",[e._v("KafkaMessageListenerContainer")]),e._v(" instances.")]),e._v(" "),a("p",[e._v("For the first constructor, Kafka distributes the partitions across the consumers using its group management capabilities.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("When listening to multiple topics, the default partition distribution may not be what you expect."),a("br"),e._v("For example, if you have three topics with five partitions each and you want to use "),a("code",[e._v("concurrency=15")]),e._v(", you see only five active consumers, each assigned one partition from each topic, with the other 10 consumers being idle."),a("br"),e._v("This is because the default Kafka "),a("code",[e._v("PartitionAssignor")]),e._v(" is the "),a("code",[e._v("RangeAssignor")]),e._v(" (see its Javadoc)."),a("br"),e._v("For this scenario, you may want to consider using the "),a("code",[e._v("RoundRobinAssignor")]),e._v(" instead, which distributes the partitions across all of the consumers."),a("br"),e._v("Then, each consumer is assigned one topic or partition."),a("br"),e._v("To change the "),a("code",[e._v("PartitionAssignor")]),e._v(", you can set the "),a("code",[e._v("partition.assignment.strategy")]),e._v(" consumer property ("),a("code",[e._v("ConsumerConfigs.PARTITION_ASSIGNMENT_STRATEGY_CONFIG")]),e._v(") in the properties provided to the "),a("code",[e._v("DefaultKafkaConsumerFactory")]),e._v("."),a("br"),a("br"),e._v("When using Spring Boot, you can assign set the strategy as follows:"),a("br"),a("br"),a("code",[e._v("
spring.kafka.consumer.properties.partition.assignment.strategy=\\
org.apache.kafka.clients.consumer.RoundRobinAssignor
")])])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("When the container properties are configured with "),a("code",[e._v("TopicPartitionOffset")]),e._v(" s, the "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(" distributes the "),a("code",[e._v("TopicPartitionOffset")]),e._v(" instances across the delegate "),a("code",[e._v("KafkaMessageListenerContainer")]),e._v(" instances.")]),e._v(" "),a("p",[e._v("If, say, six "),a("code",[e._v("TopicPartitionOffset")]),e._v(" instances are provided and the "),a("code",[e._v("concurrency")]),e._v(" is "),a("code",[e._v("3")]),e._v("; each container gets two partitions.\nFor five "),a("code",[e._v("TopicPartitionOffset")]),e._v(" instances, two containers get two partitions, and the third gets one.\nIf the "),a("code",[e._v("concurrency")]),e._v(" is greater than the number of "),a("code",[e._v("TopicPartitions")]),e._v(", the "),a("code",[e._v("concurrency")]),e._v(" is adjusted down such that each container gets one partition.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("The "),a("code",[e._v("client.id")]),e._v(" property (if set) is appended with "),a("code",[e._v("-n")]),e._v(" where "),a("code",[e._v("n")]),e._v(" is the consumer instance that corresponds to the concurrency."),a("br"),e._v("This is required to provide unique names for MBeans when JMX is enabled.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("Starting with version 1.3, the "),a("code",[e._v("MessageListenerContainer")]),e._v(" provides access to the metrics of the underlying "),a("code",[e._v("KafkaConsumer")]),e._v(".\nIn the case of "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(", the "),a("code",[e._v("metrics()")]),e._v(" method returns the metrics for all the target "),a("code",[e._v("KafkaMessageListenerContainer")]),e._v(" instances.\nThe metrics are grouped into the "),a("code",[e._v("Map")]),e._v(" by the "),a("code",[e._v("client-id")]),e._v(" provided for the underlying "),a("code",[e._v("KafkaConsumer")]),e._v(".")]),e._v(" "),a("p",[e._v("Starting with version 2.3, the "),a("code",[e._v("ContainerProperties")]),e._v(" provides an "),a("code",[e._v("idleBetweenPolls")]),e._v(" option to let the main loop in the listener container to sleep between "),a("code",[e._v("KafkaConsumer.poll()")]),e._v(" calls.\nAn actual sleep interval is selected as the minimum from the provided option and difference between the "),a("code",[e._v("max.poll.interval.ms")]),e._v(" consumer config and the current records batch processing time.")]),e._v(" "),a("h6",{attrs:{id:"committing-offsets"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#committing-offsets"}},[e._v("#")]),e._v(" Committing Offsets")]),e._v(" "),a("p",[e._v("Several options are provided for committing offsets.\nIf the "),a("code",[e._v("enable.auto.commit")]),e._v(" consumer property is "),a("code",[e._v("true")]),e._v(", Kafka auto-commits the offsets according to its configuration.\nIf it is "),a("code",[e._v("false")]),e._v(", the containers support several "),a("code",[e._v("AckMode")]),e._v(" settings (described in the next list).\nThe default "),a("code",[e._v("AckMode")]),e._v(" is "),a("code",[e._v("BATCH")]),e._v(".\nStarting with version 2.3, the framework sets "),a("code",[e._v("enable.auto.commit")]),e._v(" to "),a("code",[e._v("false")]),e._v(" unless explicitly set in the configuration.\nPreviously, the Kafka default ("),a("code",[e._v("true")]),e._v(") was used if the property was not set.")]),e._v(" "),a("p",[e._v("The consumer "),a("code",[e._v("poll()")]),e._v(" method returns one or more "),a("code",[e._v("ConsumerRecords")]),e._v(".\nThe "),a("code",[e._v("MessageListener")]),e._v(" is called for each record.\nThe following lists describes the action taken by the container for each "),a("code",[e._v("AckMode")]),e._v(" (when transactions are not being used):")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("RECORD")]),e._v(": Commit the offset when the listener returns after processing the record.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("BATCH")]),e._v(": Commit the offset when all the records returned by the "),a("code",[e._v("poll()")]),e._v(" have been processed.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("TIME")]),e._v(": Commit the offset when all the records returned by the "),a("code",[e._v("poll()")]),e._v(" have been processed, as long as the "),a("code",[e._v("ackTime")]),e._v(" since the last commit has been exceeded.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("COUNT")]),e._v(": Commit the offset when all the records returned by the "),a("code",[e._v("poll()")]),e._v(" have been processed, as long as "),a("code",[e._v("ackCount")]),e._v(" records have been received since the last commit.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("COUNT_TIME")]),e._v(": Similar to "),a("code",[e._v("TIME")]),e._v(" and "),a("code",[e._v("COUNT")]),e._v(", but the commit is performed if either condition is "),a("code",[e._v("true")]),e._v(".")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("MANUAL")]),e._v(": The message listener is responsible to "),a("code",[e._v("acknowledge()")]),e._v(" the "),a("code",[e._v("Acknowledgment")]),e._v(".\nAfter that, the same semantics as "),a("code",[e._v("BATCH")]),e._v(" are applied.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("MANUAL_IMMEDIATE")]),e._v(": Commit the offset immediately when the "),a("code",[e._v("Acknowledgment.acknowledge()")]),e._v(" method is called by the listener.")])])]),e._v(" "),a("p",[e._v("When using "),a("a",{attrs:{href:"#transactions"}},[e._v("transactions")]),e._v(", the offset(s) are sent to the transaction and the semantics are equivalent to "),a("code",[e._v("RECORD")]),e._v(" or "),a("code",[e._v("BATCH")]),e._v(", depending on the listener type (record or batch).")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[a("code",[e._v("MANUAL")]),e._v(", and "),a("code",[e._v("MANUAL_IMMEDIATE")]),e._v(" require the listener to be an "),a("code",[e._v("AcknowledgingMessageListener")]),e._v(" or a "),a("code",[e._v("BatchAcknowledgingMessageListener")]),e._v("."),a("br"),e._v("See "),a("a",{attrs:{href:"#message-listeners"}},[e._v("Message Listeners")]),e._v(".")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("Depending on the "),a("code",[e._v("syncCommits")]),e._v(" container property, the "),a("code",[e._v("commitSync()")]),e._v(" or "),a("code",[e._v("commitAsync()")]),e._v(" method on the consumer is used."),a("code",[e._v("syncCommits")]),e._v(" is "),a("code",[e._v("true")]),e._v(" by default; also see "),a("code",[e._v("setSyncCommitTimeout")]),e._v(".\nSee "),a("code",[e._v("setCommitCallback")]),e._v(" to get the results of asynchronous commits; the default callback is the "),a("code",[e._v("LoggingCommitCallback")]),e._v(" which logs errors (and successes at debug level).")]),e._v(" "),a("p",[e._v("Because the listener container has it’s own mechanism for committing offsets, it prefers the Kafka "),a("code",[e._v("ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG")]),e._v(" to be "),a("code",[e._v("false")]),e._v(".\nStarting with version 2.3, it unconditionally sets it to false unless specifically set in the consumer factory or the container’s consumer property overrides.")]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("Acknowledgment")]),e._v(" has the following method:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("public interface Acknowledgment {\n\n void acknowledge();\n\n}\n")])])]),a("p",[e._v("This method gives the listener control over when offsets are committed.")]),e._v(" "),a("p",[e._v("Starting with version 2.3, the "),a("code",[e._v("Acknowledgment")]),e._v(" interface has two additional methods "),a("code",[e._v("nack(long sleep)")]),e._v(" and "),a("code",[e._v("nack(int index, long sleep)")]),e._v(".\nThe first one is used with a record listener, the second with a batch listener.\nCalling the wrong method for your listener type will throw an "),a("code",[e._v("IllegalStateException")]),e._v(".")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("If you want to commit a partial batch, using "),a("code",[e._v("nack()")]),e._v(", When using transactions, set the "),a("code",[e._v("AckMode")]),e._v(" to "),a("code",[e._v("MANUAL")]),e._v("; invoking "),a("code",[e._v("nack()")]),e._v(" will send the offsets of the successfully processed records to the transaction.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[a("code",[e._v("nack()")]),e._v(" can only be called on the consumer thread that invokes your listener.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("With a record listener, when "),a("code",[e._v("nack()")]),e._v(" is called, any pending offsets are committed, the remaing records from the last poll are discarded, and seeks are performed on their partitions so that the failed record and unprocessed records are redelivered on the next "),a("code",[e._v("poll()")]),e._v(".\nThe consumer thread can be paused before redelivery, by setting the "),a("code",[e._v("sleep")]),e._v(" argument.\nThis is similar functionality to throwing an exception when the container is configured with a "),a("code",[e._v("DefaultErrorHandler")]),e._v(".")]),e._v(" "),a("p",[e._v("When using a batch listener, you can specify the index within the batch where the failure occurred.\nWhen "),a("code",[e._v("nack()")]),e._v(" is called, offsets will be committed for records before the index and seeks are performed on the partitions for the failed and discarded records so that they will be redelivered on the next "),a("code",[e._v("poll()")]),e._v(".")]),e._v(" "),a("p",[e._v("See "),a("a",{attrs:{href:"#error-handlers"}},[e._v("Container Error Handlers")]),e._v(" for more information.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("When using partition assignment via group management, it is important to ensure the "),a("code",[e._v("sleep")]),e._v(" argument (plus the time spent processing records from the previous poll) is less than the consumer "),a("code",[e._v("max.poll.interval.ms")]),e._v(" property.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h6",{attrs:{id:"listener-container-auto-startup"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#listener-container-auto-startup"}},[e._v("#")]),e._v(" Listener Container Auto Startup")]),e._v(" "),a("p",[e._v("The listener containers implement "),a("code",[e._v("SmartLifecycle")]),e._v(", and "),a("code",[e._v("autoStartup")]),e._v(" is "),a("code",[e._v("true")]),e._v(" by default.\nThe containers are started in a late phase ("),a("code",[e._v("Integer.MAX-VALUE - 100")]),e._v(").\nOther components that implement "),a("code",[e._v("SmartLifecycle")]),e._v(", to handle data from listeners, should be started in an earlier phase.\nThe "),a("code",[e._v("- 100")]),e._v(" leaves room for later phases to enable components to be auto-started after the containers.")]),e._v(" "),a("h5",{attrs:{id:"manually-committing-offsets"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#manually-committing-offsets"}},[e._v("#")]),e._v(" Manually Committing Offsets")]),e._v(" "),a("p",[e._v("Normally, when using "),a("code",[e._v("AckMode.MANUAL")]),e._v(" or "),a("code",[e._v("AckMode.MANUAL_IMMEDIATE")]),e._v(", the acknowledgments must be acknowledged in order, because Kafka does not maintain state for each record, only a committed offset for each group/partition.\nStarting with version 2.8, you can now set the container property "),a("code",[e._v("asyncAcks")]),e._v(", which allows the acknowledgments for records returned by the poll to be acknowledged in any order.\nThe listener container will defer the out-of-order commits until the missing acknowledgments are received.\nThe consumer will be paused (no new records delivered) until all the offsets for the previous poll have been committed.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("While this feature allows applications to process records asynchronously, it should be understood that it increases the possibility of duplicate deliveries after a failure.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h5",{attrs:{id:"kafkalistener-annotation"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#kafkalistener-annotation"}},[e._v("#")]),e._v(" "),a("code",[e._v("@KafkaListener")]),e._v(" Annotation")]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("@KafkaListener")]),e._v(" annotation is used to designate a bean method as a listener for a listener container.\nThe bean is wrapped in a "),a("code",[e._v("MessagingMessageListenerAdapter")]),e._v(" configured with various features, such as converters to convert the data, if necessary, to match the method parameters.")]),e._v(" "),a("p",[e._v("You can configure most attributes on the annotation with SpEL by using "),a("code",[e._v("#{…}")]),e._v(" or property placeholders ("),a("code",[e._v("${…}")]),e._v(").\nSee the "),a("a",{attrs:{href:"https://docs.spring.io/spring-kafka/api/org/springframework/kafka/annotation/KafkaListener.html",target:"_blank",rel:"noopener noreferrer"}},[e._v("Javadoc"),a("OutboundLink")],1),e._v(" for more information.")]),e._v(" "),a("h6",{attrs:{id:"record-listeners"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#record-listeners"}},[e._v("#")]),e._v(" Record Listeners")]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("@KafkaListener")]),e._v(" annotation provides a mechanism for simple POJO listeners.\nThe following example shows how to use it:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('public class Listener {\n\n @KafkaListener(id = "foo", topics = "myTopic", clientIdPrefix = "myClientId")\n public void listen(String data) {\n ...\n }\n\n}\n')])])]),a("p",[e._v("This mechanism requires an "),a("code",[e._v("@EnableKafka")]),e._v(" annotation on one of your "),a("code",[e._v("@Configuration")]),e._v(" classes and a listener container factory, which is used to configure the underlying "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(".\nBy default, a bean with name "),a("code",[e._v("kafkaListenerContainerFactory")]),e._v(" is expected.\nThe following example shows how to use "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(":")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("@Configuration\n@EnableKafka\npublic class KafkaConfig {\n\n @Bean\n KafkaListenerContainerFactory>\n kafkaListenerContainerFactory() {\n ConcurrentKafkaListenerContainerFactory factory =\n new ConcurrentKafkaListenerContainerFactory<>();\n factory.setConsumerFactory(consumerFactory());\n factory.setConcurrency(3);\n factory.getContainerProperties().setPollTimeout(3000);\n return factory;\n }\n\n @Bean\n public ConsumerFactory consumerFactory() {\n return new DefaultKafkaConsumerFactory<>(consumerConfigs());\n }\n\n @Bean\n public Map consumerConfigs() {\n Map props = new HashMap<>();\n props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());\n ...\n return props;\n }\n}\n")])])]),a("p",[e._v("Notice that, to set container properties, you must use the "),a("code",[e._v("getContainerProperties()")]),e._v(" method on the factory.\nIt is used as a template for the actual properties injected into the container.")]),e._v(" "),a("p",[e._v("Starting with version 2.1.1, you can now set the "),a("code",[e._v("client.id")]),e._v(" property for consumers created by the annotation.\nThe "),a("code",[e._v("clientIdPrefix")]),e._v(" is suffixed with "),a("code",[e._v("-n")]),e._v(", where "),a("code",[e._v("n")]),e._v(" is an integer representing the container number when using concurrency.")]),e._v(" "),a("p",[e._v("Starting with version 2.2, you can now override the container factory’s "),a("code",[e._v("concurrency")]),e._v(" and "),a("code",[e._v("autoStartup")]),e._v(" properties by using properties on the annotation itself.\nThe properties can be simple values, property placeholders, or SpEL expressions.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "myListener", topics = "myTopic",\n autoStartup = "${listen.auto.start:true}", concurrency = "${listen.concurrency:3}")\npublic void listen(String data) {\n ...\n}\n')])])]),a("h6",{attrs:{id:"explicit-partition-assignment"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#explicit-partition-assignment"}},[e._v("#")]),e._v(" Explicit Partition Assignment")]),e._v(" "),a("p",[e._v("You can also configure POJO listeners with explicit topics and partitions (and, optionally, their initial offsets).\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "thing2", topicPartitions =\n { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),\n @TopicPartition(topic = "topic2", partitions = "0",\n partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))\n })\npublic void listen(ConsumerRecord record) {\n ...\n}\n')])])]),a("p",[e._v("You can specify each partition in the "),a("code",[e._v("partitions")]),e._v(" or "),a("code",[e._v("partitionOffsets")]),e._v(" attribute but not both.")]),e._v(" "),a("p",[e._v("As with most annotation properties, you can use SpEL expressions; for an example of how to generate a large list of partitions, see "),a("a",{attrs:{href:"#tip-assign-all-parts"}},[e._v("[tip-assign-all-parts]")]),e._v(".")]),e._v(" "),a("p",[e._v("Starting with version 2.5.5, you can apply an initial offset to all assigned partitions:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "thing3", topicPartitions =\n { @TopicPartition(topic = "topic1", partitions = { "0", "1" },\n partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0"))\n })\npublic void listen(ConsumerRecord record) {\n ...\n}\n')])])]),a("p",[e._v("The "),a("code",[e._v("*")]),e._v(" wildcard represents all partitions in the "),a("code",[e._v("partitions")]),e._v(" attribute.\nThere must only be one "),a("code",[e._v("@PartitionOffset")]),e._v(" with the wildcard in each "),a("code",[e._v("@TopicPartition")]),e._v(".")]),e._v(" "),a("p",[e._v("In addition, when the listener implements "),a("code",[e._v("ConsumerSeekAware")]),e._v(", "),a("code",[e._v("onPartitionsAssigned")]),e._v(" is now called, even when using manual assignment.\nThis allows, for example, any arbitrary seek operations at that time.")]),e._v(" "),a("p",[e._v("Starting with version 2.6.4, you can specify a comma-delimited list of partitions, or partition ranges:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "pp", autoStartup = "false",\n topicPartitions = @TopicPartition(topic = "topic1",\n partitions = "0-5, 7, 10-15"))\npublic void process(String in) {\n ...\n}\n')])])]),a("p",[e._v("The range is inclusive; the example above will assign partitions "),a("code",[e._v("0, 1, 2, 3, 4, 5, 7, 10, 11, 12, 13, 14, 15")]),e._v(".")]),e._v(" "),a("p",[e._v("The same technique can be used when specifying initial offsets:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "thing3", topicPartitions =\n { @TopicPartition(topic = "topic1",\n partitionOffsets = @PartitionOffset(partition = "0-5", initialOffset = "0"))\n })\npublic void listen(ConsumerRecord record) {\n ...\n}\n')])])]),a("p",[e._v("The initial offset will be applied to all 6 partitions.")]),e._v(" "),a("h6",{attrs:{id:"manual-acknowledgment"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#manual-acknowledgment"}},[e._v("#")]),e._v(" Manual Acknowledgment")]),e._v(" "),a("p",[e._v("When using manual "),a("code",[e._v("AckMode")]),e._v(", you can also provide the listener with the "),a("code",[e._v("Acknowledgment")]),e._v(".\nThe following example also shows how to use a different container factory.")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "cat", topics = "myTopic",\n containerFactory = "kafkaManualAckListenerContainerFactory")\npublic void listen(String data, Acknowledgment ack) {\n ...\n ack.acknowledge();\n}\n')])])]),a("h6",{attrs:{id:"consumer-record-metadata"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#consumer-record-metadata"}},[e._v("#")]),e._v(" Consumer Record Metadata")]),e._v(" "),a("p",[e._v("Finally, metadata about the record is available from message headers.\nYou can use the following header names to retrieve the headers of the message:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("KafkaHeaders.OFFSET")])])]),e._v(" "),a("li",[a("p",[a("code",[e._v("KafkaHeaders.RECEIVED_MESSAGE_KEY")])])]),e._v(" "),a("li",[a("p",[a("code",[e._v("KafkaHeaders.RECEIVED_TOPIC")])])]),e._v(" "),a("li",[a("p",[a("code",[e._v("KafkaHeaders.RECEIVED_PARTITION_ID")])])]),e._v(" "),a("li",[a("p",[a("code",[e._v("KafkaHeaders.RECEIVED_TIMESTAMP")])])]),e._v(" "),a("li",[a("p",[a("code",[e._v("KafkaHeaders.TIMESTAMP_TYPE")])])])]),e._v(" "),a("p",[e._v("Starting with version 2.5 the "),a("code",[e._v("RECEIVED_MESSAGE_KEY")]),e._v(" is not present if the incoming record has a "),a("code",[e._v("null")]),e._v(" key; previously the header was populated with a "),a("code",[e._v("null")]),e._v(" value.\nThis change is to make the framework consistent with "),a("code",[e._v("spring-messaging")]),e._v(" conventions where "),a("code",[e._v("null")]),e._v(" valued headers are not present.")]),e._v(" "),a("p",[e._v("The following example shows how to use the headers:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "qux", topicPattern = "myTopic1")\npublic void listen(@Payload String foo,\n @Header(name = KafkaHeaders.RECEIVED_MESSAGE_KEY, required = false) Integer key,\n @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,\n @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,\n @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts\n ) {\n ...\n}\n')])])]),a("p",[e._v("Starting with version 2.5, instead of using discrete headers, you can receive record metadata in a "),a("code",[e._v("ConsumerRecordMetadata")]),e._v(" parameter.")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("@KafkaListener(...)\npublic void listen(String str, ConsumerRecordMetadata meta) {\n ...\n}\n")])])]),a("p",[e._v("This contains all the data from the "),a("code",[e._v("ConsumerRecord")]),e._v(" except the key and value.")]),e._v(" "),a("h6",{attrs:{id:"batch-listeners"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#batch-listeners"}},[e._v("#")]),e._v(" Batch Listeners")]),e._v(" "),a("p",[e._v("Starting with version 1.1, you can configure "),a("code",[e._v("@KafkaListener")]),e._v(" methods to receive the entire batch of consumer records received from the consumer poll.\nTo configure the listener container factory to create batch listeners, you can set the "),a("code",[e._v("batchListener")]),e._v(" property.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("@Bean\npublic KafkaListenerContainerFactory batchFactory() {\n ConcurrentKafkaListenerContainerFactory factory =\n new ConcurrentKafkaListenerContainerFactory<>();\n factory.setConsumerFactory(consumerFactory());\n factory.setBatchListener(true); // <<<<<<<<<<<<<<<<<<<<<<<<<\n return factory;\n}\n")])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("Starting with version 2.8, you can override the factory’s "),a("code",[e._v("batchListener")]),e._v(" propery using the "),a("code",[e._v("batch")]),e._v(" property on the "),a("code",[e._v("@KafkaListener")]),e._v(" annotation."),a("br"),e._v("This, together with the changes to "),a("a",{attrs:{href:"#error-handlers"}},[e._v("Container Error Handlers")]),e._v(" allows the same factory to be used for both record and batch listeners.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("The following example shows how to receive a list of payloads:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")\npublic void listen(List list) {\n ...\n}\n')])])]),a("p",[e._v("The topic, partition, offset, and so on are available in headers that parallel the payloads.\nThe following example shows how to use the headers:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")\npublic void listen(List list,\n @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) List keys,\n @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List partitions,\n @Header(KafkaHeaders.RECEIVED_TOPIC) List topics,\n @Header(KafkaHeaders.OFFSET) List offsets) {\n ...\n}\n')])])]),a("p",[e._v("Alternatively, you can receive a "),a("code",[e._v("List")]),e._v(" of "),a("code",[e._v("Message")]),e._v(" objects with each offset and other details in each message, but it must be the only parameter (aside from optional "),a("code",[e._v("Acknowledgment")]),e._v(", when using manual commits, and/or "),a("code",[e._v("Consumer")]),e._v(" parameters) defined on the method.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "listMsg", topics = "myTopic", containerFactory = "batchFactory")\npublic void listen14(List> list) {\n ...\n}\n\n@KafkaListener(id = "listMsgAck", topics = "myTopic", containerFactory = "batchFactory")\npublic void listen15(List> list, Acknowledgment ack) {\n ...\n}\n\n@KafkaListener(id = "listMsgAckConsumer", topics = "myTopic", containerFactory = "batchFactory")\npublic void listen16(List> list, Acknowledgment ack, Consumer consumer) {\n ...\n}\n')])])]),a("p",[e._v("No conversion is performed on the payloads in this case.")]),e._v(" "),a("p",[e._v("If the "),a("code",[e._v("BatchMessagingMessageConverter")]),e._v(" is configured with a "),a("code",[e._v("RecordMessageConverter")]),e._v(", you can also add a generic type to the "),a("code",[e._v("Message")]),e._v(" parameter and the payloads are converted.\nSee "),a("a",{attrs:{href:"#payload-conversion-with-batch"}},[e._v("Payload Conversion with Batch Listeners")]),e._v(" for more information.")]),e._v(" "),a("p",[e._v("You can also receive a list of "),a("code",[e._v("ConsumerRecord")]),e._v(" objects, but it must be the only parameter (aside from optional "),a("code",[e._v("Acknowledgment")]),e._v(", when using manual commits and "),a("code",[e._v("Consumer")]),e._v(" parameters) defined on the method.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "listCRs", topics = "myTopic", containerFactory = "batchFactory")\npublic void listen(List> list) {\n ...\n}\n\n@KafkaListener(id = "listCRsAck", topics = "myTopic", containerFactory = "batchFactory")\npublic void listen(List> list, Acknowledgment ack) {\n ...\n}\n')])])]),a("p",[e._v("Starting with version 2.2, the listener can receive the complete "),a("code",[e._v("ConsumerRecords")]),e._v(" object returned by the "),a("code",[e._v("poll()")]),e._v(" method, letting the listener access additional methods, such as "),a("code",[e._v("partitions()")]),e._v(" (which returns the "),a("code",[e._v("TopicPartition")]),e._v(" instances in the list) and "),a("code",[e._v("records(TopicPartition)")]),e._v(" (which gets selective records).\nAgain, this must be the only parameter (aside from optional "),a("code",[e._v("Acknowledgment")]),e._v(", when using manual commits or "),a("code",[e._v("Consumer")]),e._v(" parameters) on the method.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "pollResults", topics = "myTopic", containerFactory = "batchFactory")\npublic void pollResults(ConsumerRecords records) {\n ...\n}\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("If the container factory has a "),a("code",[e._v("RecordFilterStrategy")]),e._v(" configured, it is ignored for "),a("code",[e._v("ConsumerRecords")]),e._v(" listeners, with a "),a("code",[e._v("WARN")]),e._v(" log message emitted."),a("br"),e._v("Records can only be filtered with a batch listener if the "),a("code",[e._v(">")]),e._v(" form of listener is used."),a("br"),e._v("By default, records are filtered one-at-a-time; starting with version 2.8, you can override "),a("code",[e._v("filterBatch")]),e._v(" to filter the entire batch in one call.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h6",{attrs:{id:"annotation-properties"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#annotation-properties"}},[e._v("#")]),e._v(" Annotation Properties")]),e._v(" "),a("p",[e._v("Starting with version 2.0, the "),a("code",[e._v("id")]),e._v(" property (if present) is used as the Kafka consumer "),a("code",[e._v("group.id")]),e._v(" property, overriding the configured property in the consumer factory, if present.\nYou can also set "),a("code",[e._v("groupId")]),e._v(" explicitly or set "),a("code",[e._v("idIsGroup")]),e._v(" to false to restore the previous behavior of using the consumer factory "),a("code",[e._v("group.id")]),e._v(".")]),e._v(" "),a("p",[e._v("You can use property placeholders or SpEL expressions within most annotation properties, as the following example shows:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(topics = "${some.property}")\n\n@KafkaListener(topics = "#{someBean.someProperty}",\n groupId = "#{someBean.someProperty}.group")\n')])])]),a("p",[e._v("Starting with version 2.1.2, the SpEL expressions support a special token: "),a("code",[e._v("__listener")]),e._v(".\nIt is a pseudo bean name that represents the current bean instance within which this annotation exists.")]),e._v(" "),a("p",[e._v("Consider the following example:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Bean\npublic Listener listener1() {\n return new Listener("topic1");\n}\n\n@Bean\npublic Listener listener2() {\n return new Listener("topic2");\n}\n')])])]),a("p",[e._v("Given the beans in the previous example, we can then use the following:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('public class Listener {\n\n private final String topic;\n\n public Listener(String topic) {\n this.topic = topic;\n }\n\n @KafkaListener(topics = "#{__listener.topic}",\n groupId = "#{__listener.topic}.group")\n public void listen(...) {\n ...\n }\n\n public String getTopic() {\n return this.topic;\n }\n\n}\n')])])]),a("p",[e._v("If, in the unlikely event that you have an actual bean called "),a("code",[e._v("__listener")]),e._v(", you can change the expression token byusing the "),a("code",[e._v("beanRef")]),e._v(" attribute.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(beanRef = "__x", topics = "#{__x.topic}",\n groupId = "#{__x.topic}.group")\n')])])]),a("p",[e._v("Starting with version 2.2.4, you can specify Kafka consumer properties directly on the annotation, these will override any properties with the same name configured in the consumer factory. You "),a("strong",[e._v("cannot")]),e._v(" specify the "),a("code",[e._v("group.id")]),e._v(" and "),a("code",[e._v("client.id")]),e._v(" properties this way; they will be ignored; use the "),a("code",[e._v("groupId")]),e._v(" and "),a("code",[e._v("clientIdPrefix")]),e._v(" annotation properties for those.")]),e._v(" "),a("p",[e._v("The properties are specified as individual strings with the normal Java "),a("code",[e._v("Properties")]),e._v(" file format: "),a("code",[e._v("foo:bar")]),e._v(", "),a("code",[e._v("foo=bar")]),e._v(", or "),a("code",[e._v("foo bar")]),e._v(".")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(topics = "myTopic", groupId = "group", properties = {\n "max.poll.interval.ms:60000",\n ConsumerConfig.MAX_POLL_RECORDS_CONFIG + "=100"\n})\n')])])]),a("p",[e._v("The following is an example of the corresponding listeners for the example in "),a("a",{attrs:{href:"#routing-template"}},[e._v("Using "),a("code",[e._v("RoutingKafkaTemplate")])]),e._v(".")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "one", topics = "one")\npublic void listen1(String in) {\n System.out.println("1: " + in);\n}\n\n@KafkaListener(id = "two", topics = "two",\n properties = "value.deserializer:org.apache.kafka.common.serialization.ByteArrayDeserializer")\npublic void listen2(byte[] in) {\n System.out.println("2: " + new String(in));\n}\n')])])]),a("h5",{attrs:{id:"obtaining-the-consumer-group-id"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#obtaining-the-consumer-group-id"}},[e._v("#")]),e._v(" Obtaining the Consumer "),a("code",[e._v("group.id")])]),e._v(" "),a("p",[e._v("When running the same listener code in multiple containers, it may be useful to be able to determine which container (identified by its "),a("code",[e._v("group.id")]),e._v(" consumer property) that a record came from.")]),e._v(" "),a("p",[e._v("You can call "),a("code",[e._v("KafkaUtils.getConsumerGroupId()")]),e._v(" on the listener thread to do this.\nAlternatively, you can access the group id in a method parameter.")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "bar", topicPattern = "${topicTwo:annotated2}", exposeGroupId = "${always:true}")\npublic void listener(@Payload String foo,\n @Header(KafkaHeaders.GROUP_ID) String groupId) {\n...\n}\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("This is available in record listeners and batch listeners that receive a "),a("code",[e._v("List")]),e._v(" of records."),a("br"),e._v("It is "),a("strong",[e._v("not")]),e._v(" available in a batch listener that receives a "),a("code",[e._v("ConsumerRecords")]),e._v(" argument."),a("br"),e._v("Use the "),a("code",[e._v("KafkaUtils")]),e._v(" mechanism in that case.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h5",{attrs:{id:"container-thread-naming"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#container-thread-naming"}},[e._v("#")]),e._v(" Container Thread Naming")]),e._v(" "),a("p",[e._v("Listener containers currently use two task executors, one to invoke the consumer and another that is used to invoke the listener when the kafka consumer property "),a("code",[e._v("enable.auto.commit")]),e._v(" is "),a("code",[e._v("false")]),e._v(".\nYou can provide custom executors by setting the "),a("code",[e._v("consumerExecutor")]),e._v(" and "),a("code",[e._v("listenerExecutor")]),e._v(" properties of the container’s "),a("code",[e._v("ContainerProperties")]),e._v(".\nWhen using pooled executors, be sure that enough threads are available to handle the concurrency across all the containers in which they are used.\nWhen using the "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(", a thread from each is used for each consumer ("),a("code",[e._v("concurrency")]),e._v(").")]),e._v(" "),a("p",[e._v("If you do not provide a consumer executor, a "),a("code",[e._v("SimpleAsyncTaskExecutor")]),e._v(" is used.\nThis executor creates threads with names similar to "),a("code",[e._v("-C-1")]),e._v(" (consumer thread).\nFor the "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(", the "),a("code",[e._v("")]),e._v(" part of the thread name becomes "),a("code",[e._v("-m")]),e._v(", where "),a("code",[e._v("m")]),e._v(" represents the consumer instance."),a("code",[e._v("n")]),e._v(" increments each time the container is started.\nSo, with a bean name of "),a("code",[e._v("container")]),e._v(", threads in this container will be named "),a("code",[e._v("container-0-C-1")]),e._v(", "),a("code",[e._v("container-1-C-1")]),e._v(" etc., after the container is started the first time; "),a("code",[e._v("container-0-C-2")]),e._v(", "),a("code",[e._v("container-1-C-2")]),e._v(" etc., after a stop and subsequent start.")]),e._v(" "),a("h5",{attrs:{id:"kafkalistener-as-a-meta-annotation"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#kafkalistener-as-a-meta-annotation"}},[e._v("#")]),e._v(" "),a("code",[e._v("@KafkaListener")]),e._v(" as a Meta Annotation")]),e._v(" "),a("p",[e._v("Starting with version 2.2, you can now use "),a("code",[e._v("@KafkaListener")]),e._v(" as a meta annotation.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Target(ElementType.METHOD)\n@Retention(RetentionPolicy.RUNTIME)\n@KafkaListener\npublic @interface MyThreeConsumersListener {\n\n @AliasFor(annotation = KafkaListener.class, attribute = "id")\n String id();\n\n @AliasFor(annotation = KafkaListener.class, attribute = "topics")\n String[] topics();\n\n @AliasFor(annotation = KafkaListener.class, attribute = "concurrency")\n String concurrency() default "3";\n\n}\n')])])]),a("p",[e._v("You must alias at least one of "),a("code",[e._v("topics")]),e._v(", "),a("code",[e._v("topicPattern")]),e._v(", or "),a("code",[e._v("topicPartitions")]),e._v(" (and, usually, "),a("code",[e._v("id")]),e._v(" or "),a("code",[e._v("groupId")]),e._v(" unless you have specified a "),a("code",[e._v("group.id")]),e._v(" in the consumer factory configuration).\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@MyThreeConsumersListener(id = "my.group", topics = "my.topic")\npublic void listen1(String in) {\n ...\n}\n')])])]),a("h5",{attrs:{id:"kafkalistener-on-a-class"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#kafkalistener-on-a-class"}},[e._v("#")]),e._v(" "),a("code",[e._v("@KafkaListener")]),e._v(" on a Class")]),e._v(" "),a("p",[e._v("When you use "),a("code",[e._v("@KafkaListener")]),e._v(" at the class-level, you must specify "),a("code",[e._v("@KafkaHandler")]),e._v(" at the method level.\nWhen messages are delivered, the converted message payload type is used to determine which method to call.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "multi", topics = "myTopic")\nstatic class MultiListenerBean {\n\n @KafkaHandler\n public void listen(String foo) {\n ...\n }\n\n @KafkaHandler\n public void listen(Integer bar) {\n ...\n }\n\n @KafkaHandler(isDefault = true)\n public void listenDefault(Object object) {\n ...\n }\n\n}\n')])])]),a("p",[e._v("Starting with version 2.1.3, you can designate a "),a("code",[e._v("@KafkaHandler")]),e._v(" method as the default method that is invoked if there is no match on other methods.\nAt most, one method can be so designated.\nWhen using "),a("code",[e._v("@KafkaHandler")]),e._v(" methods, the payload must have already been converted to the domain object (so the match can be performed).\nUse a custom deserializer, the "),a("code",[e._v("JsonDeserializer")]),e._v(", or the "),a("code",[e._v("JsonMessageConverter")]),e._v(" with its "),a("code",[e._v("TypePrecedence")]),e._v(" set to "),a("code",[e._v("TYPE_ID")]),e._v(".\nSee "),a("a",{attrs:{href:"#serdes"}},[e._v("Serialization, Deserialization, and Message Conversion")]),e._v(" for more information.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("Due to some limitations in the way Spring resolves method arguments, a default "),a("code",[e._v("@KafkaHandler")]),e._v(" cannot receive discrete headers; it must use the "),a("code",[e._v("ConsumerRecordMetadata")]),e._v(" as discussed in "),a("a",{attrs:{href:"#consumer-record-metadata"}},[e._v("Consumer Record Metadata")]),e._v(".")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("For example:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("@KafkaHandler(isDefault = true)\npublic void listenDefault(Object object, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {\n ...\n}\n")])])]),a("p",[e._v("This won’t work if the object is a "),a("code",[e._v("String")]),e._v("; the "),a("code",[e._v("topic")]),e._v(" parameter will also get a reference to "),a("code",[e._v("object")]),e._v(".")]),e._v(" "),a("p",[e._v("If you need metadata about the record in a default method, use this:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("@KafkaHandler(isDefault = true)\nvoid listen(Object in, @Header(KafkaHeaders.RECORD_METADATA) ConsumerRecordMetadata meta) {\n String topic = meta.topic();\n ...\n}\n")])])]),a("h5",{attrs:{id:"kafkalistener-attribute-modification"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#kafkalistener-attribute-modification"}},[e._v("#")]),e._v(" "),a("code",[e._v("@KafkaListener")]),e._v(" Attribute Modification")]),e._v(" "),a("p",[e._v("Starting with version 2.7.2, you can now programmatically modify annotation attributes before the container is created.\nTo do so, add one or more "),a("code",[e._v("KafkaListenerAnnotationBeanPostProcessor.AnnotationEnhancer")]),e._v(" to the application context."),a("code",[e._v("AnnotationEnhancer")]),e._v(" is a "),a("code",[e._v("BiFunction, AnnotatedElement, Map")]),e._v(" and must return a map of attributes.\nThe attribute values can contain SpEL and/or property placeholders; the enhancer is called before any resolution is performed.\nIf more than one enhancer is present, and they implement "),a("code",[e._v("Ordered")]),e._v(", they will be invoked in order.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[a("code",[e._v("AnnotationEnhancer")]),e._v(" bean definitions must be declared "),a("code",[e._v("static")]),e._v(" because they are required very early in the application context’s lifecycle.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("An example follows:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Bean\npublic static AnnotationEnhancer groupIdEnhancer() {\n return (attrs, element) -> {\n attrs.put("groupId", attrs.get("id") + "." + (element instanceof Class\n ? ((Class) element).getSimpleName()\n : ((Method) element).getDeclaringClass().getSimpleName()\n + "." + ((Method) element).getName()));\n return attrs;\n };\n}\n')])])]),a("h5",{attrs:{id:"kafkalistener-lifecycle-management"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#kafkalistener-lifecycle-management"}},[e._v("#")]),e._v(" "),a("code",[e._v("@KafkaListener")]),e._v(" Lifecycle Management")]),e._v(" "),a("p",[e._v("The listener containers created for "),a("code",[e._v("@KafkaListener")]),e._v(" annotations are not beans in the application context.\nInstead, they are registered with an infrastructure bean of type "),a("code",[e._v("KafkaListenerEndpointRegistry")]),e._v(".\nThis bean is automatically declared by the framework and manages the containers' lifecycles; it will auto-start any containers that have "),a("code",[e._v("autoStartup")]),e._v(" set to "),a("code",[e._v("true")]),e._v(".\nAll containers created by all container factories must be in the same "),a("code",[e._v("phase")]),e._v(".\nSee "),a("a",{attrs:{href:"#container-auto-startup"}},[e._v("Listener Container Auto Startup")]),e._v(" for more information.\nYou can manage the lifecycle programmatically by using the registry.\nStarting or stopping the registry will start or stop all the registered containers.\nAlternatively, you can get a reference to an individual container by using its "),a("code",[e._v("id")]),e._v(" attribute.\nYou can set "),a("code",[e._v("autoStartup")]),e._v(" on the annotation, which overrides the default setting configured into the container factory.\nYou can get a reference to the bean from the application context, such as auto-wiring, to manage its registered containers.\nThe following examples show how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "myContainer", topics = "myTopic", autoStartup = "false")\npublic void listen(...) { ... }\n')])])]),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Autowired\nprivate KafkaListenerEndpointRegistry registry;\n\n...\n\n this.registry.getListenerContainer("myContainer").start();\n\n...\n')])])]),a("p",[e._v("The registry only maintains the life cycle of containers it manages; containers declared as beans are not managed by the registry and can be obtained from the application context.\nA collection of managed containers can be obtained by calling the registry’s "),a("code",[e._v("getListenerContainers()")]),e._v(" method.\nVersion 2.2.5 added a convenience method "),a("code",[e._v("getAllListenerContainers()")]),e._v(", which returns a collection of all containers, including those managed by the registry and those declared as beans.\nThe collection returned will include any prototype beans that have been initialized, but it will not initialize any lazy bean declarations.")]),e._v(" "),a("h5",{attrs:{id:"kafkalistener-payload-validation"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#kafkalistener-payload-validation"}},[e._v("#")]),e._v(" "),a("code",[e._v("@KafkaListener")]),e._v(" "),a("code",[e._v("@Payload")]),e._v(" Validation")]),e._v(" "),a("p",[e._v("Starting with version 2.2, it is now easier to add a "),a("code",[e._v("Validator")]),e._v(" to validate "),a("code",[e._v("@KafkaListener")]),e._v(" "),a("code",[e._v("@Payload")]),e._v(" arguments.\nPreviously, you had to configure a custom "),a("code",[e._v("DefaultMessageHandlerMethodFactory")]),e._v(" and add it to the registrar.\nNow, you can add the validator to the registrar itself.\nThe following code shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("@Configuration\n@EnableKafka\npublic class Config implements KafkaListenerConfigurer {\n\n ...\n\n @Override\n public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {\n registrar.setValidator(new MyValidator());\n }\n\n}\n")])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("When you use Spring Boot with the validation starter, a "),a("code",[e._v("LocalValidatorFactoryBean")]),e._v(" is auto-configured, as the following example shows:")])])]),e._v(" "),a("tbody")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("@Configuration\n@EnableKafka\npublic class Config implements KafkaListenerConfigurer {\n\n @Autowired\n private LocalValidatorFactoryBean validator;\n ...\n\n @Override\n public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {\n registrar.setValidator(this.validator);\n }\n}\n")])])]),a("p",[e._v("The following examples show how to validate:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("public static class ValidatedClass {\n\n @Max(10)\n private int bar;\n\n public int getBar() {\n return this.bar;\n }\n\n public void setBar(int bar) {\n this.bar = bar;\n }\n\n}\n")])])]),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id="validated", topics = "annotated35", errorHandler = "validationErrorHandler",\n containerFactory = "kafkaJsonListenerContainerFactory")\npublic void validatedListener(@Payload @Valid ValidatedClass val) {\n ...\n}\n\n@Bean\npublic KafkaListenerErrorHandler validationErrorHandler() {\n return (m, e) -> {\n ...\n };\n}\n')])])]),a("p",[e._v("Starting with version 2.5.11, validation now works on payloads for "),a("code",[e._v("@KafkaHandler")]),e._v(" methods in a class-level listener.\nSee "),a("a",{attrs:{href:"#class-level-kafkalistener"}},[a("code",[e._v("@KafkaListener")]),e._v(" on a Class")]),e._v(".")]),e._v(" "),a("h5",{attrs:{id:"rebalancing-listeners"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#rebalancing-listeners"}},[e._v("#")]),e._v(" Rebalancing Listeners")]),e._v(" "),a("p",[a("code",[e._v("ContainerProperties")]),e._v(" has a property called "),a("code",[e._v("consumerRebalanceListener")]),e._v(", which takes an implementation of the Kafka client’s "),a("code",[e._v("ConsumerRebalanceListener")]),e._v(" interface.\nIf this property is not provided, the container configures a logging listener that logs rebalance events at the "),a("code",[e._v("INFO")]),e._v(" level.\nThe framework also adds a sub-interface "),a("code",[e._v("ConsumerAwareRebalanceListener")]),e._v(".\nThe following listing shows the "),a("code",[e._v("ConsumerAwareRebalanceListener")]),e._v(" interface definition:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListener {\n\n void onPartitionsRevokedBeforeCommit(Consumer consumer, Collection partitions);\n\n void onPartitionsRevokedAfterCommit(Consumer consumer, Collection partitions);\n\n void onPartitionsAssigned(Consumer consumer, Collection partitions);\n\n void onPartitionsLost(Consumer consumer, Collection partitions);\n\n}\n")])])]),a("p",[e._v("Notice that there are two callbacks when partitions are revoked.\nThe first is called immediately.\nThe second is called after any pending offsets are committed.\nThis is useful if you wish to maintain offsets in some external repository, as the following example shows:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("containerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {\n\n @Override\n public void onPartitionsRevokedBeforeCommit(Consumer consumer, Collection partitions) {\n // acknowledge any pending Acknowledgments (if using manual acks)\n }\n\n @Override\n public void onPartitionsRevokedAfterCommit(Consumer consumer, Collection partitions) {\n // ...\n store(consumer.position(partition));\n // ...\n }\n\n @Override\n public void onPartitionsAssigned(Collection partitions) {\n // ...\n consumer.seek(partition, offsetTracker.getOffset() + 1);\n // ...\n }\n});\n")])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("Starting with version 2.4, a new method "),a("code",[e._v("onPartitionsLost()")]),e._v(" has been added (similar to a method with the same name in "),a("code",[e._v("ConsumerRebalanceLister")]),e._v(")."),a("br"),e._v("The default implementation on "),a("code",[e._v("ConsumerRebalanceLister")]),e._v(" simply calls "),a("code",[e._v("onPartionsRevoked")]),e._v("."),a("br"),e._v("The default implementation on "),a("code",[e._v("ConsumerAwareRebalanceListener")]),e._v(" does nothing."),a("br"),e._v("When supplying the listener container with a custom listener (of either type), it is important that your implementation not call "),a("code",[e._v("onPartitionsRevoked")]),e._v(" from "),a("code",[e._v("onPartitionsLost")]),e._v("."),a("br"),e._v("If you implement "),a("code",[e._v("ConsumerRebalanceListener")]),e._v(" you should override the default method."),a("br"),e._v("This is because the listener container will call its own "),a("code",[e._v("onPartitionsRevoked")]),e._v(" from its implementation of "),a("code",[e._v("onPartitionsLost")]),e._v(" after calling the method on your implementation."),a("br"),e._v("If you implementation delegates to the default behavior, "),a("code",[e._v("onPartitionsRevoked")]),e._v(" will be called twice each time the "),a("code",[e._v("Consumer")]),e._v(" calls that method on the container’s listener.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h5",{attrs:{id:"forwarding-listener-results-using-sendto"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#forwarding-listener-results-using-sendto"}},[e._v("#")]),e._v(" Forwarding Listener Results using "),a("code",[e._v("@SendTo")])]),e._v(" "),a("p",[e._v("Starting with version 2.0, if you also annotate a "),a("code",[e._v("@KafkaListener")]),e._v(" with a "),a("code",[e._v("@SendTo")]),e._v(" annotation and the method invocation returns a result, the result is forwarded to the topic specified by the "),a("code",[e._v("@SendTo")]),e._v(".")]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("@SendTo")]),e._v(" value can have several forms:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v('@SendTo("someTopic")')]),e._v(" routes to the literal topic")])]),e._v(" "),a("li",[a("p",[a("code",[e._v('@SendTo("#{someExpression}")')]),e._v(" routes to the topic determined by evaluating the expression once during application context initialization.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v('@SendTo("!{someExpression}")')]),e._v(" routes to the topic determined by evaluating the expression at runtime.\nThe "),a("code",[e._v("#root")]),e._v(" object for the evaluation has three properties:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("request")]),e._v(": The inbound "),a("code",[e._v("ConsumerRecord")]),e._v(" (or "),a("code",[e._v("ConsumerRecords")]),e._v(" object for a batch listener))")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("source")]),e._v(": The "),a("code",[e._v("org.springframework.messaging.Message")]),e._v(" converted from the "),a("code",[e._v("request")]),e._v(".")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("result")]),e._v(": The method return result.")])])])]),e._v(" "),a("li",[a("p",[a("code",[e._v("@SendTo")]),e._v(" (no properties): This is treated as "),a("code",[e._v("!{source.headers['kafka_replyTopic']}")]),e._v(" (since version 2.1.3).")])])]),e._v(" "),a("p",[e._v("Starting with versions 2.1.11 and 2.2.1, property placeholders are resolved within "),a("code",[e._v("@SendTo")]),e._v(" values.")]),e._v(" "),a("p",[e._v("The result of the expression evaluation must be a "),a("code",[e._v("String")]),e._v(" that represents the topic name.\nThe following examples show the various ways to use "),a("code",[e._v("@SendTo")]),e._v(":")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(topics = "annotated21")\n@SendTo("!{request.value()}") // runtime SpEL\npublic String replyingListener(String in) {\n ...\n}\n\n@KafkaListener(topics = "${some.property:annotated22}")\n@SendTo("#{myBean.replyTopic}") // config time SpEL\npublic Collection replyingBatchListener(List in) {\n ...\n}\n\n@KafkaListener(topics = "annotated23", errorHandler = "replyErrorHandler")\n@SendTo("annotated23reply") // static reply topic definition\npublic String replyingListenerWithErrorHandler(String in) {\n ...\n}\n...\n@KafkaListener(topics = "annotated25")\n@SendTo("annotated25reply1")\npublic class MultiListenerSendTo {\n\n @KafkaHandler\n public String foo(String in) {\n ...\n }\n\n @KafkaHandler\n @SendTo("!{\'annotated25reply2\'}")\n public String bar(@Payload(required = false) KafkaNull nul,\n @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) int key) {\n ...\n }\n\n}\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("In order to support "),a("code",[e._v("@SendTo")]),e._v(", the listener container factory must be provided with a "),a("code",[e._v("KafkaTemplate")]),e._v(" (in its "),a("code",[e._v("replyTemplate")]),e._v(" property), which is used to send the reply."),a("br"),e._v("This should be a "),a("code",[e._v("KafkaTemplate")]),e._v(" and not a "),a("code",[e._v("ReplyingKafkaTemplate")]),e._v(" which is used on the client-side for request/reply processing."),a("br"),e._v("When using Spring Boot, boot will auto-configure the template into the factory; when configuring your own factory, it must be set as shown in the examples below.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("Starting with version 2.2, you can add a "),a("code",[e._v("ReplyHeadersConfigurer")]),e._v(" to the listener container factory.\nThis is consulted to determine which headers you want to set in the reply message.\nThe following example shows how to add a "),a("code",[e._v("ReplyHeadersConfigurer")]),e._v(":")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Bean\npublic ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {\n ConcurrentKafkaListenerContainerFactory factory =\n new ConcurrentKafkaListenerContainerFactory<>();\n factory.setConsumerFactory(cf());\n factory.setReplyTemplate(template());\n factory.setReplyHeadersConfigurer((k, v) -> k.equals("cat"));\n return factory;\n}\n')])])]),a("p",[e._v("You can also add more headers if you wish.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Bean\npublic ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {\n ConcurrentKafkaListenerContainerFactory factory =\n new ConcurrentKafkaListenerContainerFactory<>();\n factory.setConsumerFactory(cf());\n factory.setReplyTemplate(template());\n factory.setReplyHeadersConfigurer(new ReplyHeadersConfigurer() {\n\n @Override\n public boolean shouldCopy(String headerName, Object headerValue) {\n return false;\n }\n\n @Override\n public Map additionalHeaders() {\n return Collections.singletonMap("qux", "fiz");\n }\n\n });\n return factory;\n}\n')])])]),a("p",[e._v("When you use "),a("code",[e._v("@SendTo")]),e._v(", you must configure the "),a("code",[e._v("ConcurrentKafkaListenerContainerFactory")]),e._v(" with a "),a("code",[e._v("KafkaTemplate")]),e._v(" in its "),a("code",[e._v("replyTemplate")]),e._v(" property to perform the send.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("Unless you use "),a("a",{attrs:{href:"#replying-template"}},[e._v("request/reply semantics")]),e._v(" only the simple "),a("code",[e._v("send(topic, value)")]),e._v(" method is used, so you may wish to create a subclass to generate the partition or key."),a("br"),e._v("The following example shows how to do so:")])])]),e._v(" "),a("tbody")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("@Bean\npublic KafkaTemplate myReplyingTemplate() {\n return new KafkaTemplate(producerFactory()) {\n\n @Override\n public ListenableFuture> send(String topic, String data) {\n return super.send(topic, partitionForData(data), keyForData(data), data);\n }\n\n ...\n\n };\n}\n")])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("If the listener method returns "),a("code",[e._v("Message")]),e._v(" or "),a("code",[e._v("Collection>")]),e._v(", the listener method is responsible for setting up the message headers for the reply."),a("br"),e._v("For example, when handling a request from a "),a("code",[e._v("ReplyingKafkaTemplate")]),e._v(", you might do the following:"),a("br"),a("br"),a("code",[e._v('
@KafkaListener(id = "messageReturned", topics = "someTopic")
public Message listen(String in, @Header(KafkaHeaders.REPLY_TOPIC) byte[] replyTo,
@Header(KafkaHeaders.CORRELATION_ID) byte[] correlation) {
return MessageBuilder.withPayload(in.toUpperCase())
.setHeader(KafkaHeaders.TOPIC, replyTo)
.setHeader(KafkaHeaders.MESSAGE_KEY, 42)
.setHeader(KafkaHeaders.CORRELATION_ID, correlation)
.setHeader("someOtherHeader", "someValue")
.build();
}
')])])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("When using request/reply semantics, the target partition can be requested by the sender.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("You can annotate a "),a("code",[e._v("@KafkaListener")]),e._v(" method with "),a("code",[e._v("@SendTo")]),e._v(" even if no result is returned."),a("br"),e._v("This is to allow the configuration of an "),a("code",[e._v("errorHandler")]),e._v(" that can forward information about a failed message delivery to some topic."),a("br"),e._v("The following example shows how to do so:"),a("br"),a("br"),a("code",[e._v('
@KafkaListener(id = "voidListenerWithReplyingErrorHandler", topics = "someTopic",
errorHandler = "voidSendToErrorHandler")
@SendTo("failures")
public void voidListenerWithReplyingErrorHandler(String in) {
throw new RuntimeException("fail");
}
@Bean
public KafkaListenerErrorHandler voidSendToErrorHandler() {
return (m, e) -> {
return ... // some information about the failure and input data
};
}
')]),a("br"),a("br"),e._v("See "),a("a",{attrs:{href:"#annotation-error-handling"}},[e._v("Handling Exceptions")]),e._v(" for more information.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("If a listener method returns an "),a("code",[e._v("Iterable")]),e._v(", by default a record for each element as the value is sent."),a("br"),e._v("Starting with version 2.3.5, set the "),a("code",[e._v("splitIterables")]),e._v(" property on "),a("code",[e._v("@KafkaListener")]),e._v(" to "),a("code",[e._v("false")]),e._v(" and the entire result will be sent as the value of a single "),a("code",[e._v("ProducerRecord")]),e._v("."),a("br"),e._v("This requires a suitable serializer in the reply template’s producer configuration."),a("br"),e._v("However, if the reply is "),a("code",[e._v("Iterable>")]),e._v(" the property is ignored and each message is sent separately.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h5",{attrs:{id:"filtering-messages"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#filtering-messages"}},[e._v("#")]),e._v(" Filtering Messages")]),e._v(" "),a("p",[e._v("In certain scenarios, such as rebalancing, a message that has already been processed may be redelivered.\nThe framework cannot know whether such a message has been processed or not.\nThat is an application-level function.\nThis is known as the "),a("a",{attrs:{href:"https://www.enterpriseintegrationpatterns.com/patterns/messaging/IdempotentReceiver.html",target:"_blank",rel:"noopener noreferrer"}},[e._v("Idempotent Receiver"),a("OutboundLink")],1),e._v(" pattern and Spring Integration provides an "),a("a",{attrs:{href:"https://docs.spring.io/spring-integration/reference/html/#idempotent-receiver",target:"_blank",rel:"noopener noreferrer"}},[e._v("implementation of it"),a("OutboundLink")],1),e._v(".")]),e._v(" "),a("p",[e._v("The Spring for Apache Kafka project also provides some assistance by means of the "),a("code",[e._v("FilteringMessageListenerAdapter")]),e._v(" class, which can wrap your "),a("code",[e._v("MessageListener")]),e._v(".\nThis class takes an implementation of "),a("code",[e._v("RecordFilterStrategy")]),e._v(" in which you implement the "),a("code",[e._v("filter")]),e._v(" method to signal that a message is a duplicate and should be discarded.\nThis has an additional property called "),a("code",[e._v("ackDiscarded")]),e._v(", which indicates whether the adapter should acknowledge the discarded record.\nIt is "),a("code",[e._v("false")]),e._v(" by default.")]),e._v(" "),a("p",[e._v("When you use "),a("code",[e._v("@KafkaListener")]),e._v(", set the "),a("code",[e._v("RecordFilterStrategy")]),e._v(" (and optionally "),a("code",[e._v("ackDiscarded")]),e._v(") on the container factory so that the listener is wrapped in the appropriate filtering adapter.")]),e._v(" "),a("p",[e._v("In addition, a "),a("code",[e._v("FilteringBatchMessageListenerAdapter")]),e._v(" is provided, for when you use a batch "),a("a",{attrs:{href:"#message-listeners"}},[e._v("message listener")]),e._v(".")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("The "),a("code",[e._v("FilteringBatchMessageListenerAdapter")]),e._v(" is ignored if your "),a("code",[e._v("@KafkaListener")]),e._v(" receives a "),a("code",[e._v("ConsumerRecords")]),e._v(" instead of "),a("code",[e._v("List>")]),e._v(", because "),a("code",[e._v("ConsumerRecords")]),e._v(" is immutable.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h5",{attrs:{id:"retrying-deliveries"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#retrying-deliveries"}},[e._v("#")]),e._v(" Retrying Deliveries")]),e._v(" "),a("p",[e._v("See the "),a("code",[e._v("DefaultErrorHandler")]),e._v(" in "),a("a",{attrs:{href:"#annotation-error-handling"}},[e._v("Handling Exceptions")]),e._v(".")]),e._v(" "),a("h5",{attrs:{id:"starting-kafkalistener-s-in-sequence"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#starting-kafkalistener-s-in-sequence"}},[e._v("#")]),e._v(" Starting "),a("code",[e._v("@KafkaListener")]),e._v(" s in Sequence")]),e._v(" "),a("p",[e._v("A common use case is to start a listener after another listener has consumed all the records in a topic.\nFor example, you may want to load the contents of one or more compacted topics into memory before processing records from other topics.\nStarting with version 2.7.3, a new component "),a("code",[e._v("ContainerGroupSequencer")]),e._v(" has been introduced.\nIt uses the "),a("code",[e._v("@KafkaListener")]),e._v(" "),a("code",[e._v("containerGroup")]),e._v(" property to group containers together and start the containers in the next group, when all the containers in the current group have gone idle.")]),e._v(" "),a("p",[e._v("It is best illustrated with an example.")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@KafkaListener(id = "listen1", topics = "topic1", containerGroup = "g1", concurrency = "2")\npublic void listen1(String in) {\n}\n\n@KafkaListener(id = "listen2", topics = "topic2", containerGroup = "g1", concurrency = "2")\npublic void listen2(String in) {\n}\n\n@KafkaListener(id = "listen3", topics = "topic3", containerGroup = "g2", concurrency = "2")\npublic void listen3(String in) {\n}\n\n@KafkaListener(id = "listen4", topics = "topic4", containerGroup = "g2", concurrency = "2")\npublic void listen4(String in) {\n}\n\n@Bean\nContainerGroupSequencer sequencer(KafkaListenerEndpointRegistry registry) {\n return new ContainerGroupSequencer(registry, 5000, "g1", "g2");\n}\n')])])]),a("p",[e._v("Here, we have 4 listeners in two groups, "),a("code",[e._v("g1")]),e._v(" and "),a("code",[e._v("g2")]),e._v(".")]),e._v(" "),a("p",[e._v("During application context initialization, the sequencer, sets the "),a("code",[e._v("autoStartup")]),e._v(" property of all the containers in the provided groups to "),a("code",[e._v("false")]),e._v(".\nIt also sets the "),a("code",[e._v("idleEventInterval")]),e._v(" for any containers (that do not already have one set) to the supplied value (5000ms in this case).\nThen, when the sequencer is started by the application context, the containers in the first group are started.\nAs "),a("code",[e._v("ListenerContainerIdleEvent")]),e._v(" s are received, each individual child container in each container is stopped.\nWhen all child containers in a "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(" are stopped, the parent container is stopped.\nWhen all containers in a group have been stopped, the containers in the next group are started.\nThere is no limit to the number of groups or containers in a group.")]),e._v(" "),a("p",[e._v("By default, the containers in the final group ("),a("code",[e._v("g2")]),e._v(" above) are not stopped when they go idle.\nTo modify that behavior, set "),a("code",[e._v("stopLastGroupWhenIdle")]),e._v(" to "),a("code",[e._v("true")]),e._v(" on the sequencer.")]),e._v(" "),a("p",[e._v("As an aside; previously, containers in each group were added to a bean of type "),a("code",[e._v("Collection")]),e._v(" with the bean name being the "),a("code",[e._v("containerGroup")]),e._v(".\nThese collections are now deprecated in favor of beans of type "),a("code",[e._v("ContainerGroup")]),e._v(" with a bean name that is the group name, suffixed with "),a("code",[e._v(".group")]),e._v("; in the example above, there would be 2 beans "),a("code",[e._v("g1.group")]),e._v(" and "),a("code",[e._v("g2.group")]),e._v(".\nThe "),a("code",[e._v("Collection")]),e._v(" beans will be removed in a future release.")]),e._v(" "),a("h5",{attrs:{id:"using-kafkatemplate-to-receive"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#using-kafkatemplate-to-receive"}},[e._v("#")]),e._v(" Using "),a("code",[e._v("KafkaTemplate")]),e._v(" to Receive")]),e._v(" "),a("p",[e._v("This section covers how to use "),a("code",[e._v("KafkaTemplate")]),e._v(" to receive messages.")]),e._v(" "),a("p",[e._v("Starting with version 2.8, the template has four "),a("code",[e._v("receive()")]),e._v(" methods:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("ConsumerRecord receive(String topic, int partition, long offset);\n\nConsumerRecord receive(String topic, int partition, long offset, Duration pollTimeout);\n\nConsumerRecords receive(Collection requested);\n\nConsumerRecords receive(Collection requested, Duration pollTimeout);\n")])])]),a("p",[e._v("As you can see, you need to know the partition and offset of the record(s) you need to retrieve; a new "),a("code",[e._v("Consumer")]),e._v(" is created (and closed) for each operation.")]),e._v(" "),a("p",[e._v("With the last two methods, each record is retrieved individually and the results assembled into a "),a("code",[e._v("ConsumerRecords")]),e._v(" object.\nWhen creating the "),a("code",[e._v("TopicPartitionOffset")]),e._v(" s for the request, only positive, absolute offsets are supported.")]),e._v(" "),a("h4",{attrs:{id:"_4-1-5-listener-container-properties"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-5-listener-container-properties"}},[e._v("#")]),e._v(" 4.1.5. Listener Container Properties")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th",[e._v("Property")]),e._v(" "),a("th",[e._v("Default")]),e._v(" "),a("th",[e._v("Description")])])]),e._v(" "),a("tbody",[a("tr",[a("td"),e._v(" "),a("td",[e._v("1")]),e._v(" "),a("td",[e._v("The number of records before committing pending offsets when the "),a("code",[e._v("ackMode")]),e._v(" is "),a("code",[e._v("COUNT")]),e._v(" or "),a("code",[e._v("COUNT_TIME")]),e._v(".")])]),e._v(" "),a("tr",[a("td",[e._v("wrapping the message listener, invoked in order.")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td",[e._v("`]")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("5000")]),e._v(" "),a("td",[e._v("The time in milliseconds after which pending offsets are committed when the "),a("code",[e._v("ackMode")]),e._v(" is "),a("code",[e._v("TIME")]),e._v(" or "),a("code",[e._v("COUNT_TIME")]),e._v(".")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("LATEST_ONLY _NO_TX")]),e._v(" "),a("td",[e._v("Whether or not to commit the initial position on assignment; by default, the initial offset will only be committed if the "),a("code",[e._v("ConsumerConfig.AUTO_OFFSET_RESET_CONFIG")]),e._v(" is "),a("code",[e._v("latest")]),e._v(" and it won’t run in a transaction even if there is a transaction manager present."),a("br"),e._v("See the javadocs for "),a("code",[e._v("ContainerProperties.AssignmentCommitOption")]),e._v(" for more information about the available options.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("null")])]),e._v(" "),a("td",[e._v("When not null, a "),a("code",[e._v("Duration")]),e._v(" to sleep between polls when an "),a("code",[e._v("AuthenticationException")]),e._v(" or "),a("code",[e._v("AuthorizationException")]),e._v(" is thrown by the Kafka client."),a("br"),e._v("When null, such exceptions are considered fatal and the container will stop.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("A prefix for the "),a("code",[e._v("client.id")]),e._v(" consumer property."),a("br"),e._v("Overrides the consumer factory "),a("code",[e._v("client.id")]),e._v(" property; in a concurrent container, "),a("code",[e._v("-n")]),e._v(" is added as a suffix for each consumer instance.")]),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("false")]),e._v(" "),a("td",[e._v("Set to "),a("code",[e._v("true")]),e._v(" to always check for a "),a("code",[e._v("DeserializationException")]),e._v(" header when a "),a("code",[e._v("null")]),e._v(" "),a("code",[e._v("key")]),e._v(" is received."),a("br"),e._v("Useful when the consumer code cannot determine that an "),a("code",[e._v("ErrorHandlingDeserializer")]),e._v(" has been configured, such as when using a delegating deserializer.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("false")]),e._v(" "),a("td",[e._v("Set to "),a("code",[e._v("true")]),e._v(" to always check for a "),a("code",[e._v("DeserializationException")]),e._v(" header when a "),a("code",[e._v("null")]),e._v(" "),a("code",[e._v("value")]),e._v(" is received."),a("br"),e._v("Useful when the consumer code cannot determine that an "),a("code",[e._v("ErrorHandlingDeserializer")]),e._v(" has been configured, such as when using a delegating deserializer.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("null")])]),e._v(" "),a("td",[e._v("When present and "),a("code",[e._v("syncCommits")]),e._v(" is "),a("code",[e._v("false")]),e._v(" a callback invoked after the commit completes.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("DEBUG")]),e._v(" "),a("td",[e._v("The logging level for logs pertaining to committing offsets.")])]),e._v(" "),a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("30s")]),e._v(" "),a("td",[e._v("The time to wait for the consumer to start before logging an error; this might happen if, say, you use a task executor with insufficient threads.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("SimpleAsyncTaskExecutor")])]),e._v(" "),a("td",[e._v("A task executor to run the consumer threads."),a("br"),e._v("The default executor creates threads named "),a("code",[e._v("-C-n")]),e._v("; with the "),a("code",[e._v("KafkaMessageListenerContainer")]),e._v(", the name is the bean name; with the "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(" the name is the bean name suffixed with "),a("code",[e._v("-n")]),e._v(" where n is incremented for each child container.")])]),e._v(" "),a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td",[e._v("for more information.")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("null")])]),e._v(" "),a("td",[e._v("Overrides the consumer "),a("code",[e._v("group.id")]),e._v(" property; automatically set by the "),a("code",[e._v("@KafkaListener")]),e._v(" "),a("code",[e._v("id")]),e._v(" or "),a("code",[e._v("groupId")]),e._v(" property.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("5.0")]),e._v(" "),a("td",[e._v("Multiplier for "),a("code",[e._v("idleEventInterval")]),e._v(" that is applied before any records are received."),a("br"),e._v("After a record is received, the multiplier is no longer applied."),a("br"),e._v("Available since version 2.8.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("0")]),e._v(" "),a("td",[e._v("Used to slow down deliveries by sleeping the thread between polls."),a("br"),e._v("The time to process a batch of records plus this value must be less than the "),a("code",[e._v("max.poll.interval.ms")]),e._v(" consumer property.")])]),e._v(" "),a("tr",[a("td",[e._v("."),a("br"),e._v("Also see "),a("code",[e._v("idleBeforeDataMultiplier")]),e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("None")]),e._v(" "),a("td",[e._v("Used to override any arbitrary consumer properties configured on the consumer factory.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("false")])]),e._v(" "),a("td",[e._v("Set to true to log at INFO level all container properties.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("null")])]),e._v(" "),a("td",[e._v("The message listener.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("true")])]),e._v(" "),a("td",[e._v("Whether or not to maintain Micrometer timers for the consumer threads.")])]),e._v(" "),a("tr",[a("td",[e._v("are not present on the broker.")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("30s")]),e._v(" "),a("td",[e._v("How often to check the state of the consumer threads for "),a("code",[e._v("NonResponsiveConsumerEvent")]),e._v(" s."),a("br"),e._v("See "),a("code",[e._v("noPollThreshold")]),e._v(" and "),a("code",[e._v("pollTimeout")]),e._v(".")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("3.0")]),e._v(" "),a("td",[e._v("Multiplied by "),a("code",[e._v("pollTimeOut")]),e._v(" to determine whether to publish a "),a("code",[e._v("NonResponsiveConsumerEvent")]),e._v("."),a("br"),e._v("See "),a("code",[e._v("monitorInterval")]),e._v(".")])]),e._v(" "),a("tr",[a("td",[e._v("`.")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td",[e._v("`.")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("ThreadPoolTaskScheduler")])]),e._v(" "),a("td",[e._v("A scheduler on which to run the consumer monitor task.")])]),e._v(" "),a("tr",[a("td",[e._v("` method until all consumers stop and before publishing the container stopped event.")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td",[e._v("for more information.")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("false")])]),e._v(" "),a("td",[e._v("When the container is stopped, stop processing after the current record instead of after processing all the records from the previous poll.")])]),e._v(" "),a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("null")])]),e._v(" "),a("td",[e._v("The timeout to use when "),a("code",[e._v("syncCommits")]),e._v(" is "),a("code",[e._v("true")]),e._v("."),a("br"),e._v("When not set, the container will attempt to determine the "),a("code",[e._v("default.api.timeout.ms")]),e._v(" consumer property and use that; otherwise it will use 60 seconds.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("true")])]),e._v(" "),a("td",[e._v("Whether to use sync or async commits for offsets; see "),a("code",[e._v("commitCallback")]),e._v(".")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("n/a")]),e._v(" "),a("td",[e._v("The configured topics, topic pattern or explicitly assigned topics/partitions."),a("br"),e._v("Mutually exclusive; at least one must be provided; enforced by "),a("code",[e._v("ContainerProperties")]),e._v(" constructors.")])]),e._v(" "),a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")])])]),e._v(" "),a("table",[a("thead",[a("tr",[a("th",[e._v("Property")]),e._v(" "),a("th",[e._v("Default")]),e._v(" "),a("th",[e._v("Description")])])]),e._v(" "),a("tbody",[a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("DefaultAfterRollbackProcessor")])]),e._v(" "),a("td",[e._v("An "),a("code",[e._v("AfterRollbackProcessor")]),e._v(" to invoke after a transaction is rolled back.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("application context")]),e._v(" "),a("td",[e._v("The event publisher.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("See desc.")]),e._v(" "),a("td",[e._v("Deprecated - see "),a("code",[e._v("commonErrorHandler")]),e._v(".")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("null")])]),e._v(" "),a("td",[e._v("Set a "),a("code",[e._v("BatchInterceptor")]),e._v(" to call before invoking the batch listener; does not apply to record listeners."),a("br"),e._v("Also see "),a("code",[e._v("interceptBeforeTx")]),e._v(".")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("bean name")]),e._v(" "),a("td",[e._v("The bean name of the container; suffixed with "),a("code",[e._v("-n")]),e._v(" for child containers.")])]),e._v(" "),a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("ContainerProperties")])]),e._v(" "),a("td",[e._v("The container properties instance.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("See desc.")]),e._v(" "),a("td",[e._v("Deprecated - see "),a("code",[e._v("commonErrorHandler")]),e._v(".")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("See desc.")]),e._v(" "),a("td",[e._v("Deprecated - see "),a("code",[e._v("commonErrorHandler")]),e._v(".")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("See desc.")]),e._v(" "),a("td",[e._v("The "),a("code",[e._v("containerProperties.groupId")]),e._v(", if present, otherwise the "),a("code",[e._v("group.id")]),e._v(" property from the consumer factory.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("true")])]),e._v(" "),a("td",[e._v("Determines whether the "),a("code",[e._v("recordInterceptor")]),e._v(" is called before or after a transaction starts.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("See desc.")]),e._v(" "),a("td",[e._v("The bean name for user-configured containers or the "),a("code",[e._v("id")]),e._v(" attribute of "),a("code",[e._v("@KafkaListener")]),e._v(" s.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("True if a consumer pause has been requested.")]),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("null")])]),e._v(" "),a("td",[e._v("Set a "),a("code",[e._v("RecordInterceptor")]),e._v(" to call before invoking the record listener; does not apply to batch listeners."),a("br"),e._v("Also see "),a("code",[e._v("interceptBeforeTx")]),e._v(".")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("30s")]),e._v(" "),a("td",[e._v("When the "),a("code",[e._v("missingTopicsFatal")]),e._v(" container property is "),a("code",[e._v("true")]),e._v(", how long to wait, in seconds, for the "),a("code",[e._v("describeTopics")]),e._v(" operation to complete.")])])])]),e._v(" "),a("table",[a("thead",[a("tr",[a("th",[e._v("Property")]),e._v(" "),a("th",[e._v("Default")]),e._v(" "),a("th",[e._v("Description")])])]),e._v(" "),a("tbody",[a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("null")])]),e._v(" "),a("td",[e._v("Used by the concurrent container to give each child container’s consumer a unique "),a("code",[e._v("client.id")]),e._v(".")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("n/a")]),e._v(" "),a("td",[e._v("True if pause has been requested and the consumer has actually paused.")])])])]),e._v(" "),a("table",[a("thead",[a("tr",[a("th",[e._v("Property")]),e._v(" "),a("th",[e._v("Default")]),e._v(" "),a("th",[e._v("Description")])])]),e._v(" "),a("tbody",[a("tr",[a("td"),e._v(" "),a("td",[a("code",[e._v("true")])]),e._v(" "),a("td",[e._v("Set to false to suppress adding a suffix to the "),a("code",[e._v("client.id")]),e._v(" consumer property, when the "),a("code",[e._v("concurrency")]),e._v(" is only 1.")])]),e._v(" "),a("tr",[a("td",[e._v(".")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td",[e._v(", keyed by the child container’s consumer’s "),a("code",[e._v("client.id")]),e._v(" property.")]),e._v(" "),a("td"),e._v(" "),a("td")]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("1")]),e._v(" "),a("td",[e._v("The number of child "),a("code",[e._v("KafkaMessageListenerContainer")]),e._v(" s to manage.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("n/a")]),e._v(" "),a("td",[e._v("True if pause has been requested and all child containers' consumer has actually paused.")])]),e._v(" "),a("tr",[a("td"),e._v(" "),a("td",[e._v("n/a")]),e._v(" "),a("td",[e._v("A reference to all child "),a("code",[e._v("KafkaMessageListenerContainer")]),e._v(" s.")])])])]),e._v(" "),a("h4",{attrs:{id:"_4-1-6-application-events"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-6-application-events"}},[e._v("#")]),e._v(" 4.1.6. Application Events")]),e._v(" "),a("p",[e._v("The following Spring application events are published by listener containers and their consumers:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("ConsumerStartingEvent")]),e._v(" - published when a consumer thread is first started, before it starts polling.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ConsumerStartedEvent")]),e._v(" - published when a consumer is about to start polling.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ConsumerFailedToStartEvent")]),e._v(" - published if no "),a("code",[e._v("ConsumerStartingEvent")]),e._v(" is published within the "),a("code",[e._v("consumerStartTimeout")]),e._v(" container property.\nThis event might signal that the configured task executor has insufficient threads to support the containers it is used in and their concurrency.\nAn error message is also logged when this condition occurs.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ListenerContainerIdleEvent")]),e._v(": published when no messages have been received in "),a("code",[e._v("idleInterval")]),e._v(" (if configured).")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ListenerContainerNoLongerIdleEvent")]),e._v(": published when a record is consumed after previously publishing a "),a("code",[e._v("ListenerContainerIdleEvent")]),e._v(".")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ListenerContainerPartitionIdleEvent")]),e._v(": published when no messages have been received from that partition in "),a("code",[e._v("idlePartitionEventInterval")]),e._v(" (if configured).")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ListenerContainerPartitionNoLongerIdleEvent")]),e._v(": published when a record is consumed from a partition that has previously published a "),a("code",[e._v("ListenerContainerPartitionIdleEvent")]),e._v(".")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("NonResponsiveConsumerEvent")]),e._v(": published when the consumer appears to be blocked in the "),a("code",[e._v("poll")]),e._v(" method.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ConsumerPartitionPausedEvent")]),e._v(": published by each consumer when a partition is paused.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ConsumerPartitionResumedEvent")]),e._v(": published by each consumer when a partition is resumed.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ConsumerPausedEvent")]),e._v(": published by each consumer when the container is paused.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ConsumerResumedEvent")]),e._v(": published by each consumer when the container is resumed.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ConsumerStoppingEvent")]),e._v(": published by each consumer just before stopping.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ConsumerStoppedEvent")]),e._v(": published after the consumer is closed.\nSee "),a("a",{attrs:{href:"#thread-safety"}},[e._v("Thread Safety")]),e._v(".")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ContainerStoppedEvent")]),e._v(": published when all consumers have stopped.")])])]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("By default, the application context’s event multicaster invokes event listeners on the calling thread."),a("br"),e._v("If you change the multicaster to use an async executor, you must not invoke any "),a("code",[e._v("Consumer")]),e._v(" methods when the event contains a reference to the consumer.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("ListenerContainerIdleEvent")]),e._v(" has the following properties:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("source")]),e._v(": The listener container instance that published the event.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("container")]),e._v(": The listener container or the parent listener container, if the source container is a child.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("id")]),e._v(": The listener ID (or container bean name).")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("idleTime")]),e._v(": The time the container had been idle when the event was published.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("topicPartitions")]),e._v(": The topics and partitions that the container was assigned at the time the event was generated.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("consumer")]),e._v(": A reference to the Kafka "),a("code",[e._v("Consumer")]),e._v(" object.\nFor example, if the consumer’s "),a("code",[e._v("pause()")]),e._v(" method was previously called, it can "),a("code",[e._v("resume()")]),e._v(" when the event is received.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("paused")]),e._v(": Whether the container is currently paused.\nSee "),a("a",{attrs:{href:"#pause-resume"}},[e._v("Pausing and Resuming Listener Containers")]),e._v(" for more information.")])])]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("ListenerContainerNoLongerIdleEvent")]),e._v(" has the same properties, except "),a("code",[e._v("idleTime")]),e._v(" and "),a("code",[e._v("paused")]),e._v(".")]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("ListenerContainerPartitionIdleEvent")]),e._v(" has the following properties:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("source")]),e._v(": The listener container instance that published the event.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("container")]),e._v(": The listener container or the parent listener container, if the source container is a child.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("id")]),e._v(": The listener ID (or container bean name).")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("idleTime")]),e._v(": The time partition consumption had been idle when the event was published.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("topicPartition")]),e._v(": The topic and partition that triggered the event.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("consumer")]),e._v(": A reference to the Kafka "),a("code",[e._v("Consumer")]),e._v(" object.\nFor example, if the consumer’s "),a("code",[e._v("pause()")]),e._v(" method was previously called, it can "),a("code",[e._v("resume()")]),e._v(" when the event is received.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("paused")]),e._v(": Whether that partition consumption is currently paused for that consumer.\nSee "),a("a",{attrs:{href:"#pause-resume"}},[e._v("Pausing and Resuming Listener Containers")]),e._v(" for more information.")])])]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("ListenerContainerPartitionNoLongerIdleEvent")]),e._v(" has the same properties, except "),a("code",[e._v("idleTime")]),e._v(" and "),a("code",[e._v("paused")]),e._v(".")]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("NonResponsiveConsumerEvent")]),e._v(" has the following properties:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("source")]),e._v(": The listener container instance that published the event.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("container")]),e._v(": The listener container or the parent listener container, if the source container is a child.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("id")]),e._v(": The listener ID (or container bean name).")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("timeSinceLastPoll")]),e._v(": The time just before the container last called "),a("code",[e._v("poll()")]),e._v(".")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("topicPartitions")]),e._v(": The topics and partitions that the container was assigned at the time the event was generated.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("consumer")]),e._v(": A reference to the Kafka "),a("code",[e._v("Consumer")]),e._v(" object.\nFor example, if the consumer’s "),a("code",[e._v("pause()")]),e._v(" method was previously called, it can "),a("code",[e._v("resume()")]),e._v(" when the event is received.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("paused")]),e._v(": Whether the container is currently paused.\nSee "),a("a",{attrs:{href:"#pause-resume"}},[e._v("Pausing and Resuming Listener Containers")]),e._v(" for more information.")])])]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("ConsumerPausedEvent")]),e._v(", "),a("code",[e._v("ConsumerResumedEvent")]),e._v(", and "),a("code",[e._v("ConsumerStopping")]),e._v(" events have the following properties:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("source")]),e._v(": The listener container instance that published the event.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("container")]),e._v(": The listener container or the parent listener container, if the source container is a child.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("partitions")]),e._v(": The "),a("code",[e._v("TopicPartition")]),e._v(" instances involved.")])])]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("ConsumerPartitionPausedEvent")]),e._v(", "),a("code",[e._v("ConsumerPartitionResumedEvent")]),e._v(" events have the following properties:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("source")]),e._v(": The listener container instance that published the event.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("container")]),e._v(": The listener container or the parent listener container, if the source container is a child.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("partition")]),e._v(": The "),a("code",[e._v("TopicPartition")]),e._v(" instance involved.")])])]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("ConsumerStartingEvent")]),e._v(", "),a("code",[e._v("ConsumerStartingEvent")]),e._v(", "),a("code",[e._v("ConsumerFailedToStartEvent")]),e._v(", "),a("code",[e._v("ConsumerStoppedEvent")]),e._v(" and "),a("code",[e._v("ContainerStoppedEvent")]),e._v(" events have the following properties:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("source")]),e._v(": The listener container instance that published the event.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("container")]),e._v(": The listener container or the parent listener container, if the source container is a child.")])])]),e._v(" "),a("p",[e._v("All containers (whether a child or a parent) publish "),a("code",[e._v("ContainerStoppedEvent")]),e._v(".\nFor a parent container, the source and container properties are identical.")]),e._v(" "),a("p",[e._v("In addition, the "),a("code",[e._v("ConsumerStoppedEvent")]),e._v(" has the following additional property:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("reason")])]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("NORMAL")]),e._v(" - the consumer stopped normally (container was stopped).")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("ERROR")]),e._v(" - a "),a("code",[e._v("java.lang.Error")]),e._v(" was thrown.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("FENCED")]),e._v(" - the transactional producer was fenced and the "),a("code",[e._v("stopContainerWhenFenced")]),e._v(" container property is "),a("code",[e._v("true")]),e._v(".")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("AUTH")]),e._v(" - an "),a("code",[e._v("AuthenticationException")]),e._v(" or "),a("code",[e._v("AuthorizationException")]),e._v(" was thrown and the "),a("code",[e._v("authExceptionRetryInterval")]),e._v(" is not configured.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("NO_OFFSET")]),e._v(" - there is no offset for a partition and the "),a("code",[e._v("auto.offset.reset")]),e._v(" policy is "),a("code",[e._v("none")]),e._v(".")])])])])]),e._v(" "),a("p",[e._v("You can use this event to restart the container after such a condition:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("if (event.getReason.equals(Reason.FENCED)) {\n event.getSource(MessageListenerContainer.class).start();\n}\n")])])]),a("h5",{attrs:{id:"detecting-idle-and-non-responsive-consumers"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#detecting-idle-and-non-responsive-consumers"}},[e._v("#")]),e._v(" Detecting Idle and Non-Responsive Consumers")]),e._v(" "),a("p",[e._v("While efficient, one problem with asynchronous consumers is detecting when they are idle.\nYou might want to take some action if no messages arrive for some period of time.")]),e._v(" "),a("p",[e._v("You can configure the listener container to publish a "),a("code",[e._v("ListenerContainerIdleEvent")]),e._v(" when some time passes with no message delivery.\nWhile the container is idle, an event is published every "),a("code",[e._v("idleEventInterval")]),e._v(" milliseconds.")]),e._v(" "),a("p",[e._v("To configure this feature, set the "),a("code",[e._v("idleEventInterval")]),e._v(" on the container.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Bean\npublic KafkaMessageListenerContainer(ConsumerFactory consumerFactory) {\n ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");\n ...\n containerProps.setIdleEventInterval(60000L);\n ...\n KafkaMessageListenerContainer container = new KafKaMessageListenerContainer<>(...);\n return container;\n}\n')])])]),a("p",[e._v("The following example shows how to set the "),a("code",[e._v("idleEventInterval")]),e._v(" for a "),a("code",[e._v("@KafkaListener")]),e._v(":")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("@Bean\npublic ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {\n ConcurrentKafkaListenerContainerFactory factory =\n new ConcurrentKafkaListenerContainerFactory<>();\n ...\n factory.getContainerProperties().setIdleEventInterval(60000L);\n ...\n return factory;\n}\n")])])]),a("p",[e._v("In each of these cases, an event is published once per minute while the container is idle.")]),e._v(" "),a("p",[e._v("If, for some reason, the consumer "),a("code",[e._v("poll()")]),e._v(" method does not exit, no messages are received and idle events cannot be generated (this was a problem with early versions of the "),a("code",[e._v("kafka-clients")]),e._v(" when the broker wasn’t reachable).\nIn this case, the container publishes a "),a("code",[e._v("NonResponsiveConsumerEvent")]),e._v(" if a poll does not return within "),a("code",[e._v("3x")]),e._v(" the "),a("code",[e._v("pollTimeout")]),e._v(" property.\nBy default, this check is performed once every 30 seconds in each container.\nYou can modify this behavior by setting the "),a("code",[e._v("monitorInterval")]),e._v(" (default 30 seconds) and "),a("code",[e._v("noPollThreshold")]),e._v(" (default 3.0) properties in the "),a("code",[e._v("ContainerProperties")]),e._v(" when configuring the listener container.\nThe "),a("code",[e._v("noPollThreshold")]),e._v(" should be greater than "),a("code",[e._v("1.0")]),e._v(" to avoid getting spurious events due to a race condition.\nReceiving such an event lets you stop the containers, thus waking the consumer so that it can stop.")]),e._v(" "),a("p",[e._v("Starting with version 2.6.2, if a container has published a "),a("code",[e._v("ListenerContainerIdleEvent")]),e._v(", it will publish a "),a("code",[e._v("ListenerContainerNoLongerIdleEvent")]),e._v(" when a record is subsequently received.")]),e._v(" "),a("h5",{attrs:{id:"event-consumption"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#event-consumption"}},[e._v("#")]),e._v(" Event Consumption")]),e._v(" "),a("p",[e._v("You can capture these events by implementing "),a("code",[e._v("ApplicationListener")]),e._v(" — either a general listener or one narrowed to only receive this specific event.\nYou can also use "),a("code",[e._v("@EventListener")]),e._v(", introduced in Spring Framework 4.2.")]),e._v(" "),a("p",[e._v("The next example combines "),a("code",[e._v("@KafkaListener")]),e._v(" and "),a("code",[e._v("@EventListener")]),e._v(" into a single class.\nYou 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.\nYou can also use the "),a("code",[e._v("@EventListener")]),e._v(" "),a("code",[e._v("condition")]),e._v(" for this purpose.")]),e._v(" "),a("p",[e._v("See "),a("a",{attrs:{href:"#events"}},[e._v("Application Events")]),e._v(" for information about event properties.")]),e._v(" "),a("p",[e._v("The event is normally published on the consumer thread, so it is safe to interact with the "),a("code",[e._v("Consumer")]),e._v(" object.")]),e._v(" "),a("p",[e._v("The following example uses both "),a("code",[e._v("@KafkaListener")]),e._v(" and "),a("code",[e._v("@EventListener")]),e._v(":")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('public class Listener {\n\n @KafkaListener(id = "qux", topics = "annotated")\n public void listen4(@Payload String foo, Acknowledgment ack) {\n ...\n }\n\n @EventListener(condition = "event.listenerId.startsWith(\'qux-\')")\n public void eventHandler(ListenerContainerIdleEvent event) {\n ...\n }\n\n}\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("Event listeners see events for all containers."),a("br"),e._v("Consequently, in the preceding example, we narrow the events received based on the listener ID."),a("br"),e._v("Since containers created for the "),a("code",[e._v("@KafkaListener")]),e._v(" support concurrency, the actual containers are named "),a("code",[e._v("id-n")]),e._v(" where the "),a("code",[e._v("n")]),e._v(" is a unique value for each instance to support the concurrency."),a("br"),e._v("That is why we use "),a("code",[e._v("startsWith")]),e._v(" in the condition.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("If you wish to use the idle event to stop the lister container, you should not call "),a("code",[e._v("container.stop()")]),e._v(" on the thread that calls the listener."),a("br"),e._v("Doing so causes delays and unnecessary log messages."),a("br"),e._v("Instead, you should hand off the event to a different thread that can then stop the container."),a("br"),e._v("Also, you should not "),a("code",[e._v("stop()")]),e._v(" the container instance if it is a child container."),a("br"),e._v("You should stop the concurrent container instead.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h6",{attrs:{id:"current-positions-when-idle"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#current-positions-when-idle"}},[e._v("#")]),e._v(" Current Positions when Idle")]),e._v(" "),a("p",[e._v("Note that you can obtain the current positions when idle is detected by implementing "),a("code",[e._v("ConsumerSeekAware")]),e._v(" in your listener.\nSee "),a("code",[e._v("onIdleContainer()")]),e._v(" in "),a("a",{attrs:{href:"#seek"}},[e._v("Seeking to a Specific Offset")]),e._v(".")]),e._v(" "),a("h4",{attrs:{id:"_4-1-7-topic-partition-initial-offset"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-7-topic-partition-initial-offset"}},[e._v("#")]),e._v(" 4.1.7. Topic/Partition Initial Offset")]),e._v(" "),a("p",[e._v("There are several ways to set the initial offset for a partition.")]),e._v(" "),a("p",[e._v("When manually assigning partitions, you can set the initial offset (if desired) in the configured "),a("code",[e._v("TopicPartitionOffset")]),e._v(" arguments (see "),a("a",{attrs:{href:"#message-listener-container"}},[e._v("Message Listener Containers")]),e._v(").\nYou can also seek to a specific offset at any time.")]),e._v(" "),a("p",[e._v("When you use group management where the broker assigns partitions:")]),e._v(" "),a("ul",[a("li",[a("p",[e._v("For a new "),a("code",[e._v("group.id")]),e._v(", the initial offset is determined by the "),a("code",[e._v("auto.offset.reset")]),e._v(" consumer property ("),a("code",[e._v("earliest")]),e._v(" or "),a("code",[e._v("latest")]),e._v(").")])]),e._v(" "),a("li",[a("p",[e._v("For an existing group ID, the initial offset is the current offset for that group ID.\nYou can, however, seek to a specific offset during initialization (or at any time thereafter).")])])]),e._v(" "),a("h4",{attrs:{id:"_4-1-8-seeking-to-a-specific-offset"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-8-seeking-to-a-specific-offset"}},[e._v("#")]),e._v(" 4.1.8. Seeking to a Specific Offset")]),e._v(" "),a("p",[e._v("In order to seek, your listener must implement "),a("code",[e._v("ConsumerSeekAware")]),e._v(", which has the following methods:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("void registerSeekCallback(ConsumerSeekCallback callback);\n\nvoid onPartitionsAssigned(Map assignments, ConsumerSeekCallback callback);\n\nvoid onPartitionsRevoked(Collection partitions)\n\nvoid onIdleContainer(Map assignments, ConsumerSeekCallback callback);\n")])])]),a("p",[e._v("The "),a("code",[e._v("registerSeekCallback")]),e._v(" is called when the container is started and whenever partitions are assigned.\nYou should use this callback when seeking at some arbitrary time after initialization.\nYou should save a reference to the callback.\nIf you use the same listener in multiple containers (or in a "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v("), you should store the callback in a "),a("code",[e._v("ThreadLocal")]),e._v(" or some other structure keyed by the listener "),a("code",[e._v("Thread")]),e._v(".")]),e._v(" "),a("p",[e._v("When using group management, "),a("code",[e._v("onPartitionsAssigned")]),e._v(" is called when partitions are assigned.\nYou can use this method, for example, for setting initial offsets for the partitions, by calling the callback.\nYou can also use this method to associate this thread’s callback with the assigned partitions (see the example below).\nYou must use the callback argument, not the one passed into "),a("code",[e._v("registerSeekCallback")]),e._v(".\nStarting with version 2.5.5, this method is called, even when using "),a("a",{attrs:{href:"#manual-assignment"}},[e._v("manual partition assignment")]),e._v(".")]),e._v(" "),a("p",[a("code",[e._v("onPartitionsRevoked")]),e._v(" is called when the container is stopped or Kafka revokes assignments.\nYou should discard this thread’s callback and remove any associations to the revoked partitions.")]),e._v(" "),a("p",[e._v("The callback has the following methods:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("void seek(String topic, int partition, long offset);\n\nvoid seekToBeginning(String topic, int partition);\n\nvoid seekToBeginning(Collection= partitions);\n\nvoid seekToEnd(String topic, int partition);\n\nvoid seekToEnd(Collection= partitions);\n\nvoid seekRelative(String topic, int partition, long offset, boolean toCurrent);\n\nvoid seekToTimestamp(String topic, int partition, long timestamp);\n\nvoid seekToTimestamp(Collection topicPartitions, long timestamp);\n")])])]),a("p",[a("code",[e._v("seekRelative")]),e._v(" was added in version 2.3, to perform relative seeks.")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("offset")]),e._v(" negative and "),a("code",[e._v("toCurrent")]),e._v(" "),a("code",[e._v("false")]),e._v(" - seek relative to the end of the partition.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("offset")]),e._v(" positive and "),a("code",[e._v("toCurrent")]),e._v(" "),a("code",[e._v("false")]),e._v(" - seek relative to the beginning of the partition.")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("offset")]),e._v(" negative and "),a("code",[e._v("toCurrent")]),e._v(" "),a("code",[e._v("true")]),e._v(" - seek relative to the current position (rewind).")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("offset")]),e._v(" positive and "),a("code",[e._v("toCurrent")]),e._v(" "),a("code",[e._v("true")]),e._v(" - seek relative to the current position (fast forward).")])])]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("seekToTimestamp")]),e._v(" methods were also added in version 2.3.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("When seeking to the same timestamp for multiple partitions in the "),a("code",[e._v("onIdleContainer")]),e._v(" or "),a("code",[e._v("onPartitionsAssigned")]),e._v(" methods, the second method is preferred because it is more efficient to find the offsets for the timestamps in a single call to the consumer’s "),a("code",[e._v("offsetsForTimes")]),e._v(" method."),a("br"),e._v("When called from other locations, the container will gather all timestamp seek requests and make one call to "),a("code",[e._v("offsetsForTimes")]),e._v(".")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("You can also perform seek operations from "),a("code",[e._v("onIdleContainer()")]),e._v(" when an idle container is detected.\nSee "),a("a",{attrs:{href:"#idle-containers"}},[e._v("Detecting Idle and Non-Responsive Consumers")]),e._v(" for how to enable idle container detection.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("The "),a("code",[e._v("seekToBeginning")]),e._v(" method that accepts a collection is useful, for example, when processing a compacted topic and you wish to seek to the beginning every time the application is started:")])])]),e._v(" "),a("tbody")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("public class MyListener implements ConsumerSeekAware {\n\n...\n\n @Override\n public void onPartitionsAssigned(Map assignments, ConsumerSeekCallback callback) {\n callback.seekToBeginning(assignments.keySet());\n }\n\n}\n")])])]),a("p",[e._v("To arbitrarily seek at runtime, use the callback reference from the "),a("code",[e._v("registerSeekCallback")]),e._v(" for the appropriate thread.")]),e._v(" "),a("p",[e._v("Here is a trivial Spring Boot application that demonstrates how to use the callback; it sends 10 records to the topic; hitting "),a("code",[e._v("")]),e._v(" in the console causes all partitions to seek to the beginning.")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@SpringBootApplication\npublic class SeekExampleApplication {\n\n public static void main(String[] args) {\n SpringApplication.run(SeekExampleApplication.class, args);\n }\n\n @Bean\n public ApplicationRunner runner(Listener listener, KafkaTemplate template) {\n return args -> {\n IntStream.range(0, 10).forEach(i -> template.send(\n new ProducerRecord<>("seekExample", i % 3, "foo", "bar")));\n while (true) {\n System.in.read();\n listener.seekToStart();\n }\n };\n }\n\n @Bean\n public NewTopic topic() {\n return new NewTopic("seekExample", 3, (short) 1);\n }\n\n}\n\n@Component\nclass Listener implements ConsumerSeekAware {\n\n private static final Logger logger = LoggerFactory.getLogger(Listener.class);\n\n private final ThreadLocal callbackForThread = new ThreadLocal<>();\n\n private final Map callbacks = new ConcurrentHashMap<>();\n\n @Override\n public void registerSeekCallback(ConsumerSeekCallback callback) {\n this.callbackForThread.set(callback);\n }\n\n @Override\n public void onPartitionsAssigned(Map assignments, ConsumerSeekCallback callback) {\n assignments.keySet().forEach(tp -> this.callbacks.put(tp, this.callbackForThread.get()));\n }\n\n @Override\n public void onPartitionsRevoked(Collection partitions) {\n partitions.forEach(tp -> this.callbacks.remove(tp));\n this.callbackForThread.remove();\n }\n\n @Override\n public void onIdleContainer(Map assignments, ConsumerSeekCallback callback) {\n }\n\n @KafkaListener(id = "seekExample", topics = "seekExample", concurrency = "3")\n public void listen(ConsumerRecord in) {\n logger.info(in.toString());\n }\n\n public void seekToStart() {\n this.callbacks.forEach((tp, callback) -> callback.seekToBeginning(tp.topic(), tp.partition()));\n }\n\n}\n')])])]),a("p",[e._v("To make things simpler, version 2.3 added the "),a("code",[e._v("AbstractConsumerSeekAware")]),e._v(" class, which keeps track of which callback is to be used for a topic/partition.\nThe following example shows how to seek to the last record processed, in each partition, each time the container goes idle.\nIt also has methods that allow arbitrary external calls to rewind partitions by one record.")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('public class SeekToLastOnIdleListener extends AbstractConsumerSeekAware {\n\n @KafkaListener(id = "seekOnIdle", topics = "seekOnIdle")\n public void listen(String in) {\n ...\n }\n\n @Override\n public void onIdleContainer(Map assignments,\n ConsumerSeekCallback callback) {\n\n assignments.keySet().forEach(tp -> callback.seekRelative(tp.topic(), tp.partition(), -1, true));\n }\n\n /**\n * Rewind all partitions one record.\n */\n public void rewindAllOneRecord() {\n getSeekCallbacks()\n .forEach((tp, callback) ->\n callback.seekRelative(tp.topic(), tp.partition(), -1, true));\n }\n\n /**\n * Rewind one partition one record.\n */\n public void rewindOnePartitionOneRecord(String topic, int partition) {\n getSeekCallbackFor(new org.apache.kafka.common.TopicPartition(topic, partition))\n .seekRelative(topic, partition, -1, true);\n }\n\n}\n')])])]),a("p",[e._v("Version 2.6 added convenience methods to the abstract class:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("seekToBeginning()")]),e._v(" - seeks all assigned partitions to the beginning")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("seekToEnd()")]),e._v(" - seeks all assigned partitions to the end")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("seekToTimestamp(long time)")]),e._v(" - seeks all assigned partitions to the offset represented by that timestamp.")])])]),e._v(" "),a("p",[e._v("Example:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("public class MyListener extends AbstractConsumerSeekAware {\n\n @KafkaListener(...)\n void listn(...) {\n ...\n }\n}\n\npublic class SomeOtherBean {\n\n MyListener listener;\n\n ...\n\n void someMethod() {\n this.listener.seekToTimestamp(System.currentTimeMillis - 60_000);\n }\n\n}\n")])])]),a("h4",{attrs:{id:"_4-1-9-container-factory"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-9-container-factory"}},[e._v("#")]),e._v(" 4.1.9. Container factory")]),e._v(" "),a("p",[e._v("As discussed in "),a("a",{attrs:{href:"#kafka-listener-annotation"}},[a("code",[e._v("@KafkaListener")]),e._v(" Annotation")]),e._v(", a "),a("code",[e._v("ConcurrentKafkaListenerContainerFactory")]),e._v(" is used to create containers for annotated methods.")]),e._v(" "),a("p",[e._v("Starting with version 2.2, you can use the same factory to create any "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(".\nThis 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.\nOnce the container is created, you can further modify its properties, many of which are set by using "),a("code",[e._v("container.getContainerProperties()")]),e._v(".\nThe following example configures a "),a("code",[e._v("ConcurrentMessageListenerContainer")]),e._v(":")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Bean\npublic ConcurrentMessageListenerContainer(\n ConcurrentKafkaListenerContainerFactory factory) {\n\n ConcurrentMessageListenerContainer container =\n factory.createContainer("topic1", "topic2");\n container.setMessageListener(m -> { ... } );\n return container;\n}\n')])])]),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("Containers created this way are not added to the endpoint registry."),a("br"),e._v("They should be created as "),a("code",[e._v("@Bean")]),e._v(" definitions so that they are registered with the application context.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("Starting with version 2.3.4, you can add a "),a("code",[e._v("ContainerCustomizer")]),e._v(" to the factory to further configure each container after it has been created and configured.")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v("@Bean\npublic KafkaListenerContainerFactory kafkaListenerContainerFactory() {\n ConcurrentKafkaListenerContainerFactory factory =\n new ConcurrentKafkaListenerContainerFactory<>();\n ...\n factory.setContainerCustomizer(container -> { /* customize the container */ });\n return factory;\n}\n")])])]),a("h4",{attrs:{id:"_4-1-10-thread-safety"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-10-thread-safety"}},[e._v("#")]),e._v(" 4.1.10. Thread Safety")]),e._v(" "),a("p",[e._v("When using a concurrent message listener container, a single listener instance is invoked on all consumer threads.\nListeners, therefore, need to be thread-safe, and it is preferable to use stateless listeners.\nIf it is not possible to make your listener thread-safe or adding synchronization would significantly reduce the benefit of adding concurrency, you can use one of a few techniques:")]),e._v(" "),a("ul",[a("li",[a("p",[e._v("Use "),a("code",[e._v("n")]),e._v(" containers with "),a("code",[e._v("concurrency=1")]),e._v(" with a prototype scoped "),a("code",[e._v("MessageListener")]),e._v(" bean so that each container gets its own instance (this is not possible when using "),a("code",[e._v("@KafkaListener")]),e._v(").")])]),e._v(" "),a("li",[a("p",[e._v("Keep the state in "),a("code",[e._v("ThreadLocal")]),e._v(" instances.")])]),e._v(" "),a("li",[a("p",[e._v("Have the singleton listener delegate to a bean that is declared in "),a("code",[e._v("SimpleThreadScope")]),e._v(" (or a similar scope).")])])]),e._v(" "),a("p",[e._v("To facilitate cleaning up thread state (for the second and third items in the preceding list), starting with version 2.2, the listener container publishes a "),a("code",[e._v("ConsumerStoppedEvent")]),e._v(" when each thread exits.\nYou can consume these events with an "),a("code",[e._v("ApplicationListener")]),e._v(" or "),a("code",[e._v("@EventListener")]),e._v(" method to remove "),a("code",[e._v("ThreadLocal")]),e._v(" instances or "),a("code",[e._v("remove()")]),e._v(" thread-scoped beans from the scope.\nNote that "),a("code",[e._v("SimpleThreadScope")]),e._v(" does not destroy beans that have a destruction interface (such as "),a("code",[e._v("DisposableBean")]),e._v("), so you should "),a("code",[e._v("destroy()")]),e._v(" the instance yourself.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("By default, the application context’s event multicaster invokes event listeners on the calling thread."),a("br"),e._v("If you change the multicaster to use an async executor, thread cleanup is not effective.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h4",{attrs:{id:"_4-1-11-monitoring"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-11-monitoring"}},[e._v("#")]),e._v(" 4.1.11. Monitoring")]),e._v(" "),a("h5",{attrs:{id:"monitoring-listener-performance"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#monitoring-listener-performance"}},[e._v("#")]),e._v(" Monitoring Listener Performance")]),e._v(" "),a("p",[e._v("Starting with version 2.3, the listener container will automatically create and update Micrometer "),a("code",[e._v("Timer")]),e._v(" s for the listener, if "),a("code",[e._v("Micrometer")]),e._v(" is detected on the class path, and a single "),a("code",[e._v("MeterRegistry")]),e._v(" is present in the application context.\nThe timers can be disabled by setting the "),a("code",[e._v("ContainerProperty")]),e._v(" "),a("code",[e._v("micrometerEnabled")]),e._v(" to "),a("code",[e._v("false")]),e._v(".")]),e._v(" "),a("p",[e._v("Two timers are maintained - one for successful calls to the listener and one for failures.")]),e._v(" "),a("p",[e._v("The timers are named "),a("code",[e._v("spring.kafka.listener")]),e._v(" and have the following tags:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("name")]),e._v(" : (container bean name)")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("result")]),e._v(" : "),a("code",[e._v("success")]),e._v(" or "),a("code",[e._v("failure")])])]),e._v(" "),a("li",[a("p",[a("code",[e._v("exception")]),e._v(" : "),a("code",[e._v("none")]),e._v(" or "),a("code",[e._v("ListenerExecutionFailedException")])])])]),e._v(" "),a("p",[e._v("You can add additional tags using the "),a("code",[e._v("ContainerProperties")]),e._v(" "),a("code",[e._v("micrometerTags")]),e._v(" property.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("With the concurrent container, timers are created for each thread and the "),a("code",[e._v("name")]),e._v(" tag is suffixed with "),a("code",[e._v("-n")]),e._v(" where n is "),a("code",[e._v("0")]),e._v(" to "),a("code",[e._v("concurrency-1")]),e._v(".")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h5",{attrs:{id:"monitoring-kafkatemplate-performance"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#monitoring-kafkatemplate-performance"}},[e._v("#")]),e._v(" Monitoring KafkaTemplate Performance")]),e._v(" "),a("p",[e._v("Starting with version 2.5, the template will automatically create and update Micrometer "),a("code",[e._v("Timer")]),e._v(" s for send operations, if "),a("code",[e._v("Micrometer")]),e._v(" is detected on the class path, and a single "),a("code",[e._v("MeterRegistry")]),e._v(" is present in the application context.\nThe timers can be disabled by setting the template’s "),a("code",[e._v("micrometerEnabled")]),e._v(" property to "),a("code",[e._v("false")]),e._v(".")]),e._v(" "),a("p",[e._v("Two timers are maintained - one for successful calls to the listener and one for failures.")]),e._v(" "),a("p",[e._v("The timers are named "),a("code",[e._v("spring.kafka.template")]),e._v(" and have the following tags:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("name")]),e._v(" : (template bean name)")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("result")]),e._v(" : "),a("code",[e._v("success")]),e._v(" or "),a("code",[e._v("failure")])])]),e._v(" "),a("li",[a("p",[a("code",[e._v("exception")]),e._v(" : "),a("code",[e._v("none")]),e._v(" or the exception class name for failures")])])]),e._v(" "),a("p",[e._v("You can add additional tags using the template’s "),a("code",[e._v("micrometerTags")]),e._v(" property.")]),e._v(" "),a("h5",{attrs:{id:"micrometer-native-metrics"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#micrometer-native-metrics"}},[e._v("#")]),e._v(" Micrometer Native Metrics")]),e._v(" "),a("p",[e._v("Starting with version 2.5, the framework provides "),a("a",{attrs:{href:"#factory-listeners"}},[e._v("Factory Listeners")]),e._v(" to manage a Micrometer "),a("code",[e._v("KafkaClientMetrics")]),e._v(" instance whenever producers and consumers are created and closed.")]),e._v(" "),a("p",[e._v("To enable this feature, simply add the listeners to your producer and consumer factories:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Bean\npublic ConsumerFactory myConsumerFactory() {\n Map configs = consumerConfigs();\n ...\n DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(configs);\n ...\n cf.addListener(new MicrometerConsumerListener(meterRegistry(),\n Collections.singletonList(new ImmutableTag("customTag", "customTagValue"))));\n ...\n return cf;\n}\n\n@Bean\npublic ProducerFactory myProducerFactory() {\n Map configs = producerConfigs();\n configs.put(ProducerConfig.CLIENT_ID_CONFIG, "myClientId");\n ...\n DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(configs);\n ...\n pf.addListener(new MicrometerProducerListener(meterRegistry(),\n Collections.singletonList(new ImmutableTag("customTag", "customTagValue"))));\n ...\n return pf;\n}\n')])])]),a("p",[e._v("The consumer/producer "),a("code",[e._v("id")]),e._v(" passed to the listener is added to the meter’s tags with tag name "),a("code",[e._v("spring.id")]),e._v(".")]),e._v(" "),a("p",[e._v("An example of obtaining one of the Kafka metrics")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('double count = this.meterRegistry.get("kafka.producer.node.incoming.byte.total")\n .tag("customTag", "customTagValue")\n .tag("spring.id", "myProducerFactory.myClientId-1")\n .functionCounter()\n .count()\n')])])]),a("p",[e._v("A similar listener is provided for the "),a("code",[e._v("StreamsBuilderFactoryBean")]),e._v(" - see "),a("a",{attrs:{href:"#streams-micrometer"}},[e._v("KafkaStreams Micrometer Support")]),e._v(".")]),e._v(" "),a("h4",{attrs:{id:"_4-1-12-transactions"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-12-transactions"}},[e._v("#")]),e._v(" 4.1.12. Transactions")]),e._v(" "),a("p",[e._v("This section describes how Spring for Apache Kafka supports transactions.")]),e._v(" "),a("h5",{attrs:{id:"overview-2"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#overview-2"}},[e._v("#")]),e._v(" Overview")]),e._v(" "),a("p",[e._v("The 0.11.0.0 client library added support for transactions.\nSpring for Apache Kafka adds support in the following ways:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("KafkaTransactionManager")]),e._v(": Used with normal Spring transaction support ("),a("code",[e._v("@Transactional")]),e._v(", "),a("code",[e._v("TransactionTemplate")]),e._v(" etc).")])]),e._v(" "),a("li",[a("p",[e._v("Transactional "),a("code",[e._v("KafkaMessageListenerContainer")])])]),e._v(" "),a("li",[a("p",[e._v("Local transactions with "),a("code",[e._v("KafkaTemplate")])])]),e._v(" "),a("li",[a("p",[e._v("Transaction synchronization with other transaction managers")])])]),e._v(" "),a("p",[e._v("Transactions are enabled by providing the "),a("code",[e._v("DefaultKafkaProducerFactory")]),e._v(" with a "),a("code",[e._v("transactionIdPrefix")]),e._v(".\nIn that case, instead of managing a single shared "),a("code",[e._v("Producer")]),e._v(", the factory maintains a cache of transactional producers.\nWhen the user calls "),a("code",[e._v("close()")]),e._v(" on a producer, it is returned to the cache for reuse instead of actually being closed.\nThe "),a("code",[e._v("transactional.id")]),e._v(" property of each producer is "),a("code",[e._v("transactionIdPrefix")]),e._v(" + "),a("code",[e._v("n")]),e._v(", where "),a("code",[e._v("n")]),e._v(" starts with "),a("code",[e._v("0")]),e._v(" and is incremented for each new producer, unless the transaction is started by a listener container with a record-based listener.\nIn that case, the "),a("code",[e._v("transactional.id")]),e._v(" is "),a("code",[e._v("...")]),e._v(".\nThis is to properly support fencing zombies, "),a("a",{attrs:{href:"https://www.confluent.io/blog/transactions-apache-kafka/",target:"_blank",rel:"noopener noreferrer"}},[e._v("as described here"),a("OutboundLink")],1),e._v(".\nThis new behavior was added in versions 1.3.7, 2.0.6, 2.1.10, and 2.2.0.\nIf you wish to revert to the previous behavior, you can set the "),a("code",[e._v("producerPerConsumerPartition")]),e._v(" property on the "),a("code",[e._v("DefaultKafkaProducerFactory")]),e._v(" to "),a("code",[e._v("false")]),e._v(".")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("While transactions are supported with batch listeners, by default, zombie fencing is not supported because a batch may contain records from multiple topics or partitions."),a("br"),e._v("However, starting with version 2.3.2, zombie fencing is supported if you set the container property "),a("code",[e._v("subBatchPerPartition")]),e._v(" to true."),a("br"),e._v("In that case, the batch listener is invoked once per partition received from the last poll, as if each poll only returned records for a single partition."),a("br"),e._v("This is "),a("code",[e._v("true")]),e._v(" by default since version 2.5 when transactions are enabled with "),a("code",[e._v("EOSMode.ALPHA")]),e._v("; set it to "),a("code",[e._v("false")]),e._v(" if you are using transactions but are not concerned about zombie fencing."),a("br"),e._v("Also see "),a("a",{attrs:{href:"#exactly-once"}},[e._v("Exactly Once Semantics")]),e._v(".")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("Also see "),a("a",{attrs:{href:"#transaction-id-prefix"}},[a("code",[e._v("transactionIdPrefix")])]),e._v(".")]),e._v(" "),a("p",[e._v("With Spring Boot, it is only necessary to set the "),a("code",[e._v("spring.kafka.producer.transaction-id-prefix")]),e._v(" property - Boot will automatically configure a "),a("code",[e._v("KafkaTransactionManager")]),e._v(" bean and wire it into the listener container.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("Starting with version 2.5.8, you can now configure the "),a("code",[e._v("maxAge")]),e._v(" property on the producer factory."),a("br"),e._v("This is useful when using transactional producers that might lay idle for the broker’s "),a("code",[e._v("transactional.id.expiration.ms")]),e._v("."),a("br"),e._v("With current "),a("code",[e._v("kafka-clients")]),e._v(", this can cause a "),a("code",[e._v("ProducerFencedException")]),e._v(" without a rebalance."),a("br"),e._v("By setting the "),a("code",[e._v("maxAge")]),e._v(" to less than "),a("code",[e._v("transactional.id.expiration.ms")]),e._v(", the factory will refresh the producer if it is past it’s max age.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h5",{attrs:{id:"using-kafkatransactionmanager"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#using-kafkatransactionmanager"}},[e._v("#")]),e._v(" Using "),a("code",[e._v("KafkaTransactionManager")])]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("KafkaTransactionManager")]),e._v(" is an implementation of Spring Framework’s "),a("code",[e._v("PlatformTransactionManager")]),e._v(".\nIt is provided with a reference to the producer factory in its constructor.\nIf you provide a custom producer factory, it must support transactions.\nSee "),a("code",[e._v("ProducerFactory.transactionCapable()")]),e._v(".")]),e._v(" "),a("p",[e._v("You can use the "),a("code",[e._v("KafkaTransactionManager")]),e._v(" with normal Spring transaction support ("),a("code",[e._v("@Transactional")]),e._v(", "),a("code",[e._v("TransactionTemplate")]),e._v(", and others).\nIf a transaction is active, any "),a("code",[e._v("KafkaTemplate")]),e._v(" operations performed within the scope of the transaction use the transaction’s "),a("code",[e._v("Producer")]),e._v(".\nThe manager commits or rolls back the transaction, depending on success or failure.\nYou must configure the "),a("code",[e._v("KafkaTemplate")]),e._v(" to use the same "),a("code",[e._v("ProducerFactory")]),e._v(" as the transaction manager.")]),e._v(" "),a("h5",{attrs:{id:"transaction-synchronization"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#transaction-synchronization"}},[e._v("#")]),e._v(" Transaction Synchronization")]),e._v(" "),a("p",[e._v("This section refers to producer-only transactions (transactions not started by a listener container); see "),a("a",{attrs:{href:"#container-transaction-manager"}},[e._v("Using Consumer-Initiated Transactions")]),e._v(" for information about chaining transactions when the container starts the transaction.")]),e._v(" "),a("p",[e._v("If you want to send records to kafka and perform some database updates, you can use normal Spring transaction management with, say, a "),a("code",[e._v("DataSourceTransactionManager")]),e._v(".")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@Transactional\npublic void process(List things) {\n things.forEach(thing -> this.kafkaTemplate.send("topic", thing));\n updateDb(things);\n}\n')])])]),a("p",[e._v("The interceptor for the "),a("code",[e._v("@Transactional")]),e._v(" annotation starts the transaction and the "),a("code",[e._v("KafkaTemplate")]),e._v(" will synchronize a transaction with that transaction manager; each send will participate in that transaction.\nWhen the method exits, the database transaction will commit followed by the Kafka transaction.\nIf you wish the commits to be performed in the reverse order (Kafka first), use nested "),a("code",[e._v("@Transactional")]),e._v(" methods, with the outer method configured to use the "),a("code",[e._v("DataSourceTransactionManager")]),e._v(", and the inner method configured to use the "),a("code",[e._v("KafkaTransactionManager")]),e._v(".")]),e._v(" "),a("p",[e._v("See "),a("a",{attrs:{href:"#ex-jdbc-sync"}},[e._v("[ex-jdbc-sync]")]),e._v(" for examples of an application that synchronizes JDBC and Kafka transactions in Kafka-first or DB-first configurations.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("Starting with versions 2.5.17, 2.6.12, 2.7.9 and 2.8.0, if the commit fails on the synchronized transaction (after the primary transaction has committed), the exception will be thrown to the caller."),a("br"),e._v("Previously, this was silently ignored (logged at debug)."),a("br"),e._v("Applications should take remedial action, if necessary, to compensate for the committed primary transaction.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("h5",{attrs:{id:"using-consumer-initiated-transactions"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#using-consumer-initiated-transactions"}},[e._v("#")]),e._v(" Using Consumer-Initiated Transactions")]),e._v(" "),a("p",[e._v("The "),a("code",[e._v("ChainedKafkaTransactionManager")]),e._v(" is now deprecated, since version 2.7; see the javadocs for its super class "),a("code",[e._v("ChainedTransactionManager")]),e._v(" for more information.\nInstead, use a "),a("code",[e._v("KafkaTransactionManager")]),e._v(" in the container to start the Kafka transaction and annotate the listener method with "),a("code",[e._v("@Transactional")]),e._v(" to start the other transaction.")]),e._v(" "),a("p",[e._v("See "),a("a",{attrs:{href:"#ex-jdbc-sync"}},[e._v("[ex-jdbc-sync]")]),e._v(" for an example application that chains JDBC and Kafka transactions.")]),e._v(" "),a("h5",{attrs:{id:"kafkatemplate-local-transactions"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#kafkatemplate-local-transactions"}},[e._v("#")]),e._v(" "),a("code",[e._v("KafkaTemplate")]),e._v(" Local Transactions")]),e._v(" "),a("p",[e._v("You can use the "),a("code",[e._v("KafkaTemplate")]),e._v(" to execute a series of operations within a local transaction.\nThe following example shows how to do so:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('boolean result = template.executeInTransaction(t -> {\n t.sendDefault("thing1", "thing2");\n t.sendDefault("cat", "hat");\n return true;\n});\n')])])]),a("p",[e._v("The argument in the callback is the template itself ("),a("code",[e._v("this")]),e._v(").\nIf the callback exits normally, the transaction is committed.\nIf an exception is thrown, the transaction is rolled back.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("If there is a "),a("code",[e._v("KafkaTransactionManager")]),e._v(" (or synchronized) transaction in process, it is not used."),a("br"),e._v('Instead, a new "nested" transaction is used.')])])]),e._v(" "),a("tbody")]),e._v(" "),a("h5",{attrs:{id:"transactionidprefix"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#transactionidprefix"}},[e._v("#")]),e._v(" "),a("code",[e._v("transactionIdPrefix")])]),e._v(" "),a("p",[e._v("As mentioned in "),a("a",{attrs:{href:"#transactions"}},[e._v("the overview")]),e._v(", the producer factory is configured with this property to build the producer "),a("code",[e._v("transactional.id")]),e._v(" property.\nThere is a dichotomy when specifying this property in that, when running multiple instances of the application with "),a("code",[e._v("EOSMode.ALPHA")]),e._v(", it must be the same on all instances to satisfy fencing zombies (also mentioned in the overview) when producing records on a listener container thread.\nHowever, when producing records using transactions that are "),a("strong",[e._v("not")]),e._v(" started by a listener container, the prefix has to be different on each instance.\nVersion 2.3, makes this simpler to configure, especially in a Spring Boot application.\nIn previous versions, you had to create two producer factories and "),a("code",[e._v("KafkaTemplate")]),e._v(" s - one for producing records on a listener container thread and one for stand-alone transactions started by "),a("code",[e._v("kafkaTemplate.executeInTransaction()")]),e._v(" or by a transaction interceptor on a "),a("code",[e._v("@Transactional")]),e._v(" method.")]),e._v(" "),a("p",[e._v("Now, you can override the factory’s "),a("code",[e._v("transactionalIdPrefix")]),e._v(" on the "),a("code",[e._v("KafkaTemplate")]),e._v(" and the "),a("code",[e._v("KafkaTransactionManager")]),e._v(".")]),e._v(" "),a("p",[e._v("When using a transaction manager and template for a listener container, you would normally leave this to default to the producer factory’s property.\nThis value should be the same for all application instances when using "),a("code",[e._v("EOSMode.ALPHA")]),e._v(".\nWith "),a("code",[e._v("EOSMode.BETA")]),e._v(" it is no longer necessary to use the same "),a("code",[e._v("transactional.id")]),e._v(", even for consumer-initiated transactions; in fact, it must be unique on each instance the same as producer-initiated transactions.\nFor transactions started by the template (or the transaction manager for "),a("code",[e._v("@Transaction")]),e._v(") you should set the property on the template and transaction manager respectively.\nThis property must have a different value on each application instance.")]),e._v(" "),a("p",[e._v("This problem (different rules for "),a("code",[e._v("transactional.id")]),e._v(") has been eliminated when "),a("code",[e._v("EOSMode.BETA")]),e._v(" is being used (with broker versions >= 2.5); see "),a("a",{attrs:{href:"#exactly-once"}},[e._v("Exactly Once Semantics")]),e._v(".")]),e._v(" "),a("h5",{attrs:{id:"kafkatemplate-transactional-and-non-transactional-publishing"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#kafkatemplate-transactional-and-non-transactional-publishing"}},[e._v("#")]),e._v(" "),a("code",[e._v("KafkaTemplate")]),e._v(" Transactional and non-Transactional Publishing")]),e._v(" "),a("p",[e._v("Normally, when a "),a("code",[e._v("KafkaTemplate")]),e._v(" is transactional (configured with a transaction-capable producer factory), transactions are required.\nThe transaction can be started by a "),a("code",[e._v("TransactionTemplate")]),e._v(", a "),a("code",[e._v("@Transactional")]),e._v(" method, calling "),a("code",[e._v("executeInTransaction")]),e._v(", or by a listener container, when configured with a "),a("code",[e._v("KafkaTransactionManager")]),e._v(".\nAny attempt to use the template outside the scope of a transaction results in the template throwing an "),a("code",[e._v("IllegalStateException")]),e._v(".\nStarting with version 2.4.3, you can set the template’s "),a("code",[e._v("allowNonTransactional")]),e._v(" property to "),a("code",[e._v("true")]),e._v(".\nIn that case, the template will allow the operation to run without a transaction, by calling the "),a("code",[e._v("ProducerFactory")]),e._v(" 's "),a("code",[e._v("createNonTransactionalProducer()")]),e._v(" method; the producer will be cached, or thread-bound, as normal for reuse.\nSee "),a("a",{attrs:{href:"#producer-factory"}},[e._v("Using "),a("code",[e._v("DefaultKafkaProducerFactory")])]),e._v(".")]),e._v(" "),a("h5",{attrs:{id:"transactions-with-batch-listeners"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#transactions-with-batch-listeners"}},[e._v("#")]),e._v(" Transactions with Batch Listeners")]),e._v(" "),a("p",[e._v("When a listener fails while transactions are being used, the "),a("code",[e._v("AfterRollbackProcessor")]),e._v(" is invoked to take some action after the rollback occurs.\nWhen using the default "),a("code",[e._v("AfterRollbackProcessor")]),e._v(" with a record listener, seeks are performed so that the failed record will be redelivered.\nWith a batch listener, however, the whole batch will be redelivered because the framework doesn’t know which record in the batch failed.\nSee "),a("a",{attrs:{href:"#after-rollback"}},[e._v("After-rollback Processor")]),e._v(" for more information.")]),e._v(" "),a("p",[e._v("When using a batch listener, version 2.4.2 introduced an alternative mechanism to deal with failures while processing a batch; the "),a("code",[e._v("BatchToRecordAdapter")]),e._v(".\nWhen a container factory with "),a("code",[e._v("batchListener")]),e._v(" set to true is configured with a "),a("code",[e._v("BatchToRecordAdapter")]),e._v(", the listener is invoked with one record at a time.\nThis enables error handling within the batch, while still making it possible to stop processing the entire batch, depending on the exception type.\nA default "),a("code",[e._v("BatchToRecordAdapter")]),e._v(" is provided, that can be configured with a standard "),a("code",[e._v("ConsumerRecordRecoverer")]),e._v(" such as the "),a("code",[e._v("DeadLetterPublishingRecoverer")]),e._v(".\nThe following test case configuration snippet illustrates how to use this feature:")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('public static class TestListener {\n\n final List values = new ArrayList<>();\n\n @KafkaListener(id = "batchRecordAdapter", topics = "test")\n public void listen(String data) {\n values.add(data);\n if ("bar".equals(data)) {\n throw new RuntimeException("reject partial");\n }\n }\n\n}\n\n@Configuration\n@EnableKafka\npublic static class Config {\n\n ConsumerRecord failed;\n\n @Bean\n public TestListener test() {\n return new TestListener();\n }\n\n @Bean\n public ConsumerFactory consumerFactory() {\n return mock(ConsumerFactory.class);\n }\n\n @Bean\n public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {\n ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();\n factory.setConsumerFactory(consumerFactory());\n factory.setBatchListener(true);\n factory.setBatchToRecordAdapter(new DefaultBatchToRecordAdapter<>((record, ex) -> {\n this.failed = record;\n }));\n return factory;\n }\n\n}\n')])])]),a("h4",{attrs:{id:"_4-1-13-exactly-once-semantics"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-13-exactly-once-semantics"}},[e._v("#")]),e._v(" 4.1.13. Exactly Once Semantics")]),e._v(" "),a("p",[e._v("You can provide a listener container with a "),a("code",[e._v("KafkaAwareTransactionManager")]),e._v(" instance.\nWhen so configured, the container starts a transaction before invoking the listener.\nAny "),a("code",[e._v("KafkaTemplate")]),e._v(" operations performed by the listener participate in the transaction.\nIf the listener successfully processes the record (or multiple records, when using a "),a("code",[e._v("BatchMessageListener")]),e._v("), the container sends the offset(s) to the transaction by using "),a("code",[e._v("producer.sendOffsetsToTransaction()")]),e._v("), before the transaction manager commits the transaction.\nIf 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.\nSee "),a("a",{attrs:{href:"#after-rollback"}},[e._v("After-rollback Processor")]),e._v(" for more information and for handling records that repeatedly fail.")]),e._v(" "),a("p",[e._v("Using transactions enables Exactly Once Semantics (EOS).")]),e._v(" "),a("p",[e._v("This means that, for a "),a("code",[e._v("read→process-write")]),e._v(" sequence, it is guaranteed that the "),a("strong",[e._v("sequence")]),e._v(" is completed exactly once.\n(The read and process are have at least once semantics).")]),e._v(" "),a("p",[e._v("Spring for Apache Kafka version 2.5 and later supports two EOS modes:")]),e._v(" "),a("ul",[a("li",[a("p",[a("code",[e._v("ALPHA")]),e._v(" - alias for "),a("code",[e._v("V1")]),e._v(" (deprecated)")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("BETA")]),e._v(" - alias for "),a("code",[e._v("V2")]),e._v(" (deprecated)")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("V1")]),e._v(" - aka "),a("code",[e._v("transactional.id")]),e._v(" fencing (since version 0.11.0.0)")])]),e._v(" "),a("li",[a("p",[a("code",[e._v("V2")]),e._v(" - aka fetch-offset-request fencing (since version 2.5)")])])]),e._v(" "),a("p",[e._v("With mode "),a("code",[e._v("V1")]),e._v(', the producer is "fenced" if another instance with the same '),a("code",[e._v("transactional.id")]),e._v(" is started.\nSpring manages this by using a "),a("code",[e._v("Producer")]),e._v(" for each "),a("code",[e._v("group.id/topic/partition")]),e._v("; when a rebalance occurs a new instance will use the same "),a("code",[e._v("transactional.id")]),e._v(" and the old producer is fenced.")]),e._v(" "),a("p",[e._v("With mode "),a("code",[e._v("V2")]),e._v(", it is not necessary to have a producer for each "),a("code",[e._v("group.id/topic/partition")]),e._v(" 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.")]),e._v(" "),a("p",[e._v("Starting with version 2.6, the default "),a("code",[e._v("EOSMode")]),e._v(" is "),a("code",[e._v("V2")]),e._v(".")]),e._v(" "),a("p",[e._v("To configure the container to use mode "),a("code",[e._v("ALPHA")]),e._v(", set the container property "),a("code",[e._v("EOSMode")]),e._v(" to "),a("code",[e._v("ALPHA")]),e._v(", to revert to the previous behavior.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("With "),a("code",[e._v("V2")]),e._v(" (default), your brokers must be version 2.5 or later; "),a("code",[e._v("kafka-clients")]),e._v(" version 3.0, the producer will no longer fall back to "),a("code",[e._v("V1")]),e._v("; if the broker does not support "),a("code",[e._v("V2")]),e._v(", an exception is thrown."),a("br"),e._v("If your brokers are earlier than 2.5, you must set the "),a("code",[e._v("EOSMode")]),e._v(" to "),a("code",[e._v("V1")]),e._v(", leave the "),a("code",[e._v("DefaultKafkaProducerFactory")]),e._v(" "),a("code",[e._v("producerPerConsumerPartition")]),e._v(" set to "),a("code",[e._v("true")]),e._v(" and, if you are using a batch listener, you should set "),a("code",[e._v("subBatchPerPartition")]),e._v(" to "),a("code",[e._v("true")]),e._v(".")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("When your brokers are upgraded to 2.5 or later, you should switch the mode to "),a("code",[e._v("V2")]),e._v(", but the number of producers will remain as before.\nYou can then do a rolling upgrade of your application with "),a("code",[e._v("producerPerConsumerPartition")]),e._v(" set to "),a("code",[e._v("false")]),e._v(" to reduce the number of producers; you should also no longer set the "),a("code",[e._v("subBatchPerPartition")]),e._v(" container property.")]),e._v(" "),a("p",[e._v("If your brokers are already 2.5 or newer, you should set the "),a("code",[e._v("DefaultKafkaProducerFactory")]),e._v(" "),a("code",[e._v("producerPerConsumerPartition")]),e._v(" property to "),a("code",[e._v("false")]),e._v(", to reduce the number of producers needed.")]),e._v(" "),a("table",[a("thead",[a("tr",[a("th"),e._v(" "),a("th",[e._v("When using "),a("code",[e._v("EOSMode.V2")]),e._v(" with "),a("code",[e._v("producerPerConsumerPartition=false")]),e._v(" the "),a("code",[e._v("transactional.id")]),e._v(" must be unique across all application instances.")])])]),e._v(" "),a("tbody")]),e._v(" "),a("p",[e._v("When using "),a("code",[e._v("V2")]),e._v(" mode, it is no longer necessary to set the "),a("code",[e._v("subBatchPerPartition")]),e._v(" to "),a("code",[e._v("true")]),e._v("; it will default to "),a("code",[e._v("false")]),e._v(" when the "),a("code",[e._v("EOSMode")]),e._v(" is "),a("code",[e._v("V2")]),e._v(".")]),e._v(" "),a("p",[e._v("Refer to "),a("a",{attrs:{href:"https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics",target:"_blank",rel:"noopener noreferrer"}},[e._v("KIP-447"),a("OutboundLink")],1),e._v(" for more information.")]),e._v(" "),a("p",[a("code",[e._v("V1")]),e._v(" and "),a("code",[e._v("V2")]),e._v(" were previously "),a("code",[e._v("ALPHA")]),e._v(" and "),a("code",[e._v("BETA")]),e._v("; they have been changed to align the framework with "),a("a",{attrs:{href:"https://cwiki.apache.org/confluence/display/KAFKA/KIP-732%3A+Deprecate+eos-alpha+and+replace+eos-beta+with+eos-v2",target:"_blank",rel:"noopener noreferrer"}},[e._v("KIP-732"),a("OutboundLink")],1),e._v(".")]),e._v(" "),a("h4",{attrs:{id:"_4-1-14-wiring-spring-beans-into-producer-consumer-interceptors"}},[a("a",{staticClass:"header-anchor",attrs:{href:"#_4-1-14-wiring-spring-beans-into-producer-consumer-interceptors"}},[e._v("#")]),e._v(" 4.1.14. Wiring Spring Beans into Producer/Consumer Interceptors")]),e._v(" "),a("p",[e._v("Apache Kafka provides a mechanism to add interceptors to producers and consumers.\nThese objects are managed by Kafka, not Spring, and so normal Spring dependency injection won’t work for wiring in dependent Spring Beans.\nHowever, you can manually wire in those dependencies using the interceptor "),a("code",[e._v("config()")]),e._v(" method.\nThe following Spring Boot application shows how to do this by overriding boot’s default factories to add some dependent bean into the configuration properties.")]),e._v(" "),a("div",{staticClass:"language- extra-class"},[a("pre",{pre:!0,attrs:{class:"language-text"}},[a("code",[e._v('@SpringBootApplication\npublic class Application {\n\n public static void main(String[] args) {\n SpringApplication.run(Application.class, args);\n }\n\n @Bean\n public ConsumerFactory kafkaConsumerFactory(SomeBean someBean) {\n Map