diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java index 4ef50b20..eee2f6c1 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java @@ -54,6 +54,7 @@ import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.kafka.support.micrometer.MicrometerHolder; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; +import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; import org.springframework.util.concurrent.ListenableFuture; @@ -109,6 +110,8 @@ public class KafkaTemplate implements KafkaOperations, ApplicationCo private boolean allowNonTransactional; + private boolean converterSet; + private volatile boolean micrometerEnabled = true; private volatile MicrometerHolder micrometerHolder; @@ -244,8 +247,23 @@ public class KafkaTemplate implements KafkaOperations, ApplicationCo public void setMessageConverter(RecordMessageConverter messageConverter) { Assert.notNull(messageConverter, "'messageConverter' cannot be null"); this.messageConverter = messageConverter; + this.converterSet = true; } + /** + * Set the {@link SmartMessageConverter} to use with the default + * {@link MessagingMessageConverter}. Not allowed when a custom + * {@link #setMessageConverter(RecordMessageConverter) messageConverter} is provided. + * @param messageConverter the converter. + * @since 2.7.1 + */ + public void setMessagingConverter(SmartMessageConverter messageConverter) { + Assert.isTrue(!this.converterSet, "Cannot set the SmartMessageConverter when setting the messageConverter, " + + "add the SmartConverter to the message converter instead"); + ((MessagingMessageConverter) this.messageConverter).setMessagingConverter(messageConverter); + } + + @Override public boolean isTransactional() { return this.transactional; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java index 91103bff..05d54d09 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java @@ -223,6 +223,12 @@ public class MessagingMessageConverter implements RecordMessageConverter { return null; } else { + if (this.messagingConverter != null) { + Message message2 = this.messagingConverter.toMessage(payload, message.getHeaders()); + if (message2 != null) { + return message2.getPayload(); + } + } return payload; } } 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 5ec05d0e..0bca1220 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 @@ -237,6 +237,9 @@ public class EnableKafkaIntegrationTests { @Autowired private MeterRegistry meterRegistry; + @Autowired + private SmartMessageConverter fooContentConverter; + @Test public void testAnonymous() { MessageListenerContainer container = this.registry @@ -941,7 +944,9 @@ public class EnableKafkaIntegrationTests { @Test public void testContentConversion() throws InterruptedException { - template.send(MessageBuilder.withPayload("foo") + KafkaTemplate template = new KafkaTemplate<>(this.template.getProducerFactory()); + template.setMessagingConverter(this.fooContentConverter); + template.send(MessageBuilder.withPayload(new Foo("bar")) .setHeader(KafkaHeaders.TOPIC, "annotated41") .setHeader(MessageHeaders.CONTENT_TYPE, "application/foo") .build()); @@ -1626,6 +1631,14 @@ public class EnableKafkaIntegrationTests { return new Foo("bar"); } + @Override + @Nullable + protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, + @Nullable Object conversionHint) { + + return payload instanceof Foo ? ((Foo) payload).getBar() : null; + } + }; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/converter/MessagingMessageConverterTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/converter/MessagingMessageConverterTests.java index 11430866..e99a28c2 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/support/converter/MessagingMessageConverterTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/support/converter/MessagingMessageConverterTests.java @@ -22,6 +22,7 @@ import java.util.Arrays; import java.util.Collection; 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; @@ -110,17 +111,26 @@ public class MessagingMessageConverterTests { Headers headers = new RecordHeaders(); headers.add(new RecordHeader(MessageHeaders.CONTENT_TYPE, "application/foo".getBytes())); ConsumerRecord record = - new ConsumerRecord<>("foo", 1, 42, -1L, null, 0L, 0, 0, "bar", "foo", headers); + new ConsumerRecord<>("foo", 1, 42, -1L, null, 0L, 0, 0, "bar", "qux", headers); Message message = converter.toMessage(record, null, null, Foo.class); assertThat(message.getPayload()).isEqualTo(new Foo("bar")); + ProducerRecord pr = converter.fromMessage(message, "test"); + assertThat(pr.topic()).isEqualTo("test"); + assertThat(pr.value()).isEqualTo("foo".getBytes()); headers.remove(MessageHeaders.CONTENT_TYPE); headers.add(new RecordHeader(MessageHeaders.CONTENT_TYPE, "application/bar".getBytes())); message = converter.toMessage(record, null, null, Bar.class); assertThat(message.getPayload()).isEqualTo(new Bar("bar")); + pr = converter.fromMessage(message, "test"); + assertThat(pr.topic()).isEqualTo("test"); + assertThat(pr.value()).isEqualTo("bar".getBytes()); headers.remove(MessageHeaders.CONTENT_TYPE); headers.add(new RecordHeader(MessageHeaders.CONTENT_TYPE, "application/baz".getBytes())); message = converter.toMessage(record, null, null, Bar.class); - assertThat(message.getPayload()).isEqualTo("foo"); // no contentType match + assertThat(message.getPayload()).isEqualTo("qux"); // no contentType match + pr = converter.fromMessage(message, "test"); + assertThat(pr.topic()).isEqualTo("test"); + assertThat(pr.value()).isEqualTo("qux"); } static class FooConverter extends AbstractMessageConverter { @@ -142,6 +152,14 @@ public class MessagingMessageConverterTests { return new Foo("bar"); } + @Override + @Nullable + protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, + @Nullable Object conversionHint) { + + return "foo".getBytes(); + } + } static class BarConverter extends FooConverter { @@ -158,6 +176,14 @@ public class MessagingMessageConverterTests { return new Bar("bar"); } + @Override + @Nullable + protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, + @Nullable Object conversionHint) { + + return "bar".getBytes(); + } + } public static class Foo {