GH-1811: MessageConverter Nullable - Kotlin Compat.

Resolves #1811

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

@@ -482,7 +482,9 @@ public void afterPropertiesSet() {
 	}
 
 	@Override
-	public void setupListenerContainer(MessageListenerContainer listenerContainer, MessageConverter messageConverter) {
+	public void setupListenerContainer(MessageListenerContainer listenerContainer,
+			@Nullable MessageConverter messageConverter) {
+
 		setupMessageListener(listenerContainer, messageConverter);
 	}
 
@@ -494,10 +496,12 @@ public void setupListenerContainer(MessageListenerContainer listenerContainer, M
 	 * @return a {@link MessageListener} instance.
 	 */
 	protected abstract MessagingMessageListenerAdapter<K, V> createMessageListener(MessageListenerContainer container,
-			MessageConverter messageConverter);
+			@Nullable MessageConverter messageConverter);
 
 	@SuppressWarnings("unchecked")
-	private void setupMessageListener(MessageListenerContainer container, MessageConverter messageConverter) {
+	private void setupMessageListener(MessageListenerContainer container,
+			@Nullable MessageConverter messageConverter) {
+
 		MessagingMessageListenerAdapter<K, V> adapter = createMessageListener(container, messageConverter);
 		if (this.replyHeadersConfigurer != null) {
 			adapter.setReplyHeadersConfigurer(this.replyHeadersConfigurer);

spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java

@@ -135,7 +135,8 @@ default Properties getConsumerProperties() {
 	 * @param listenerContainer the listener container to configure
 	 * @param messageConverter the message converter - can be null
 	 */
-	void setupListenerContainer(MessageListenerContainer listenerContainer, MessageConverter messageConverter);
+	void setupListenerContainer(MessageListenerContainer listenerContainer,
+			@Nullable MessageConverter messageConverter);
 
 	/**
 	 * When true, {@link Iterable} return results will be split into discrete records.

spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointAdapter.java

@@ -93,7 +93,7 @@ public Boolean getAutoStartup() { // NOSONAR
 
 	@Override
 	public void setupListenerContainer(MessageListenerContainer listenerContainer,
-			MessageConverter messageConverter) {
+			@Nullable MessageConverter messageConverter) {
 	}
 
 	@Override

spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java

@@ -169,7 +169,7 @@ protected MessageHandlerMethodFactory getMessageHandlerMethodFactory() {
 
 	@Override
 	protected MessagingMessageListenerAdapter<K, V> createMessageListener(MessageListenerContainer container,
-			MessageConverter messageConverter) {
+			@Nullable MessageConverter messageConverter) {
 
 		Assert.state(this.messageHandlerMethodFactory != null,
 				"Could not create message listener - MessageHandlerMethodFactory not set");
@@ -202,7 +202,9 @@ protected HandlerAdapter configureListenerAdapter(MessagingMessageListenerAdapte
 	 * @param messageConverter the converter (may be null).
 	 * @return the {@link MessagingMessageListenerAdapter} instance.
 	 */
-	protected MessagingMessageListenerAdapter<K, V> createMessageListenerInstance(MessageConverter messageConverter) {
+	protected MessagingMessageListenerAdapter<K, V> createMessageListenerInstance(
+			@Nullable MessageConverter messageConverter) {
+
 		MessagingMessageListenerAdapter<K, V> listener;
 		if (isBatchListener()) {
 			BatchMessagingMessageListenerAdapter<K, V> messageListener = new BatchMessagingMessageListenerAdapter<K, V>(