diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java index 89fc0601..781c4f95 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java @@ -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 extends MessagingMessage private static final Message 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 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 extends MessagingMessage * being able to convert {@link org.springframework.messaging.Message}. */ protected final BatchMessageConverter getBatchMessageConverter() { - return this.messageConverter; + return this.batchMessageConverter; } /** diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java index bea9516b..c84a1c63 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java @@ -428,9 +428,12 @@ public abstract class MessagingMessageListenerAdapter 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]; + } } } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessageConverter.java index a0e1699d..1d4b6cdb 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessageConverter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessageConverter.java @@ -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> 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; + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java index 7edbb209..b2b1bc3c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java @@ -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> records, Acknowledgment acknowledgment, Consumer consumer, Type type) { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversionTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversionTests.java index a064636b..1dabe75d 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversionTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversionTests.java @@ -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> received; + + @KafkaListener(topics = "blc3", groupId = "blc3") + public void listen1(List> foos) { + if (this.received == null) { + this.received = foos; + } + this.latch1.countDown(); + } + + } + public static class Foo { public String bar; diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index bcb3eff9..84dc73bc 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -1,4 +1,3 @@ - [[kafka]] === Using Spring for Apache Kafka @@ -863,6 +862,11 @@ public void listen15(List> 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 <> 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` and `org.apache.kafka.common.serialization.Deserializer` 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 foos, @Header(KafkaHeaders.OFFSET) List 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> 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`