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

GH-601: @KafkaListener: Support List<Message<Foo>>

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/601
This commit is contained in:
Gary Russell
2018-03-16 14:15:11 -04:00
committed by Artem Bilan
parent 6bd58b2b66
commit 522f6b482d
6 changed files with 95 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -61,7 +61,7 @@ public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessage
private static final Message<KafkaNull> NULL_MESSAGE = new GenericMessage<>(KafkaNull.INSTANCE);
private BatchMessageConverter messageConverter = new BatchMessagingMessageConverter();
private BatchMessageConverter batchMessageConverter = new BatchMessagingMessageConverter();
private KafkaListenerErrorHandler errorHandler;
@@ -79,7 +79,10 @@ public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessage
* @param messageConverter the converter.
*/
public void setBatchMessageConverter(BatchMessageConverter messageConverter) {
this.messageConverter = messageConverter;
this.batchMessageConverter = messageConverter;
if (messageConverter.getRecordMessageConverter() != null) {
setMessageConverter(messageConverter.getRecordMessageConverter());
}
}
/**
@@ -89,7 +92,7 @@ public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessage
* being able to convert {@link org.springframework.messaging.Message}.
*/
protected final BatchMessageConverter getBatchMessageConverter() {
return this.messageConverter;
return this.batchMessageConverter;
}
/**

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

@@ -428,9 +428,12 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
this.isConsumerRecordList = paramType.equals(ConsumerRecord.class)
|| (paramType instanceof ParameterizedType
&& ((ParameterizedType) paramType).getRawType().equals(ConsumerRecord.class));
this.isMessageList = paramType.equals(Message.class)
|| (paramType instanceof ParameterizedType
&& ((ParameterizedType) paramType).getRawType().equals(Message.class));
boolean messageHasGeneric = paramType instanceof ParameterizedType
&& ((ParameterizedType) paramType).getRawType().equals(Message.class);
this.isMessageList = paramType.equals(Message.class) || messageHasGeneric;
if (messageHasGeneric) {
genericParameterType = ((ParameterizedType) paramType).getActualTypeArguments()[0];
}
}
}
}

14
spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -24,6 +24,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
/**
@@ -53,4 +54,15 @@ public interface BatchMessageConverter extends MessageConverter {
*/
List<ProducerRecord<?, ?>> fromMessage(Message<?> message, String defaultTopic);
/**
* Return the record converter used by this batch converter, if configured,
* or null.
* @return the converter or null.
* @since 2.1.5
*/
@Nullable
default RecordMessageConverter getRecordMessageConverter() {
return null;
}
}

