From 4480301a63e680ec0aef7ac246461ee321ef4692 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 17 Dec 2019 11:58:17 -0500 Subject: [PATCH] Upgrade to 2.4 clients - Remove references to ZkClient - deprecation of `Consumer.committed(TopicPartition tp)` - Addition of `onPartitionsLost` to `ConsumerRebalanceListener` - since `onPartitionsLost` calls `onPartitionsRevoked` by default with a possible partial list, remove the revoked partitions from the assignments - fix some reflection uses in tests (`groupId` field and `NewTopic`) - fix `EKIT` `listen12` listener - we now always get the full batch - `ConsumerRebalanceListener.onPartitionsRevoked` is no longer called with an empty collection - deprecation of some `TopologyDriver` helper classes - update what's new and change log appendix - add `unregisterSeekCallback` to `ConsumerSeekAware` - `AbstractCSA` now removes TL there since we may have only a partial revoke * Fix what's new * GH-1277: isAckAfterHandle default true Resolves: https://github.com/spring-projects/spring-kafka/issues/1277 * Fix SBFB Javadocs * Update version * Fix test for isAckAfterHandle default true * Upgrade to 2.4.0 Clients * Remove mavenLocal() repo --- build.gradle | 2 +- gradle.properties | 2 +- .../kafka/test/EmbeddedKafkaBroker.java | 36 +--- .../config/StreamsBuilderFactoryBean.java | 2 +- .../ReactiveKafkaConsumerTemplate.java | 12 ++ .../listener/AbstractConsumerSeekAware.java | 4 + .../AbstractMessageListenerContainer.java | 6 + .../ConsumerAwareRebalanceListener.java | 15 ++ .../kafka/listener/ConsumerSeekAware.java | 9 + .../kafka/listener/GenericErrorHandler.java | 3 +- .../KafkaMessageListenerContainer.java | 17 +- .../EnableKafkaIntegrationTests.java | 43 +++-- .../kafka/core/KafkaAdminTests.java | 28 ++- ...ncurrentMessageListenerContainerTests.java | 17 +- .../kafka/streams/HeaderEnricherTests.java | 14 +- .../messaging/MessagingTransformerTests.java | 21 ++- src/reference/asciidoc/appendix.adoc | 35 ++-- src/reference/asciidoc/changes-since-1.0.adoc | 160 +++++++++++++++- src/reference/asciidoc/kafka.adoc | 15 +- src/reference/asciidoc/whats-new.adoc | 173 ++---------------- 20 files changed, 356 insertions(+), 258 deletions(-) diff --git a/build.gradle b/build.gradle index 3591c1c8..709df14c 100644 --- a/build.gradle +++ b/build.gradle @@ -45,7 +45,7 @@ ext { jaywayJsonPathVersion = '2.4.0' junit4Version = '4.12' junitJupiterVersion = '5.5.2' - kafkaVersion = '2.3.1' + kafkaVersion = '2.4.0' log4jVersion = '2.12.1' micrometerVersion = '1.3.2' mockitoVersion = '3.0.0' diff --git a/gradle.properties b/gradle.properties index 35419b1c..663c9c0c 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1 +1 @@ -version=2.3.5.BUILD-SNAPSHOT +version=2.4.0.BUILD-SNAPSHOT diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java index 81e7801e..fa6c02ce 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java @@ -34,8 +34,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; -import org.I0Itec.zkclient.ZkClient; -import org.I0Itec.zkclient.exception.ZkInterruptedException; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.AdminClientConfig; @@ -66,7 +64,6 @@ import kafka.server.KafkaServer; import kafka.server.NotRunning; import kafka.utils.CoreUtils; import kafka.utils.TestUtils; -import kafka.utils.ZKStringSerializer$; import kafka.zk.ZkFourLetterWords; import kafka.zookeeper.ZooKeeperClient; @@ -133,8 +130,6 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { private volatile ZooKeeperClient zooKeeperClient; - private volatile ZkClient zkClient; - public EmbeddedKafkaBroker(int count) { this(count, false); } @@ -298,7 +293,8 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { scala.Option.apply(null), scala.Option.apply(null), scala.Option.apply(null), - true, false, 0, false, 0, false, 0, scala.Option.apply(null), 1, false); + true, false, 0, false, 0, false, 0, scala.Option.apply(null), 1, false, + this.partitionsPerTopic, (short) this.count); } /** @@ -395,19 +391,11 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { // do nothing } } - try { - synchronized (this) { - if (this.zooKeeperClient != null) { - this.zooKeeperClient.close(); - } - if (this.zkClient != null) { - this.zkClient.close(); - } + synchronized (this) { + if (this.zooKeeperClient != null) { + this.zooKeeperClient.close(); } } - catch (ZkInterruptedException e) { - // do nothing - } try { this.zookeeper.shutdown(); this.zkConnect = null; @@ -433,20 +421,6 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { return this.zookeeper; } - /** - * Return the ZkClient. - * @return the client. - * @deprecated in favor of {@link #getZooKeeperClient()}. - */ - @Deprecated - public synchronized ZkClient getZkClient() { - if (this.zkClient == null) { - this.zkClient = new ZkClient(this.zkConnect, ZK_SESSION_TIMEOUT, ZK_CONNECTION_TIMEOUT, - ZKStringSerializer$.MODULE$); - } - return this.zkClient; - } - /** * Return the ZooKeeperClient. * @return the client. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java b/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java index 9e38c09e..e479b500 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java @@ -126,7 +126,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean { return doOnConsumer(consumer -> consumer.position(partition)); } + /** + * Get the committed {@link OffsetAndMetadata} for the partition. + * @param partition the partition. + * @return the {@link OffsetAndMetadata}. + * @deprecated in favor of {@link #committed(Set)}. + */ + @Deprecated + @SuppressWarnings("deprecation") public Mono committed(TopicPartition partition) { return doOnConsumer(consumer -> consumer.committed(partition)); } + public Mono> committed(Set partitions) { + return doOnConsumer(consumer -> consumer.committed(partitions)); + } + public Flux partitionsFromConsumerFor(String topic) { Mono> partitions = doOnConsumer(c -> c.partitionsFor(topic)); return partitions.flatMapIterable(Function.identity()); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java index 0bfc5079..47845e75 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java @@ -56,6 +56,10 @@ public abstract class AbstractConsumerSeekAware implements ConsumerSeekAware { @Override public void onPartitionsRevoked(Collection partitions) { partitions.forEach(tp -> this.callbacks.remove(tp)); + } + + @Override + public void unregisterSeekCallback() { this.callbackForThread.remove(); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java index b4abb8bb..9127f668 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java @@ -465,6 +465,12 @@ public abstract class AbstractMessageListenerContainer getGroupId() + ": partitions assigned: " + partitions); } + @Override + public void onPartitionsLost(Collection partitions) { + AbstractMessageListenerContainer.this.logger.info(() -> + getGroupId() + ": partitions lost: " + partitions); + } + }; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListener.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListener.java index 8d70594e..073b4a37 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListener.java @@ -53,6 +53,16 @@ public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListene // do nothing } + /** + * The same as {@link #onPartitionsLost(Collection)} with an additional consumer parameter. + * @param consumer the consumer. + * @param partitions the partitions. + * @since 2.4 + */ + default void onPartitionsLost(Consumer consumer, Collection partitions) { + // do nothing + } + /** * The same as {@link #onPartitionsAssigned(Collection)} with the additional consumer * parameter. @@ -73,4 +83,9 @@ public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListene throw new UnsupportedOperationException("Listener container should never call this"); } + @Override + default void onPartitionsLost(Collection partitions) { + throw new UnsupportedOperationException("Listener container should never call this"); + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java index 3ced6ff8..18779ec2 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java @@ -72,6 +72,15 @@ public interface ConsumerSeekAware { // do nothing } + /** + * Called when the listener consumer terminates allowing implementations to clean up + * state, such as thread locals. + * @since 2.4 + */ + default void unregisterSeekCallback() { + // do nothing + } + /** * A callback that a listener can invoke to seek to a specific offset. */ diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java index 1ef680be..da68ae75 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java @@ -63,8 +63,7 @@ public interface GenericErrorHandler { * @since 2.3.2 */ default boolean isAckAfterHandle() { - // TODO: Default true in the next release. - return false; + return true; } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 100c8fa0..4aa08185 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -1111,6 +1111,7 @@ public class KafkaMessageListenerContainer // NOSONAR line count } if (this.consumerSeekAwareListener != null) { this.consumerSeekAwareListener.onPartitionsRevoked(partitions); + this.consumerSeekAwareListener.unregisterSeekCallback(); } this.logger.info(() -> getGroupId() + ": Consumer stopped"); publishConsumerStoppedEvent(); @@ -2181,6 +2182,9 @@ public class KafkaMessageListenerContainer // NOSONAR line count if (ListenerConsumer.this.consumerSeekAwareListener != null) { ListenerConsumer.this.consumerSeekAwareListener.onPartitionsRevoked(partitions); } + if (ListenerConsumer.this.assignedPartitions != null) { + ListenerConsumer.this.assignedPartitions.removeAll(partitions); + } } finally { if (ListenerConsumer.this.kafkaTxManager != null) { @@ -2196,7 +2200,7 @@ public class KafkaMessageListenerContainer // NOSONAR line count ListenerConsumer.this.logger.warn("Paused consumer resumed by Kafka due to rebalance; " + "consumer paused again, so the initial poll() will never return any records"); } - ListenerConsumer.this.assignedPartitions = partitions; + ListenerConsumer.this.assignedPartitions = new LinkedList<>(partitions); if (!ListenerConsumer.this.autoCommit) { // Commit initial positions - this is generally redundant but // it protects us from the case when another consumer starts @@ -2271,6 +2275,17 @@ public class KafkaMessageListenerContainer // NOSONAR line count } } + @Override + public void onPartitionsLost(Collection partitions) { + if (this.consumerAwareListener != null) { + this.consumerAwareListener.onPartitionsLost(ListenerConsumer.this.consumer, partitions); + } + else { + this.userListener.onPartitionsLost(partitions); + } + onPartitionsRevoked(partitions); + } + } private final class InitialOrIdleSeekCallback implements ConsumerSeekCallback { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index 3658caca..fcf97a44 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -304,7 +304,7 @@ public class EnableKafkaIntegrationTests { offset = KafkaTestUtils.getPropertyValue(fizContainer, "topicPartitions", TopicPartitionOffset[].class)[3]; assertThat(offset.isRelativeToCurrent()).isTrue(); - assertThat(KafkaTestUtils.getPropertyValue(fizContainer, "listenerConsumer.consumer.coordinator.groupId")) + assertThat(KafkaTestUtils.getPropertyValue(fizContainer, "listenerConsumer.consumer.groupId")) .isEqualTo("fiz"); assertThat(KafkaTestUtils.getPropertyValue(fizContainer, "listenerConsumer.consumer.clientId")) .isEqualTo("clientIdViaAnnotation-0"); @@ -329,7 +329,7 @@ public class EnableKafkaIntegrationTests { MessageListenerContainer rebalanceContainer = (MessageListenerContainer) KafkaTestUtils .getPropertyValue(rebalanceConcurrentContainer, "containers", List.class).get(0); - assertThat(KafkaTestUtils.getPropertyValue(rebalanceContainer, "listenerConsumer.consumer.coordinator.groupId")) + assertThat(KafkaTestUtils.getPropertyValue(rebalanceContainer, "listenerConsumer.consumer.groupId")) .isNotEqualTo("rebalanceListener"); String clientId = KafkaTestUtils.getPropertyValue(rebalanceContainer, "listenerConsumer.consumer.clientId", String.class); @@ -424,7 +424,7 @@ public class EnableKafkaIntegrationTests { assertThat(buzConcurrentContainer).isNotNull(); MessageListenerContainer buzContainer = (MessageListenerContainer) KafkaTestUtils .getPropertyValue(buzConcurrentContainer, "containers", List.class).get(0); - assertThat(KafkaTestUtils.getPropertyValue(buzContainer, "listenerConsumer.consumer.coordinator.groupId")) + assertThat(KafkaTestUtils.getPropertyValue(buzContainer, "listenerConsumer.consumer.groupId")) .isEqualTo("buz.explicitGroupId"); } @@ -512,6 +512,7 @@ public class EnableKafkaIntegrationTests { assertThat(this.listener.listen12Consumer).isSameAs(KafkaTestUtils.getPropertyValue(KafkaTestUtils .getPropertyValue(this.registry.getListenerContainer("list3"), "containers", List.class).get(0), "listenerConsumer.consumer")); + assertThat(this.config.listen12Latch.await(10, TimeUnit.SECONDS)).isNotNull(); assertThat(this.config.listen12Exception).isNotNull(); assertThat(this.config.listen12Message.getPayload()).isInstanceOf(List.class); List errorPayload = (List) this.config.listen12Message.getPayload(); @@ -1307,12 +1308,15 @@ public class EnableKafkaIntegrationTests { private Message listen12Message; + private final CountDownLatch listen12Latch = new CountDownLatch(1); + @Bean public ConsumerAwareListenerErrorHandler listen12ErrorHandler() { return (m, e, c) -> { this.listen12Exception = e; this.listen12Message = m; resetAllOffsets(m, c); + this.listen12Latch.countDown(); return null; }; } @@ -1651,12 +1655,12 @@ public class EnableKafkaIntegrationTests { @KafkaListener(id = "list3", topics = "annotated16", containerFactory = "batchFactory", errorHandler = "listen12ErrorHandler") public void listen12(List> list, Consumer consumer) { - if (this.reposition12.compareAndSet(false, true)) { - throw new RuntimeException("reposition"); - } this.payload = list; this.listen12Consumer = consumer; this.latch12.countDown(); + if (this.reposition12.compareAndSet(false, true)) { + throw new RuntimeException("reposition"); + } } @KafkaListener(id = "list4", topics = "annotated17", containerFactory = "batchManualFactory") @@ -1729,19 +1733,28 @@ public class EnableKafkaIntegrationTests { @KafkaListener(id = "batchAckListener", topics = { "annotated26", "annotated27" }, containerFactory = "batchFactory") - public void batchAckListener(List in, - @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List partitions, - @Header(KafkaHeaders.RECEIVED_TOPIC) List topics, + public void batchAckListener(@SuppressWarnings("unused") List in, + @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List partitionsHeader, + @Header(KafkaHeaders.RECEIVED_TOPIC) List topicsHeader, Consumer consumer) { - for (int i = 0; i < topics.size(); i++) { + + for (int i = 0; i < topicsHeader.size(); i++) { this.latch17.countDown(); - String topic = topics.get(i); - if ("annotated26".equals(topic) && consumer.committed( - new org.apache.kafka.common.TopicPartition(topic, partitions.get(i))).offset() == 1) { + String inTopic = topicsHeader.get(i); + if ("annotated26".equals(inTopic) && consumer.committed(Collections.singleton( + new org.apache.kafka.common.TopicPartition(inTopic, partitionsHeader.get(i)))) + .values() + .iterator() + .next() + .offset() == 1) { this.latch18.countDown(); } - else if ("annotated27".equals(topic) && consumer.committed( - new org.apache.kafka.common.TopicPartition(topic, partitions.get(i))).offset() == 3) { + else if ("annotated27".equals(inTopic) && consumer.committed(Collections.singleton( + new org.apache.kafka.common.TopicPartition(inTopic, partitionsHeader.get(i)))) + .values() + .iterator() + .next() + .offset() == 3) { this.latch18.countDown(); } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java index 04254163..b04aedb3 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java @@ -24,6 +24,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Optional; import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.admin.AdminClient; @@ -82,16 +83,29 @@ public class KafkaAdminTests { } @Test - public void testAddTopics() throws Exception { + public void testAddTopicsAndAddPartitions() throws Exception { AdminClient adminClient = AdminClient.create(this.admin.getConfig()); DescribeTopicsResult topics = adminClient.describeTopics(Arrays.asList("foo", "bar")); - topics.all().get(); - new DirectFieldAccessor(this.topic1).setPropertyValue("numPartitions", 2); - new DirectFieldAccessor(this.topic2).setPropertyValue("numPartitions", 3); - this.admin.initialize(); - topics = adminClient.describeTopics(Arrays.asList("foo", "bar")); Map results = topics.all().get(); - results.forEach((name, td) -> assertThat(td.partitions()).hasSize(name.equals("foo") ? 2 : 3)); + results.forEach((name, td) -> assertThat(td.partitions()).hasSize(name.equals("foo") ? 2 : 1)); + new DirectFieldAccessor(this.topic1).setPropertyValue("numPartitions", Optional.of(4)); + new DirectFieldAccessor(this.topic2).setPropertyValue("numPartitions", Optional.of(3)); + this.admin.initialize(); + int n = 0; + TopicDescription bar = results.values().stream() + .filter(td -> td.name().equals("bar")) + .findFirst() + .get(); + while (n++ < 100 && bar.partitions().size() == 1) { + Thread.sleep(100); + topics = adminClient.describeTopics(Arrays.asList("foo", "bar")); + results = topics.all().get(); + bar = results.values().stream() + .filter(tp -> tp.name().equals("bar")) + .findFirst() + .get(); + } + results.forEach((name, td) -> assertThat(td.partitions()).hasSize(name.equals("foo") ? 4 : 3)); adminClient.close(Duration.ofSeconds(10)); } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index 746ceab5..739935ea 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java @@ -42,6 +42,7 @@ import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.TopicPartition; @@ -238,13 +239,11 @@ public class ConcurrentMessageListenerContainerTests { consumerProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); containerProps.setKafkaConsumerProperties(consumerProperties); final CountDownLatch rebalancePartitionsAssignedLatch = new CountDownLatch(2); - final CountDownLatch rebalancePartitionsRevokedLatch = new CountDownLatch(2); containerProps.setConsumerRebalanceListener(new ConsumerRebalanceListener() { @Override public void onPartitionsRevoked(Collection partitions) { ConcurrentMessageListenerContainerTests.this.logger.info("In test, partitions revoked:" + partitions); - rebalancePartitionsRevokedLatch.countDown(); } @Override @@ -274,7 +273,6 @@ public class ConcurrentMessageListenerContainerTests { template.flush(); assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); assertThat(rebalancePartitionsAssignedLatch.await(60, TimeUnit.SECONDS)).isTrue(); - assertThat(rebalancePartitionsRevokedLatch.await(60, TimeUnit.SECONDS)).isTrue(); for (String threadName : listenerThreadNames) { assertThat(threadName).contains("-C-"); } @@ -619,6 +617,19 @@ public class ConcurrentMessageListenerContainerTests { containerProps); container.setConcurrency(2); container.setBeanName("testAckOnError"); + container.setErrorHandler(new LoggingErrorHandler() { + + @Override + public void handle(Exception thrownException, ConsumerRecord record) { + // nothing + } + + @Override + public boolean isAckAfterHandle() { + return false; + } + + }); container.start(); ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/streams/HeaderEnricherTests.java b/spring-kafka/src/test/java/org/springframework/kafka/streams/HeaderEnricherTests.java index b5ad4c5a..8199b76b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/streams/HeaderEnricherTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/streams/HeaderEnricherTests.java @@ -22,13 +22,15 @@ import java.util.HashMap; import java.util.Map; import java.util.Properties; -import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.TestInputTopic; +import org.apache.kafka.streams.TestOutputTopic; import org.apache.kafka.streams.TopologyTestDriver; import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.test.ConsumerRecordFactory; +import org.apache.kafka.streams.test.TestRecord; import org.junit.jupiter.api.Test; import org.springframework.expression.Expression; @@ -64,10 +66,12 @@ public class HeaderEnricherTests { config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); TopologyTestDriver driver = new TopologyTestDriver(builder.build(), config); - ConsumerRecordFactory recordFactory = new ConsumerRecordFactory<>(new StringSerializer(), + TestInputTopic inputTopic = driver.createInputTopic(INPUT, new StringSerializer(), new StringSerializer()); - driver.pipeInput(recordFactory.create(INPUT, "key", "value")); - ProducerRecord result = driver.readOutput(OUTPUT); + inputTopic.pipeInput("key", "value"); + TestOutputTopic outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), + new StringDeserializer()); + TestRecord result = outputTopic.readRecord(); assertThat(result.headers().lastHeader("foo")).isNotNull(); assertThat(result.headers().lastHeader("foo").value()).isEqualTo("bar".getBytes()); assertThat(result.headers().lastHeader("spel")).isNotNull(); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/streams/messaging/MessagingTransformerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/streams/messaging/MessagingTransformerTests.java index 5ca06875..eb214173 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/streams/messaging/MessagingTransformerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/streams/messaging/MessagingTransformerTests.java @@ -18,17 +18,21 @@ package org.springframework.kafka.streams.messaging; import static org.assertj.core.api.Assertions.assertThat; +import java.util.Collections; import java.util.Properties; -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.TestInputTopic; +import org.apache.kafka.streams.TestOutputTopic; import org.apache.kafka.streams.TopologyTestDriver; import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.test.ConsumerRecordFactory; +import org.apache.kafka.streams.test.TestRecord; import org.junit.jupiter.api.Test; import org.springframework.kafka.support.SimpleKafkaHeaderMapper; @@ -67,12 +71,13 @@ public class MessagingTransformerTests { config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); TopologyTestDriver driver = new TopologyTestDriver(builder.build(), config); - ConsumerRecordFactory recordFactory = new ConsumerRecordFactory<>(new StringSerializer(), + TestInputTopic inputTopic = driver.createInputTopic(INPUT, new StringSerializer(), new StringSerializer()); - ConsumerRecord consumerRecord = recordFactory.create(INPUT, "key", "value"); - consumerRecord.headers().add(new RecordHeader("fiz", "buz".getBytes())); - driver.pipeInput(consumerRecord); - ProducerRecord result = driver.readOutput(OUTPUT); + Headers headers = new RecordHeaders(Collections.singletonList(new RecordHeader("fiz", "buz".getBytes()))); + inputTopic.pipeInput(new TestRecord<>("key", "value", headers)); + TestOutputTopic outputTopic = driver.createOutputTopic(OUTPUT, new ByteArrayDeserializer(), + new ByteArrayDeserializer()); + TestRecord result = outputTopic.readRecord(); assertThat(result.value()).isEqualTo("bar".getBytes()); assertThat(result.headers().lastHeader("fiz").value()).isEqualTo("buz".getBytes()); assertThat(result.headers().lastHeader("baz").value()).isEqualTo("qux".getBytes()); diff --git a/src/reference/asciidoc/appendix.adoc b/src/reference/asciidoc/appendix.adoc index ca2d6e98..02817273 100644 --- a/src/reference/asciidoc/appendix.adoc +++ b/src/reference/asciidoc/appendix.adoc @@ -1,8 +1,7 @@ -//// -[[deps-for-21x]] -== Override Dependencies to use the 2.1.x kafka-clients with an Embedded Broker +[[deps-for-24x]] +== Override Spring Boot Dependencies to use Spring for Apache Kafka 2.4 -When you use `spring-kafka-test` (version 2.2.x) with the 2.1.x `kafka-clients` jar, you need to override certain transitive dependencies, as follows: +When you use `spring-kafka-test` (version 2.4.x) with Spring Boot, you need to override certain Boot managed dependencies, as follows: **maven** @@ -19,39 +18,34 @@ When you use `spring-kafka-test` (version 2.2.x) with the 2.1.x `kafka-clients` org.springframework.kafka spring-kafka-test {project-version} - - - org.apache.kafka - kafka_2.11 - - test org.apache.kafka kafka-clients - 2.1.1 + 2.4.0 org.apache.kafka kafka-clients - 2.1.1 + 2.4.0 test - - - - org.apache.kafka - kafka_2.12 - 2.1.1 test org.apache.kafka kafka_2.12 - 2.1.1 + 2.4.0 + test + + + + org.apache.kafka + kafka_2.12 + 2.4.0 test test @@ -79,7 +73,8 @@ dependencies { ---- ==== -Note that when switching to scala 2.12 (recommended for 2.1.x and higher), the 2.11 version must be excluded from spring-kafka-test. +The test scope dependencies are only needed if you are using the embedded Kafka broker in tests. + //// [appendix] diff --git a/src/reference/asciidoc/changes-since-1.0.adoc b/src/reference/asciidoc/changes-since-1.0.adoc index bebe48d3..9024c620 100644 --- a/src/reference/asciidoc/changes-since-1.0.adoc +++ b/src/reference/asciidoc/changes-since-1.0.adoc @@ -1,5 +1,157 @@ [[migration]] -=== Changes between 2.1 and 2.2 +=== Changes Between 2.2 and 2.3 + +This section covers the changes made from version 2.2 to version 2.3. + +Also see <>. + +==== Tips, Tricks and Examples + +A new chapter <> has been added. +Please submit GitHub issues and/or pull requests for additional entries in that chapter. + +[[kafka-client-2.2]] +==== Kafka Client Version + +This version requires the 2.3.0 `kafka-clients` or higher. + +==== Class/Package Changes + +`TopicPartitionInitialOffset` is deprecated in favor of `TopicPartitionOffset`. + +==== Configuration Changes + +Starting with version 2.3.4, the `missingTopicsFatal` container property is false by default. +When this is true, the application fails to start if the broker is down; many users were affected by this change; given that Kafka is a high-availability platform, we did not anticipate that starting an application with no active brokers would be a common use case. + +==== Producer and Consumer Factory Changes + +The `DefaultKafkaProducerFactory` can now be configured to create a producer per thread. +You can also provide `Supplier` instances in the constructor as an alternative to either configured classes (which require no-arg constructors), or constructing with `Serializer` instances, which are then shared between all Producers. +See <> for more information. + +The same option is available with `Supplier` instances in `DefaultKafkaConsumerFactory`. +See <> for more information. + +==== Listener Container Changes + +Previously, error handlers received `ListenerExecutionFailedException` (with the actual listener exception as the `cause`) when the listener was invoked using a listener adapter (such as `@KafkaListener` s). +Exceptions thrown by native `GenericMessageListener` s were passed to the error handler unchanged. +Now a `ListenerExecutionFailedException` is always the argument (with the actual listener exception as the `cause`), which provides access to the container's `group.id` property. + +Because the listener container has it's own mechanism for committing offsets, it prefers the Kafka `ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG` to be `false`. +It now sets it to false automatically unless specifically set in the consumer factory or the container's consumer property overrides. + +The `ackOnError` property is now `false` by default. +See <> for more information. + +It is now possible to obtain the consumer's `group.id` property in the listener method. +See <> for more information. + +The container has a new property `recordInterceptor` allowing records to be inspected or modified before invoking the listener. +A `CompositeRecordInterceptor` is also provided in case you need to invoke multiple interceptors. +See <> for more information. + +The `ConsumerSeekAware` has new methods allowing you to perform seeks relative to the beginning, end, or current position and to seek to the first offset greater than or equal to a time stamp. +See <> for more information. + +A convenience class `AbstractConsumerSeekAware` is now provided to simplify seeking. +See <> for more information. + +The `ContainerProperties` provides an `idleBetweenPolls` option to let the main loop in the listener container to sleep between `KafkaConsumer.poll()` calls. +See its JavaDocs and <> for more information. + +When using `AckMode.MANUAL` (or `MANUAL_IMMEDIATE`) you can now cause a redelivery by calling `nack` on the `Acknowledgment`. +See <> for more information. + +Listener performance can now be monitored using Micrometer `Timer` s. +See <> for more information. + +The containers now publish additional consumer lifecyle events relating to startup. +See <> for more information. + +Transactional batch listeners can now support zombie fencing. +See <> for more information. + +The listener container factory can now be configured with a `ContainerCustomizer` to further configure each container after it has been created and configured. +See <> for more information. + +==== ErrorHandler Changes + +The `SeekToCurrentErrorHandler` now treats certain exceptions as fatal and disables retry for those, invoking the recoverer on first failure. + +The `SeekToCurrentErrorHandler` and `SeekToCurrentBatchErrorHandler` can now be configured to apply a `BackOff` (thread sleep) between delivery attempts. + +Starting with version 2.3.2, recovered records' offsets will be committed when the error handler returns after recovering a failed record. + +See <> for more information. + +The `DeadLetterPublishingRecoverer`, when used in conjunction with an `ErrorHandlingDeserializer2`, now sets the payload of the message sent to the dead-letter topic, to the original value that could not be deserialized. +Previously, it was `null` and user code needed to extract the `DeserializationException` from the message headers. +See <> for more information. + +==== TopicBuilder + +A new class `TopicBuilder` is provided for more convenient creation of `NewTopic` `@Bean` s for automatic topic provisioning. +See <> for more information. + +==== Kafka Streams Changes + +You can now perform additional configuration of the `StreamsBuilderFactoryBean` created by `@EnableKafkaStreams`. +See <> for more information. + +A `RecoveringDeserializationExceptionHandler` is now provided which allows records with deserialization errors to be recovered. +It can be used in conjunction with a `DeadLetterPublishingRecoverer` to send these records to a dead-letter topic. +See <> for more information. + +The `HeaderEnricher` transformer has been provided, using SpEL to generate the header values. +See <> for more information. + +The `MessagingTransformer` has been provided. +This allows a Kafka streams topology to interact with a spring-messaging component, such as a Spring Integration flow. +See <> and <> for more information. + +==== JSON Component Changes + +Now all the JSON-aware components are configured by default with a Jackson `ObjectMapper` produced by the `JacksonUtils.enhancedObjectMapper()`. +The `JsonDeserializer` now provides `TypeReference`-based constructors for better handling of target generic container types. +Also a `JacksonMimeTypeModule` has been introduced for serialization of `org.springframework.util.MimeType` to plain string. +See its JavaDocs and <> for more information. + +A `ByteArrayJsonMessageConverter` has been provided as well as a new super class for all Json converters, `JsonMessageConverter`. +Also, a `StringOrBytesSerializer` is now available; it can serialize `byte[]`, `Bytes` and `String` values in `ProducerRecord` s. +See <> for more information. + +The `JsonSerializer`, `JsonDeserializer` and `JsonSerde` now have fluent APIs to make programmatic configuration simpler. +See the javadocs, <>, and <> for more informaion. + +==== ReplyingKafkaTemplate + +When a reply times out, the future is completed exceptionally with a `KafkaReplyTimeoutException` instead of a `KafkaException`. + +Also, an overloaded `sendAndReceive` method is now provided that allows specifying the reply timeout on a per message basis. + +==== AggregatingReplyingKafkaTemplate + +Extends the `ReplyingKafkaTemplate` by aggregating replies from multiple receivers. +See <> for more information. + +==== Transaction Changes + +You can now override the producer factory's `transactionIdPrefix` on the `KafkaTemplate` and `KafkaTransactionManager`. +See <> for more information. + +==== New Delegating Serializer/Deserializer + +The framework now provides a delegating `Serializer` and `Deserializer`, utilizing a header to enable producing and consuming records with multiple key/value types. +See <> for more information. + +==== New Retrying Deserializer + +The framework now provides a delegating `RetryingDeserializer`, to retry serialization when transient errors such as network problems might occur. +See <> for more information. + +=== Changes Between 2.1 and 2.2 [[kafka-client-2.0]] ==== Kafka Client Version @@ -110,7 +262,7 @@ When a transaction is started by the listener container, the `transactional.id` This change allows proper fencing of zombies, https://www.confluent.io/blog/transactions-apache-kafka/[as described here]. -=== Changes between 2.0 and 2.1 +=== Changes Between 2.0 and 2.1 [[kafka-client-1.0]] ==== Kafka Client Version @@ -247,11 +399,11 @@ Support for configuring Kerberos is now provided. See <> for more information. -=== Changes between 1.1 and 1.2 +=== Changes Between 1.1 and 1.2 This version uses the 0.10.2.x client. -=== Changes between 1.0 and 1.1 +=== Changes Between 1.0 and 1.1 ==== Kafka Client diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 7093cdbe..e6c31ef3 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -1496,6 +1496,8 @@ public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListene void onPartitionsAssigned(Consumer consumer, Collection partitions); + void onPartitionsLost(Consumer consumer, Collection partitions); + } ---- ==== @@ -1532,6 +1534,14 @@ containerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListe ---- ==== +IMPORTANT: Starting with version 2.4, a new method `onPartitionsLost()` has been added (similar to a method with the same name in `ConsumerRebalanceLister`). +The default implementation on `ConsumerRebalanceLister` simply calls `onPartionsRevoked`. +The default implementation on `ConsumerAwareRebalanceListener` does nothing. +When supplying the listener container with a custom listener (of either type), it is important that your implementation not call `onPartitionsRevoked` from `onPartitionsLost`. +If you implement `ConsumerRebalanceListener` you should override the default method. +This is because the listener container will call its own `onPartitionsRevoked` from its implementation of `onPartitionsLost` after calling the method on your implementation. +If you implementation delegates to the default behavior, `onPartitionsRevoked` will be called twice each time the `Consumer` calls that method on the container's listener. + [[annotation-send-to]] ===== Forwarding Listener Results using `@SendTo` @@ -3445,10 +3455,7 @@ When transactions are being used, no error handlers are configured, by default, If you provide a custom error handler when using transactions, it must throw an exception if you want the transaction rolled back. Starting with version 2.3.2, these interfaces have a default method `isAckAfterHandle()` which is called by the container to determine whether the offset(s) should be committed if the error handler returns without throwing an exception. -This returns false by default, for backwards compatibility. -In most cases, however, we expect that the offset should be committed. -For example, the <> returns `true` if a record is recovered (after any retries, if so configured). -In a future release, we expect to change this default to `true`. +Starting with version 2.4, this returns true by default. You can specify a global error handler to be used for all listeners in the container factory. The following example shows how to do so: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index c796f1cf..9c2e7690 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -1,171 +1,34 @@ -=== What's New in 2.3 Since 2.2 +=== What's New in 2.4 Since 2.3 -This section covers the changes made from version 2.2 to version 2.3. +This section covers the changes made from version 2.3 to version 2.4. Also see <>. -==== Tips, Tricks and Examples - -A new chapter <> has been added. -Please submit GitHub issues and/or pull requests for additional entries in that chapter. - -[[kafka-client-2.2]] +[[kafka-client-2.4]] ==== Kafka Client Version -This version requires the 2.3.0 `kafka-clients` or higher. +This version requires the 2.4.0 `kafka-clients` or higher. -==== Class/Package Changes +[[x24-carl]] +==== ConsumerAwareRabalanceListener -`TopicPartitionInitialOffset` is deprecated in favor of `TopicPartitionOffset`. +Like `ConsumerRebalanceListener`, this interface now has an additional method `onPartitionsLost`. +Refer to the Apache Kafka documentation for more information. -==== Configuration Changes +Unlike the `ConsumerRebalanceListener`, The default implementation does **not** call `onPartitionsRevoked`. +Instead, the listener container will call that method after it has called `onPartitionsLost`; you should not, therefore, do the same when implementing `ConsumerAwareRabalanceListener`. -Starting with version 2.3.4, the `missingTopicsFatal` container property is false by default. -When this is true, the application fails to start if the broker is down; many users were affected by this change; given that Kafka is a high-availability platform, we did not anticipate that starting an application with no active brokers would be a common use case. +See the IMPORTANT note at the end of <> for more information. -==== Producer and Consumer Factory Changes +[[x24-eh]] +==== GenericErrorHandler -The `DefaultKafkaProducerFactory` can now be configured to create a producer per thread. -You can also provide `Supplier` instances in the constructor as an alternative to either configured classes (which require no-arg constructors), or constructing with `Serializer` instances, which are then shared between all Producers. -See <> for more information. +The `isAckAfterHandle()` default implementation now returns true by default. -The same option is available with `Supplier` instances in `DefaultKafkaConsumerFactory`. -See <> for more information. +=== Migration Guide -==== Listener Container Changes +* This release is essentially the same as the 2.3.x line, except it has been compiled against the 2.4 `kafka-clients` jar, due to a binary incompatibility. -Previously, error handlers received `ListenerExecutionFailedException` (with the actual listener exception as the `cause`) when the listener was invoked using a listener adapter (such as `@KafkaListener` s). -Exceptions thrown by native `GenericMessageListener` s were passed to the error handler unchanged. -Now a `ListenerExecutionFailedException` is always the argument (with the actual listener exception as the `cause`), which provides access to the container's `group.id` property. +* To use Spring for Apache Kafka 2.4.x with Spring Boot 2.2.x, see <>. -Because the listener container has it's own mechanism for committing offsets, it prefers the Kafka `ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG` to be `false`. -It now sets it to false automatically unless specifically set in the consumer factory or the container's consumer property overrides. - -The `ackOnError` property is now `false` by default. -See <> for more information. - -It is now possible to obtain the consumer's `group.id` property in the listener method. -See <> for more information. - -The container has a new property `recordInterceptor` allowing records to be inspected or modified before invoking the listener. -A `CompositeRecordInterceptor` is also provided in case you need to invoke multiple interceptors. -See <> for more information. - -The `ConsumerSeekAware` has new methods allowing you to perform seeks relative to the beginning, end, or current position and to seek to the first offset greater than or equal to a time stamp. -See <> for more information. - -A convenience class `AbstractConsumerSeekAware` is now provided to simplify seeking. -See <> for more information. - -The `ContainerProperties` provides an `idleBetweenPolls` option to let the main loop in the listener container to sleep between `KafkaConsumer.poll()` calls. -See its JavaDocs and <> for more information. - -When using `AckMode.MANUAL` (or `MANUAL_IMMEDIATE`) you can now cause a redelivery by calling `nack` on the `Acknowledgment`. -See <> for more information. - -Listener performance can now be monitored using Micrometer `Timer` s. -See <> for more information. - -The containers now publish additional consumer lifecyle events relating to startup. -See <> for more information. - -Transactional batch listeners can now support zombie fencing. -See <> for more information. - -The listener container factory can now be configured with a `ContainerCustomizer` to further configure each container after it has been created and configured. -See <> for more information. - -==== ErrorHandler Changes - -The `SeekToCurrentErrorHandler` now treats certain exceptions as fatal and disables retry for those, invoking the recoverer on first failure. - -The `SeekToCurrentErrorHandler` and `SeekToCurrentBatchErrorHandler` can now be configured to apply a `BackOff` (thread sleep) between delivery attempts. - -Starting with version 2.3.2, recovered records' offsets will be committed when the error handler returns after recovering a failed record. - -See <> for more information. - -The `DeadLetterPublishingRecoverer`, when used in conjunction with an `ErrorHandlingDeserializer2`, now sets the payload of the message sent to the dead-letter topic, to the original value that could not be deserialized. -Previously, it was `null` and user code needed to extract the `DeserializationException` from the message headers. -See <> for more information. - -==== TopicBuilder - -A new class `TopicBuilder` is provided for more convenient creation of `NewTopic` `@Bean` s for automatic topic provisioning. -See <> for more information. - -==== Kafka Streams Changes - -You can now perform additional configuration of the `StreamsBuilderFactoryBean` created by `@EnableKafkaStreams`. -See <> for more information. - -A `RecoveringDeserializationExceptionHandler` is now provided which allows records with deserialization errors to be recovered. -It can be used in conjunction with a `DeadLetterPublishingRecoverer` to send these records to a dead-letter topic. -See <> for more information. - -The `HeaderEnricher` transformer has been provided, using SpEL to generate the header values. -See <> for more information. - -The `MessagingTransformer` has been provided. -This allows a Kafka streams topology to interact with a spring-messaging component, such as a Spring Integration flow. -See <> and <> for more information. - -==== JSON Component Changes - -Now all the JSON-aware components are configured by default with a Jackson `ObjectMapper` produced by the `JacksonUtils.enhancedObjectMapper()`. -The `JsonDeserializer` now provides `TypeReference`-based constructors for better handling of target generic container types. -Also a `JacksonMimeTypeModule` has been introduced for serialization of `org.springframework.util.MimeType` to plain string. -See its JavaDocs and <> for more information. - -A `ByteArrayJsonMessageConverter` has been provided as well as a new super class for all Json converters, `JsonMessageConverter`. -Also, a `StringOrBytesSerializer` is now available; it can serialize `byte[]`, `Bytes` and `String` values in `ProducerRecord` s. -See <> for more information. - -The `JsonSerializer`, `JsonDeserializer` and `JsonSerde` now have fluent APIs to make programmatic configuration simpler. -See the javadocs, <>, and <> for more informaion. - -==== ReplyingKafkaTemplate - -When a reply times out, the future is completed exceptionally with a `KafkaReplyTimeoutException` instead of a `KafkaException`. - -Also, an overloaded `sendAndReceive` method is now provided that allows specifying the reply timeout on a per message basis. - -==== AggregatingReplyingKafkaTemplate - -Extends the `ReplyingKafkaTemplate` by aggregating replies from multiple receivers. -See <> for more information. - -==== Transaction Changes - -You can now override the producer factory's `transactionIdPrefix` on the `KafkaTemplate` and `KafkaTransactionManager`. -See <> for more information. - -==== New Delegating Serializer/Deserializer - -The framework now provides a delegating `Serializer` and `Deserializer`, utilizing a header to enable producing and consuming records with multiple key/value types. -See <> for more information. - -==== New Retrying Deserializer - -The framework now provides a delegating `RetryingDeserializer`, to retry serialization when transient errors such as network problems might occur. -See <> for more information. - -==== New function for recovering from deserializing errors - -`ErrorHandlingDeserializer2` now uses a POJO (`FailedDeserializationInfo`) for passing all the contextual information around a deserialization error. -This enables the code to access to extra information that was missing in the old `BiFunction failedDeserializationFunction`. - -==== EmbeddedKafkaBroker Changes - -You can now override the default broker list property name in the annotation. -See <> for more information. - -==== ReplyingKafkaTemplate Changes - -You can now customize the header names for correlation, reply topic and reply partition. -See <> for more information. - -==== Header Mapper Changes - -The `DefaultKafkaHeaderMapper` no longer encodes simple String-valued headers as JSON. -See <> for more information. +* See <> for important information about rebalance listeners.