Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Upgrade to 2.4 clients

Browse Source 
- 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
This commit is contained in:
Gary Russell
2019-12-17 11:58:17 -05:00
committed by Artem Bilan
parent a34f48d554
commit 4480301a63
20 changed files with 356 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

2
gradle.properties
View File

@@ -1 +1 @@
version=2.3.5.BUILD-SNAPSHOT
version=2.4.0.BUILD-SNAPSHOT

36
spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java
View File

@@ -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.

2
spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java
View File

@@ -126,7 +126,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
}
/**
* Set {@link StreamsConfig} on this factory.
* Set the streams configuration {@link Properties} on this factory.
* @param streamsConfig the streams configuration.
* @since 2.2
*/

12
spring-kafka/src/main/java/org/springframework/kafka/core/reactive/ReactiveKafkaConsumerTemplate.java
View File

@@ -146,10 +146,22 @@ public class ReactiveKafkaConsumerTemplate<K, V> {
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<OffsetAndMetadata> committed(TopicPartition partition) {
return doOnConsumer(consumer -> consumer.committed(partition));
}
public Mono<Map<TopicPartition, OffsetAndMetadata>> committed(Set<TopicPartition> partitions) {
return doOnConsumer(consumer -> consumer.committed(partitions));
}
public Flux<PartitionInfo> partitionsFromConsumerFor(String topic) {
Mono<List<PartitionInfo>> partitions = doOnConsumer(c -> c.partitionsFor(topic));
return partitions.flatMapIterable(Function.identity());

4
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java
View File

@@ -56,6 +56,10 @@ public abstract class AbstractConsumerSeekAware implements ConsumerSeekAware {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
partitions.forEach(tp -> this.callbacks.remove(tp));
}
@Override
public void unregisterSeekCallback() {
this.callbackForThread.remove();
}

6
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java
View File

@@ -465,6 +465,12 @@ public abstract class AbstractMessageListenerContainer<K, V>
getGroupId() + ": partitions assigned: " + partitions);
}
@Override
public void onPartitionsLost(Collection<TopicPartition> partitions) {
AbstractMessageListenerContainer.this.logger.info(() ->
getGroupId() + ": partitions lost: " + partitions);
}
};
}

15
spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRebalanceListener.java
View File

@@ -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<TopicPartition> 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<TopicPartition> partitions) {
throw new UnsupportedOperationException("Listener container should never call this");
}
}

9
spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java
View File

@@ -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.
*/

3
spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java
View File

@@ -63,8 +63,7 @@ public interface GenericErrorHandler<T> {
* @since 2.3.2
*/
default boolean isAckAfterHandle() {
// TODO: Default true in the next release.
return false;
return true;
}
}

17
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -1111,6 +1111,7 @@ public class KafkaMessageListenerContainer<K, V> // 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<K, V> // 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<K, V> // 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<K, V> // NOSONAR line count
}
}
@Override
public void onPartitionsLost(Collection<TopicPartition> 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 {

43
spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
View File

@@ -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<ConsumerRecord<Integer, String>> 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<String> in,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
@Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
public void batchAckListener(@SuppressWarnings("unused") List<String> in,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitionsHeader,
@Header(KafkaHeaders.RECEIVED_TOPIC) List<String> 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();
}
}

28
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java
View File

@@ -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<String, TopicDescription> 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));
}

17
spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
View File

@@ -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<TopicPartition> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);

14
spring-kafka/src/test/java/org/springframework/kafka/streams/HeaderEnricherTests.java
View File

@@ -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<String, String> recordFactory = new ConsumerRecordFactory<>(new StringSerializer(),
TestInputTopic<String, String> inputTopic = driver.createInputTopic(INPUT, new StringSerializer(),
new StringSerializer());
driver.pipeInput(recordFactory.create(INPUT, "key", "value"));
ProducerRecord<byte[], byte[]> result = driver.readOutput(OUTPUT);
inputTopic.pipeInput("key", "value");
TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(),
new StringDeserializer());
TestRecord<String, String> result = outputTopic.readRecord();
assertThat(result.headers().lastHeader("foo")).isNotNull();
assertThat(result.headers().lastHeader("foo").value()).isEqualTo("bar".getBytes());
assertThat(result.headers().lastHeader("spel")).isNotNull();