7
spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -117,6 +117,11 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
this.headerMapper = headerMapper;
}
@Override
public RecordMessageConverter getRecordMessageConverter() {
return this.recordConverter;
}
@Override
public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {

40
spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -43,6 +43,7 @@ import org.springframework.kafka.support.converter.BatchMessagingMessageConverte
import org.springframework.kafka.support.converter.StringJsonMessageConverter;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
@@ -62,7 +63,7 @@ public class BatchListenerConversionTests {
private static final String DEFAULT_TEST_GROUP_ID = "blc";
@ClassRule // one topic to preserve order
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 1, "blc1", "blc2");
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 1, "blc1", "blc2", "blc3");
@Autowired
private Config config;
@@ -90,6 +91,20 @@ public class BatchListenerConversionTests {
assertThat((listener.receivedPartitions).get(0)).isEqualTo(0);
}
@Test
public void testBatchOfPojoMessages() throws Exception {
String topic = "blc3";
this.template.send(new GenericMessage<>(
new Foo("bar"), Collections.singletonMap(KafkaHeaders.TOPIC, topic)));
this.template.send(new GenericMessage<>(
new Foo("baz"), Collections.singletonMap(KafkaHeaders.TOPIC, topic)));
Listener3 listener = this.config.listener3();
assertThat(listener.latch1.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(listener.received.size()).isGreaterThan(0);
assertThat(listener.received.get(0).getPayload()).isInstanceOf(Foo.class);
assertThat(listener.received.get(0).getPayload().getBar()).isEqualTo("bar");
}
@Configuration
@EnableKafka
public static class Config {
@@ -149,6 +164,11 @@ public class BatchListenerConversionTests {
return new Listener("blc2");
}
@Bean
public Listener3 listener3() {
return new Listener3();
}
}
public static class Listener {
@@ -191,6 +211,22 @@ public class BatchListenerConversionTests {
}
public static class Listener3 {
private final CountDownLatch latch1 = new CountDownLatch(1);
private List<Message<Foo>> received;
@KafkaListener(topics = "blc3", groupId = "blc3")
public void listen1(List<Message<Foo>> foos) {
if (this.received == null) {
this.received = foos;
}
this.latch1.countDown();
}
}
public static class Foo {
public String bar;

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

@@ -1,4 +1,3 @@
[[kafka]]
=== Using Spring for Apache Kafka
@@ -863,6 +862,11 @@ public void listen15(List<Message<?>> list, Acknowledgment ack) {
}
----
No conversion is performed on the payloads in this case.
If the `BatchMessagingMessageConverter` is configured with a `RecordMessageConverter`, you can also add a generic type to the `Message` parameter and the payloads will be converted.
See <<payload-conversion-with-batch>> for more information.
You can also receive a list of `ConsumerRecord<?, ?>` objects but it must be the only parameter (aside from an optional `Acknowledgment` when using manual commits) defined on the method:
[source, java]
@@ -1382,6 +1386,8 @@ When a container is paused, it continues to `poll()` the consumer, avoiding a re
[[serdes]]
==== Serialization/Deserialization and Message Conversion
===== Overview
Apache Kafka provides a high-level API for serializing/deserializing record values as well as their keys.
It is present with the `org.apache.kafka.common.serialization.Serializer<T>` and
`org.apache.kafka.common.serialization.Deserializer<T>` abstractions with some built-in implementations.
@@ -1460,7 +1466,10 @@ With a class-level `@KafkaListener`, the payload type is used to select which `@
NOTE: When using the `StringJsonMessageConverter`, you should use a `StringDeserializer` in the kafka consumer configuration and `StringSerializer` in the kafka producer configuration, when using Spring Integration or the `KafkaTemplate.send(Message<?> message)` method.
Starting with _version 1.3.2_ you can also use a `StringJsonMessageConverter` within a `BatchMessagingMessageConverter` for converting batch messages, when using a batch listener container factory.
[[payload-conversion-with-batch]]
===== Payload Conversion with Batch Listeners
Starting with _version 1.3.2_, you can also use a `StringJsonMessageConverter` within a `BatchMessagingMessageConverter` for converting batch messages, when using a batch listener container factory.
By default, the type for the conversion is inferred from the listener argument.
If you configure the `StringJsonMessageConverter` with a `DefaultJackson2TypeMapper` that has its `TypePrecedence` set to `TYPE_ID` (instead of the default `INFERRED`), then the converter will use type information in headers (if present) instead.
@@ -1498,8 +1507,20 @@ public void listen(List<Foo> foos, @Header(KafkaHeaders.OFFSET) List<Long> offse
Notice that you can still access the batch headers too.
Starting with _versions 2.1.1_, the `org.springframework.core.convert.ConversionService` used by the default
`o.s.messaging.handler.annotation.support.MessageHandlerMethodFactory` to reslove parameters for the invocation
If the batch converter has a record converter that supports it, you can also receive a list of messages where the payloads are converted according to the generic type:
[source, java]
----
@KafkaListener(topics = "blc3", groupId = "blc3")
public void listen1(List<Message<Foo>> fooMessages) {
...
}
----
===== ConversionService Customization
Starting with _version 2.1.1_, the `org.springframework.core.convert.ConversionService` used by the default
`o.s.messaging.handler.annotation.support.MessageHandlerMethodFactory` to resolve parameters for the invocation
of a listener method is supplied with all beans implementing any of the following interfaces:
- `org.springframework.core.convert.converter.Converter`