GH-1789: Refactoring

Resolves #1789

Move the `SmartMessageConverter` invocations to a more logical place,
based on their expected inputs for comversion.

* Apply suggestions from code review

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -4205,6 +4205,36 @@ Again, using `byte[]` or `Bytes` is more efficient because they avoid a `String`
 For convenience, starting with version 2.3, the framework also provides a `StringOrBytesSerializer` which can serialize all three value types so it can be used with any of the message converters.
 ====
 
+Starting with version 2.7.1, message payload conversion can be delegated to a `spring-messaging` `SmartMessageConverter`; this enables conversion, for example, to be based on the `MessageHeaders.CONTENT_TYPE` header.
+
+IMPORTANT:	The `KafkaMessageConverter.fromMessage()` method is called for outbound conversion to a `ProducerRecord` with the message payload in the `ProducerRecord.value()` property.
+The `KafkaMessageConverter.toMessage()` method is called for inbound conversion from `ConsumerRecord` with the payload being the `ConsumerRecord.value()` property.
+The `SmartMessageConverter.toMessage()` method is called to create a new outbound `Message<?>` from the `Message` passed to`fromMessage()` (usually by `KafkaTemplate.send(Message<?> msg)`).
+Similarly, in the `KafkaMessageConverter.toMessage()` method, after the converter has created a new `Message<?>` from the `ConsumerRecord`, the `SmartMessageConverter.fromMessage()` method is called and then the final inbound message is created with the newly converted payload.
+In either case, if the `SmartMessageConverter` returns `null`, the original message is used.
+
+When the default converter is used in the `KafkaTemplate` and listener container factory, you configure the `SmartMessageConverter` by calling `setMessagingConverter()` on the template and via the `contentMessageConverter` property on `@KafkaListener` methods.
+
+Examples:
+
+====
+[source, java]
+----
+template.setMessagingConverter(mySmartConverter);
+----
+====
+
+====
+[source, java]
+----
+@KafkaListener(id = "withSmartConverter", topics = "someTopic",
+    contentTypeConverter = "mySmartConverter")
+public void smart(Thing thing) {
+    ...
+}
+----
+====
+
 [[data-projection]]
 ====== Using Spring Data Projection Interfaces
 

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -85,3 +85,9 @@ See <<streams-config>> for more information.
 New methods `createOrModifyTopics` and `describeTopics` have been added.
 `KafkaAdmin.NewTopics` has been added to facilitate configuring multiple topics in a single bean.
 See <<configuring-topics>> for more information.
+
+[[x27-conv]]
+==== `MessageConverter` Changes
+
+It is now possible to add a `spring-messaging` `SmartMessageConverter` to the `MessagingMessageConverter`, allowing content negotiation based on the `contentType` header.
+See <<messaging-message-conversion>> for more information.

spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java

@@ -25,6 +25,7 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 
@@ -37,7 +38,6 @@
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.kafka.support.SimpleKafkaHeaderMapper;
-import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.SmartMessageConverter;
@@ -124,8 +124,26 @@ protected org.springframework.messaging.converter.MessageConverter getMessagingC
 
 	/**
 	 * Set a spring-messaging {@link SmartMessageConverter} to convert the record value to
-	 * the desired type. This will also cause the {@link MessageHeaders#CONTENT_TYPE} to be
-	 * converted to String when mapped inbound.
+	 * the desired type. This will also cause the {@link MessageHeaders#CONTENT_TYPE} to
+	 * be converted to String when mapped inbound.
+	 * <p>
+	 * IMPORTANT: This converter's {@link #fromMessage(Message, String)} method is called
+	 * for outbound conversion to a {@link ProducerRecord} with the message payload in the
+	 * {@link ProducerRecord#value()} property.
+	 * {@link #toMessage(ConsumerRecord, Acknowledgment, Consumer, Type)} is called for
+	 * inbound conversion from {@link ConsumerRecord} with the payload being the
+	 * {@link ConsumerRecord#value()} property.
+	 * <p>
+	 * The {@link SmartMessageConverter#toMessage(Object, MessageHeaders)} method is
+	 * called to create a new outbound {@link Message} from the {@link Message} passed to
+	 * {@link #fromMessage(Message, String)}. Similarly, in
+	 * {@link #toMessage(ConsumerRecord, Acknowledgment, Consumer, Type)}, after this
+	 * converter has created a new {@link Message} from the {@link ConsumerRecord} the
+	 * {@link SmartMessageConverter#fromMessage(Message, Class)} method is called and then
+	 * the final inbound message is created with the newly converted payload.
+	 * <p>
+	 * In either case, if the {@link SmartMessageConverter} returns {@code null}, the
+	 * original message is used.
 	 * @param messagingConverter the converter.
 	 * @since 2.7.1
 	 */
@@ -144,36 +162,51 @@ public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowle
 				this.generateTimestamp);
 
 		Map<String, Object> rawHeaders = kafkaMessageHeaders.getRawHeaders();
