GH-1789: Content Type Conversion (Outbound)

See https://github.com/spring-projects/spring-kafka/issues/1789

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<K, V> implements KafkaOperations<K, V>, ApplicationCo
 
 	private boolean allowNonTransactional;
 
+	private boolean converterSet;
+
 	private volatile boolean micrometerEnabled = true;
 
 	private volatile MicrometerHolder micrometerHolder;
@@ -244,8 +247,23 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, 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;

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;
 		}
 	}

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<Integer, String> 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;
+				}
+
 			};
 		}
 

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<String, String> 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 {