21
spring-kafka/src/test/java/org/springframework/kafka/streams/messaging/MessagingTransformerTests.java
View File

@@ -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<String, String> recordFactory = new ConsumerRecordFactory<>(new StringSerializer(),
TestInputTopic<String, String> inputTopic = driver.createInputTopic(INPUT, new StringSerializer(),
new StringSerializer());
ConsumerRecord<byte[], byte[]> consumerRecord = recordFactory.create(INPUT, "key", "value");
consumerRecord.headers().add(new RecordHeader("fiz", "buz".getBytes()));
driver.pipeInput(consumerRecord);
ProducerRecord<byte[], byte[]> result = driver.readOutput(OUTPUT);
Headers headers = new RecordHeaders(Collections.singletonList(new RecordHeader("fiz", "buz".getBytes())));
inputTopic.pipeInput(new TestRecord<>("key", "value", headers));
TestOutputTopic<byte[], byte[]> outputTopic = driver.createOutputTopic(OUTPUT, new ByteArrayDeserializer(),
new ByteArrayDeserializer());
TestRecord<byte[], byte[]> 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());

35
src/reference/asciidoc/appendix.adoc
View File

@@ -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`
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<version>{project-version}</version>
<exclusions>
<exclusion>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
</exclusion>
</exclusions>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.1.1</version>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.1.1</version>
<version>2.4.0</version>
<classifier>test</classifier>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.12</artifactId>
<version>2.1.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.12</artifactId>
<version>2.1.1</version>
<version>2.4.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.12</artifactId>
<version>2.4.0</version>
<classifier>test</classifier>
<scope>test</scope>
</dependency>
@@ -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]

160
src/reference/asciidoc/changes-since-1.0.adoc
View File

@@ -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 <<new-in-sik>>.
==== Tips, Tricks and Examples
A new chapter <<tips-n-tricks>> 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<Serializer>` 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 <<producer-factory>> for more information.
The same option is available with `Supplier<Deserializer>` instances in `DefaultKafkaConsumerFactory`.
See <<kafka-container>> 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 <<seek-to-current>> for more information.
It is now possible to obtain the consumer's `group.id` property in the listener method.
See <<listener-group-id>> 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 <<message-listener-container>> 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 <<seek>> for more information.
A convenience class `AbstractConsumerSeekAware` is now provided to simplify seeking.
See <<seek>> 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 <<kafka-container>> for more information.
When using `AckMode.MANUAL` (or `MANUAL_IMMEDIATE`) you can now cause a redelivery by calling `nack` on the `Acknowledgment`.
See <<committing-offsets>> for more information.
Listener performance can now be monitored using Micrometer `Timer` s.
See <<micrometer>> for more information.
The containers now publish additional consumer lifecyle events relating to startup.
See <<events>> for more information.
Transactional batch listeners can now support zombie fencing.
See <<transactions>> 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 <<container-factory>> 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 <<seek-to-current>> 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 <<dead-letters>> for more information.
==== TopicBuilder
A new class `TopicBuilder` is provided for more convenient creation of `NewTopic` `@Bean` s for automatic topic provisioning.
See <<configuring-topics>> for more information.
==== Kafka Streams Changes
You can now perform additional configuration of the `StreamsBuilderFactoryBean` created by `@EnableKafkaStreams`.
See <<streams-config, Streams Configuration>> 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 <<streams-deser-recovery>> for more information.
The `HeaderEnricher` transformer has been provided, using SpEL to generate the header values.
See <<streams-header-enricher>> 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 <<streams-messaging>> and <<streams-integration>> 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 <<serdes>> 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 <<messaging-message-conversion>> for more information.
The `JsonSerializer`, `JsonDeserializer` and `JsonSerde` now have fluent APIs to make programmatic configuration simpler.
See the javadocs, <<serdes>>, and <<serde>> 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 <<aggregating-request-reply>> for more information.
==== Transaction Changes
You can now override the producer factory's `transactionIdPrefix` on the `KafkaTemplate` and `KafkaTransactionManager`.
See <<transaction-id-prefix>> 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 <<delegating-serialization>> 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 <<retrying-deserialization>> 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 <<kerberos>> 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