-		boolean removeNative = true;
-		if (this.headerMapper != null && record.headers() != null) {
-			this.headerMapper.toHeaders(record.headers(), rawHeaders);
-		}
-		else {
-			this.logger.debug(() ->
-					"No header mapper is available; Jackson is required for the default mapper; "
-					+ "headers (if present) are not mapped but provided raw in "
-					+ KafkaHeaders.NATIVE_HEADERS);
-			removeNative = false;
+		if (record.headers() != null) {
+			if (this.headerMapper != null) {
+				this.headerMapper.toHeaders(record.headers(), rawHeaders);
+			}
+			else {
+				this.logger.debug(() ->
+						"No header mapper is available; Jackson is required for the default mapper; "
+						+ "headers (if present) are not mapped but provided raw in "
+						+ KafkaHeaders.NATIVE_HEADERS);
+				rawHeaders.put(KafkaHeaders.NATIVE_HEADERS, record.headers());
+				Header contentType = record.headers().lastHeader(MessageHeaders.CONTENT_TYPE);
+				if (contentType != null) {
+					rawHeaders.put(MessageHeaders.CONTENT_TYPE,
+							new String(contentType.value(), StandardCharsets.UTF_8));
+				}
+			}
 		}
-		rawHeaders.put(KafkaHeaders.NATIVE_HEADERS, record.headers());
 		String ttName = record.timestampType() != null ? record.timestampType().name() : null;
 		commonHeaders(acknowledgment, consumer, rawHeaders, record.key(), record.topic(), record.partition(),
 				record.offset(), ttName, record.timestamp());
 		if (this.rawRecordHeader) {
 			rawHeaders.put(KafkaHeaders.RAW_DATA, record);
 		}
-		Object value = this.messagingConverter == null
-				? extractAndConvertValue(record, type)
-				: extractAndConvertValue(record, type, kafkaMessageHeaders);
-		if (removeNative) {
-			rawHeaders.remove(KafkaHeaders.NATIVE_HEADERS);
+		Message<?> message = MessageBuilder.createMessage(extractAndConvertValue(record, type), kafkaMessageHeaders);
+		if (this.messagingConverter != null && !message.getPayload().equals(KafkaNull.INSTANCE)) {
+			Class<?> clazz = type instanceof Class ? (Class<?>) type : type instanceof ParameterizedType
+					? (Class<?>) ((ParameterizedType) type).getRawType() : Object.class;
+			Object payload = this.messagingConverter.fromMessage(message, clazz, type);
+			if (payload != null) {
+				message = new GenericMessage<>(payload, message.getHeaders());
+			}
 		}
-		return MessageBuilder.createMessage(value, kafkaMessageHeaders);
+		return message;
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Override
-	public ProducerRecord<?, ?> fromMessage(Message<?> message, String defaultTopic) {
+	public ProducerRecord<?, ?> fromMessage(Message<?> messageArg, String defaultTopic) {
+		Message<?> message = messageArg;
+		if (this.messagingConverter != null) {
+			Message<?> converted = this.messagingConverter.toMessage(message.getPayload(), message.getHeaders());
+			if (converted != null) {
+				message = converted;
+			}
+		}
 		MessageHeaders headers = message.getHeaders();
 		Object topicHeader = headers.get(KafkaHeaders.TOPIC);
 		String topic = null;
@@ -223,12 +256,6 @@ protected Object convertPayload(Message<?> message) {
 			return null;
 		}
 		else {
-			if (this.messagingConverter != null) {
-				Message<?> message2 = this.messagingConverter.toMessage(payload, message.getHeaders());
-				if (message2 != null) {
-					return message2.getPayload();
-				}
-			}
 			return payload;
 		}
 	}
@@ -241,31 +268,7 @@ protected Object convertPayload(Message<?> message) {
 	 * @return the value.
 	 */
 	protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type) {
-		return extractAndConvertValue(record, type, null);
-	}
-
-	/**
-	 * Subclasses can convert the value; by default, it's returned as provided by Kafka
-	 * unless there is a {@link SmartMessageConverter} that can convert it.
-	 * @param record the record.
-	 * @param type the required type.
-	 * @param headers the mapped headers.
-	 * @return the value.
-	 */
-	protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type, @Nullable MessageHeaders headers) {
-		if (record.value() == null) {
-			return KafkaNull.INSTANCE;
-		}
-		if (this.messagingConverter != null) {
-			Class<?> clazz = type instanceof Class ? (Class<?>) type : type instanceof ParameterizedType
-					? (Class<?>) ((ParameterizedType) type).getRawType() : Object.class;
-			Object payload = this.messagingConverter
-					.fromMessage(new GenericMessage<>(record.value(), headers), clazz, type);
-			if (payload != null) {
-				return payload;
-			}
-		}
-		return record.value();
+		return record.value() == null ? KafkaNull.INSTANCE : record.value();
 	}
 
 }

spring-kafka/src/test/java/org/springframework/kafka/support/converter/MessagingMessageConverterTests.java

@@ -133,6 +133,24 @@ void contentNegotiation() {
 		assertThat(pr.value()).isEqualTo("qux");
 	}
 
+	@Test
+	void contentNegotiationNoHeaderMapper() {
+		MessagingMessageConverter converter = new MessagingMessageConverter();
+		converter.setHeaderMapper(null);
+		Collection<MessageConverter> converters = Arrays.asList(new FooConverter(MimeType.valueOf("application/foo")),
+				new BarConverter(MimeType.valueOf("application/bar")));
+		converter.setMessagingConverter(new CompositeMessageConverter(converters));
+		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", "qux", headers);
+		Message<?> message = converter.toMessage(record, null, null, Foo.class);
+		assertThat(message.getPayload()).isEqualTo(new Foo("bar"));
+		headers.remove(MessageHeaders.CONTENT_TYPE);
+		message = converter.toMessage(record, null, null, Foo.class);
+		assertThat(message.getPayload()).isEqualTo(new Foo("bar")); // no contentType header
+	}
+
 	static class FooConverter extends AbstractMessageConverter {
 
 		FooConverter(MimeType supportedMimeType) {