Add nullability changes in listener/adapter package

#3762

Signed-off-by: Soby Chacko <[email protected]>

spring-kafka/src/main/java/org/springframework/kafka/listener/GenericMessageListener.java

@@ -56,7 +56,7 @@ default void onMessage(T data, @Nullable Acknowledgment acknowledgment) {
 	 * @param consumer the consumer.
 	 * @since 2.0
 	 */
-	default void onMessage(T data, Consumer<?, ?> consumer) {
+	default void onMessage(T data, @Nullable Consumer<?, ?> consumer) {
 		throw new UnsupportedOperationException("Container should never call this");
 	}
 
@@ -68,7 +68,7 @@ default void onMessage(T data, Consumer<?, ?> consumer) {
 	 * @param consumer the consumer.
 	 * @since 2.0
 	 */
-	default void onMessage(T data, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
+	default void onMessage(T data, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer) {
 		throw new UnsupportedOperationException("Container should never call this");
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerBackoffManager.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2025 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.
@@ -18,6 +18,7 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
+import org.jspecify.annotations.Nullable;
 
 /**
  * Interface for backing off a {@link MessageListenerContainer}
@@ -32,7 +33,7 @@ public interface KafkaConsumerBackoffManager {
 	void backOffIfNecessary(Context context);
 
 	default Context createContext(long dueTimestamp, String listenerId, TopicPartition topicPartition,
-								Consumer<?, ?> messageConsumer) {
+								@Nullable Consumer<?, ?> messageConsumer) {
 		return new Context(dueTimestamp, topicPartition, listenerId, messageConsumer);
 	}
 
@@ -64,7 +65,7 @@ class Context {
 		private final Consumer<?, ?> consumerForTimingAdjustment;
 
 		Context(long dueTimestamp, TopicPartition topicPartition, String listenerId,
-				Consumer<?, ?> consumerForTimingAdjustment) {
+				@Nullable Consumer<?, ?> consumerForTimingAdjustment) {
 
 			this.dueTimestamp = dueTimestamp;
 			this.listenerId = listenerId;
@@ -84,7 +85,7 @@ public TopicPartition getTopicPartition() {
 			return this.topicPartition;
 		}
 
-		public Consumer<?, ?> getConsumerForTimingAdjustment() {
+		public @Nullable Consumer<?, ?> getConsumerForTimingAdjustment() {
 			return this.consumerForTimingAdjustment;
 		}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractDelegatingMessageListenerAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2025 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.
@@ -21,6 +21,7 @@
 
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.common.TopicPartition;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.listener.ConsumerSeekAware;
@@ -46,7 +47,7 @@ public abstract class AbstractDelegatingMessageListenerAdapter<T>
 
 	protected final ListenerType delegateType; // NOSONAR
 
-	private final ConsumerSeekAware seekAware;
+	private final @Nullable ConsumerSeekAware seekAware;
 
 	public AbstractDelegatingMessageListenerAdapter(T delegate) {
 		this.delegate = delegate;

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.java

@@ -37,7 +37,7 @@ public abstract class AbstractRetryingMessageListenerAdapter<K, V, T>
 
 	private final RetryTemplate retryTemplate;
 
-	private final RecoveryCallback<? extends Object> recoveryCallback;
+	private final @Nullable RecoveryCallback<? extends Object> recoveryCallback;
 
 	/**
 	 * Construct an instance with the supplied retry template. The exception will be
@@ -69,7 +69,7 @@ public RetryTemplate getRetryTemplate() {
 		return this.retryTemplate;
 	}
 
-	public RecoveryCallback<? extends Object> getRecoveryCallback() {
+	public @Nullable RecoveryCallback<? extends Object> getRecoveryCallback() {
 		return this.recoveryCallback;
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java

@@ -62,7 +62,7 @@ public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessage
 
 	private BatchMessageConverter batchMessageConverter = new BatchMessagingMessageConverter();
 
-	private BatchToRecordAdapter<K, V> batchToRecordAdapter;
+	private @Nullable BatchToRecordAdapter<K, V> batchToRecordAdapter;
 
 	/**
 	 * Create an instance with the provided parameters.

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchToRecordAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2025 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.
@@ -20,6 +20,7 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.messaging.Message;
@@ -47,7 +48,7 @@ public interface BatchToRecordAdapter<K, V> {
 	 * @param consumer the consumer.
 	 * @param callback the callback.
 	 */
-	void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records, Acknowledgment ack,
+	void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment ack,
 			Consumer<?, ?> consumer, Callback<K, V> callback);
 
 	/**
@@ -66,7 +67,7 @@ interface Callback<K, V> {
 		 * @param consumer the consumer.
 		 * @param message the message.
 		 */
-		void invoke(ConsumerRecord<K, V> record, Acknowledgment ack, Consumer<?, ?> consumer,
+		void invoke(ConsumerRecord<K, V> record, @Nullable Acknowledgment ack, Consumer<?, ?> consumer,
 				Message<?> message);
 
 	}

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2025 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.
@@ -21,6 +21,7 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.kafka.listener.AcknowledgingConsumerAwareMessageListener;
 import org.springframework.kafka.listener.AcknowledgingMessageListener;
@@ -60,7 +61,7 @@ public class ConvertingMessageListener<V> implements DelegatingMessageListener<M
 
 	private MessageConverter messageConverter;
 
-	private KafkaHeaderMapper headerMapper;
+	private @Nullable KafkaHeaderMapper headerMapper;
 
 	/**
 	 * Construct an instance with the provided {@link MessageListener} and {@link Class}
@@ -106,7 +107,7 @@ public MessageListener getDelegate() {
 
 	@Override
 	@SuppressWarnings("unchecked")
-	public void onMessage(ConsumerRecord receivedRecord, Acknowledgment acknowledgment, Consumer consumer) {
+	public void onMessage(ConsumerRecord receivedRecord, @Nullable Acknowledgment acknowledgment, Consumer consumer) {
 		ConsumerRecord convertedConsumerRecord = convertConsumerRecord(receivedRecord);
 		if (this.delegate instanceof AcknowledgingConsumerAwareMessageListener) {
 			this.delegate.onMessage(convertedConsumerRecord, acknowledgment, consumer);

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DefaultBatchToRecordAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2025 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.
@@ -20,6 +20,7 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.listener.ConsumerRecordRecoverer;
@@ -62,7 +63,7 @@ public DefaultBatchToRecordAdapter(ConsumerRecordRecoverer recoverer) {
 	}
 
 	@Override
-	public void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records, Acknowledgment ack,
+	public void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment ack,
 			Consumer<?, ?> consumer, Callback<K, V> callback) {
 
 		for (int i = 0; i < messages.size(); i++) {

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java

@@ -69,7 +69,7 @@ public class DelegatingInvocableHandler {
 	private final ConcurrentMap<InvocableHandlerMethod, MethodParameter> payloadMethodParameters =
 			new ConcurrentHashMap<>();
 
-	private final InvocableHandlerMethod defaultHandler;
+	private final @Nullable InvocableHandlerMethod defaultHandler;
 
 	private final Map<InvocableHandlerMethod, Expression> handlerSendTo = new ConcurrentHashMap<>();
 
@@ -79,13 +79,13 @@ public class DelegatingInvocableHandler {
 
 	private final Object bean;
 
-	private final BeanExpressionResolver resolver;
+	private final @Nullable BeanExpressionResolver resolver;
 
-	private final BeanExpressionContext beanExpressionContext;
+	private final @Nullable BeanExpressionContext beanExpressionContext;
 
-	private final ConfigurableListableBeanFactory beanFactory;
+	private final @Nullable ConfigurableListableBeanFactory beanFactory;
 
-	private final PayloadValidator validator;
+	private final @Nullable PayloadValidator validator;
 
 	private final boolean asyncReplies;
 
@@ -168,7 +168,8 @@ public boolean isAsyncReplies() {
 	 * @throws Exception raised if no suitable argument resolver can be found,
 	 * or the method raised an exception.
 	 */
-	public Object invoke(Message<?> message, Object... providedArgs) throws Exception { //NOSONAR
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
+	public Object invoke(Message<?> message, @Nullable Object... providedArgs) throws Exception { //NOSONAR
 		Class<?> payloadClass = message.getPayload().getClass();
 		InvocableHandlerMethod handler = getHandlerForPayload(payloadClass);
 		if (this.validator != null && this.defaultHandler != null) {
@@ -345,6 +346,7 @@ private boolean assignPayload(MethodParameter methodParameter, Class<?> payloadC
 	 * @since 3.2
 	 */
 	@Nullable
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	public InvocationResult getInvocationResultFor(Object result, Object inboundPayload) {
 		InvocableHandlerMethod handler = findHandlerForPayload(inboundPayload.getClass());
 		if (handler != null) {

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java

@@ -78,7 +78,7 @@ public FilteringBatchMessageListenerAdapter(BatchMessageListener<K, V> delegate,
 
 	@Override
 	public void onMessage(List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment acknowledgment,
-			Consumer<?, ?> consumer) {
+			@Nullable Consumer<?, ?> consumer) {
 
 		final RecordFilterStrategy<K, V> recordFilterStrategy = getRecordFilterStrategy();
 		final List<ConsumerRecord<K, V>> consumerRecords = recordFilterStrategy.filterBatch(records);
@@ -100,8 +100,8 @@ else if (!consumerRecords.isEmpty() || this.consumerAware
 		}
 	}
 
-	private void invokeDelegate(List<ConsumerRecord<K, V>> consumerRecords, Acknowledgment acknowledgment,
-			Consumer<?, ?> consumer) {
+	private void invokeDelegate(List<ConsumerRecord<K, V>> consumerRecords, @Nullable Acknowledgment acknowledgment,
+			@Nullable Consumer<?, ?> consumer) {
 
 		switch (this.delegateType) {
 			case ACKNOWLEDGING_CONSUMER_AWARE:
@@ -129,12 +129,12 @@ public void onMessage(List<ConsumerRecord<K, V>> data) {
 	}
 
 	@Override
-	public void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment) {
+	public void onMessage(List<ConsumerRecord<K, V>> data, @Nullable Acknowledgment acknowledgment) {
 		onMessage(data, acknowledgment, null); // NOSONAR
 	}
 
 	@Override
-	public void onMessage(List<ConsumerRecord<K, V>> data, Consumer<?, ?> consumer) {
+	public void onMessage(List<ConsumerRecord<K, V>> data, @Nullable Consumer<?, ?> consumer) {
 		onMessage(data, null, consumer);
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringMessageListenerAdapter.java

@@ -66,7 +66,7 @@ public FilteringMessageListenerAdapter(MessageListener<K, V> delegate,
 
 	@Override
 	public void onMessage(ConsumerRecord<K, V> consumerRecord, @Nullable Acknowledgment acknowledgment,
-			Consumer<?, ?> consumer) {
+			@Nullable Consumer<?, ?> consumer) {
 
 		if (!filter(consumerRecord)) {
 			switch (this.delegateType) {
@@ -104,12 +104,12 @@ public void onMessage(ConsumerRecord<K, V> data) {
 	}
 
 	@Override
-	public void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment) {
+	public void onMessage(ConsumerRecord<K, V> data, @Nullable Acknowledgment acknowledgment) {
 		onMessage(data, acknowledgment, null); // NOSONAR
 	}
 
 	@Override
-	public void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer) {
+	public void onMessage(ConsumerRecord<K, V> data, @Nullable Consumer<?, ?> consumer) {
 		onMessage(data, null, consumer);
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/HandlerAdapter.java

@@ -17,6 +17,7 @@
 package org.springframework.kafka.listener.adapter;
 
 import java.lang.reflect.Method;
+import java.util.Objects;
 
 import org.jspecify.annotations.Nullable;
 
@@ -35,9 +36,9 @@
  */
 public class HandlerAdapter {
 
-	private final InvocableHandlerMethod invokerHandlerMethod;
+	private final @Nullable InvocableHandlerMethod invokerHandlerMethod;
 
-	private final DelegatingInvocableHandler delegatingHandler;
+	private final @Nullable DelegatingInvocableHandler delegatingHandler;
 
 	private final boolean asyncReplies;
 
@@ -74,11 +75,11 @@ public boolean isAsyncReplies() {
 	}
 
 	@Nullable
-	public Object invoke(Message<?> message, Object... providedArgs) throws Exception { //NOSONAR
+	public Object invoke(Message<?> message, @Nullable Object... providedArgs) throws Exception { //NOSONAR
 		if (this.invokerHandlerMethod != null) {
 			return this.invokerHandlerMethod.invoke(message, providedArgs); // NOSONAR
 		}
-		else if (this.delegatingHandler.hasDefaultHandler()) {
+		else if (Objects.requireNonNull(this.delegatingHandler).hasDefaultHandler()) {
 			// Needed to avoid returning raw Message which matches Object
 			Object[] args = new Object[providedArgs.length + 1];
 			args[0] = message.getPayload();
@@ -95,7 +96,7 @@ public String getMethodAsString(Object payload) {
 			return this.invokerHandlerMethod.getMethod().toGenericString();
 		}
 		else {
-			return this.delegatingHandler.getMethodNameFor(payload);
+			return Objects.requireNonNull(this.delegatingHandler).getMethodNameFor(payload);
 		}
 	}
 
@@ -104,7 +105,7 @@ public Object getBean() {
 			return this.invokerHandlerMethod.getBean();
 		}
 		else {
-			return this.delegatingHandler.getBean();
+			return Objects.requireNonNull(this.delegatingHandler).getBean();
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapter.java

@@ -101,7 +101,7 @@ public void onMessage(ConsumerRecord<K, V> consumerRecord, @Nullable Acknowledgm
 		}
 	}
 
-	private void invokeDelegateOnMessage(ConsumerRecord<K, V> consumerRecord, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
+	private void invokeDelegateOnMessage(ConsumerRecord<K, V> consumerRecord, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer) {
 		switch (this.delegateType) {
 			case ACKNOWLEDGING_CONSUMER_AWARE -> this.delegate.onMessage(consumerRecord, acknowledgment, consumer);
 			case ACKNOWLEDGING -> this.delegate.onMessage(consumerRecord, acknowledgment);
@@ -111,7 +111,7 @@ private void invokeDelegateOnMessage(ConsumerRecord<K, V> consumerRecord, Acknow
 	}
 
 	private KafkaConsumerBackoffManager.Context createContext(ConsumerRecord<K, V> data, long nextExecutionTimestamp,
-			Consumer<?, ?> consumer) {
+			@Nullable Consumer<?, ?> consumer) {
 
 		return this.kafkaConsumerBackoffManager.createContext(nextExecutionTimestamp, this.listenerId,
 				new TopicPartition(data.topic(), data.partition()), consumer);
@@ -135,12 +135,12 @@ public void onMessage(ConsumerRecord<K, V> data) {
 	}
 
 	@Override
-	public void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment) {
+	public void onMessage(ConsumerRecord<K, V> data, @Nullable Acknowledgment acknowledgment) {
 		onMessage(data, acknowledgment, null); // NOSONAR
 	}
 
 	@Override
-	public void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer) {
+	public void onMessage(ConsumerRecord<K, V> data, @Nullable Consumer<?, ?> consumer) {
 		onMessage(data, null, consumer);
 	}
 }