15
src/reference/asciidoc/kafka.adoc
View File

@@ -1496,6 +1496,8 @@ public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListene
void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);
void onPartitionsLost(Consumer<?, ?> consumer, Collection<TopicPartition> 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 <<seek-to-current, `SeekToCurrentErrorHandler`>> 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:

173
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<new-in-sik>>.
==== Tips, Tricks and Examples
A new chapter <<tips-n-tricks>> 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 <<rebalance-listeners>> 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<Serializer>` 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 <<producer-factory>> for more information.
The `isAckAfterHandle()` default implementation now returns true by default.
The same option is available with `Supplier<Deserializer>` instances in `DefaultKafkaConsumerFactory`.
See <<kafka-container>> 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 <<deps-for-24x>>.
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 <<seek-to-current>> for more information.
It is now possible to obtain the consumer's `group.id` property in the listener method.
See <<listener-group-id>> 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 <<message-listener-container>> 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 <<seek>> for more information.
A convenience class `AbstractConsumerSeekAware` is now provided to simplify seeking.
See <<seek>> 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 <<kafka-container>> for more information.
When using `AckMode.MANUAL` (or `MANUAL_IMMEDIATE`) you can now cause a redelivery by calling `nack` on the `Acknowledgment`.
See <<committing-offsets>> for more information.
Listener performance can now be monitored using Micrometer `Timer` s.
See <<micrometer>> for more information.
The containers now publish additional consumer lifecyle events relating to startup.
See <<events>> for more information.
Transactional batch listeners can now support zombie fencing.
See <<transactions>> 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 <<container-factory>> 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 <<seek-to-current>> 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 <<dead-letters>> for more information.
==== TopicBuilder
A new class `TopicBuilder` is provided for more convenient creation of `NewTopic` `@Bean` s for automatic topic provisioning.
See <<configuring-topics>> for more information.
==== Kafka Streams Changes
You can now perform additional configuration of the `StreamsBuilderFactoryBean` created by `@EnableKafkaStreams`.
See <<streams-config, Streams Configuration>> 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 <<streams-deser-recovery>> for more information.
The `HeaderEnricher` transformer has been provided, using SpEL to generate the header values.
See <<streams-header-enricher>> 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 <<streams-messaging>> and <<streams-integration>> 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 <<serdes>> 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 <<messaging-message-conversion>> for more information.
The `JsonSerializer`, `JsonDeserializer` and `JsonSerde` now have fluent APIs to make programmatic configuration simpler.
See the javadocs, <<serdes>>, and <<serde>> 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 <<aggregating-request-reply>> for more information.
==== Transaction Changes
You can now override the producer factory's `transactionIdPrefix` on the `KafkaTemplate` and `KafkaTransactionManager`.
See <<transaction-id-prefix>> 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 <<delegating-serialization>> 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 <<retrying-deserialization>> 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<byte[], Headers, T> failedDeserializationFunction`.
==== EmbeddedKafkaBroker Changes
You can now override the default broker list property name in the annotation.
See <<kafka-testing-embeddedkafka-annotation>> for more information.
==== ReplyingKafkaTemplate Changes
You can now customize the header names for correlation, reply topic and reply partition.
See <<replying-template>> for more information.
==== Header Mapper Changes
The `DefaultKafkaHeaderMapper` no longer encodes simple String-valued headers as JSON.
See <<header-mapping>> for more information.
* See <<x24-carl>> for important information about rebalance listeners.