Further changes for nullability in the config package

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>

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

@@ -105,8 +105,7 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private @Nullable BatchToRecordAdapter<K, V> batchToRecordAdapter;
 
-	@SuppressWarnings("NullAway.Init")
-	private ApplicationContext applicationContext;
+	private @Nullable ApplicationContext applicationContext;
 
 	private @Nullable ContainerCustomizer<K, V, C> containerCustomizer;
 
@@ -404,7 +403,7 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 	 * @param instance the container instance to configure.
 	 * @param endpoint the endpoint.
 	 */
-	@SuppressWarnings({"deprecation", "NullAway"})
+	@SuppressWarnings({"NullAway"})
 	protected void initializeContainer(C instance, KafkaListenerEndpoint endpoint) {
 		ContainerProperties properties = instance.getContainerProperties();
 		BeanUtils.copyProperties(this.containerProperties, properties, "topics", "topicPartitions", "topicPattern",

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

@@ -82,8 +82,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private final Collection<TopicPartitionOffset> topicPartitions = new ArrayList<>();
 
-	@SuppressWarnings("NullAway.Init")
-	private BeanFactory beanFactory;
+	private @Nullable BeanFactory beanFactory;
 
 	private @Nullable BeanExpressionResolver resolver;
 
@@ -115,8 +114,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private @Nullable BatchToRecordAdapter<K, V> batchToRecordAdapter;
 
-	@SuppressWarnings("NullAway.Init")
-	private byte[] listenerInfo;
+	private byte @Nullable [] listenerInfo;
 
 	private @Nullable String correlationHeaderName;
 
@@ -135,23 +133,19 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 		this.beanResolver = new BeanFactoryResolver(beanFactory);
 	}
 
-	@Nullable
-	protected BeanFactory getBeanFactory() {
+	protected @Nullable BeanFactory getBeanFactory() {
 		return this.beanFactory;
 	}
 
-	@Nullable
-	protected BeanExpressionResolver getResolver() {
+	protected @Nullable BeanExpressionResolver getResolver() {
 		return this.resolver;
 	}
 
-	@Nullable
-	protected BeanExpressionContext getBeanExpressionContext() {
+	protected @Nullable BeanExpressionContext getBeanExpressionContext() {
 		return this.expressionContext;
 	}
 
-	@Nullable
-	protected BeanResolver getBeanResolver() {
+	protected @Nullable BeanResolver getBeanResolver() {
 		return this.beanResolver;
 	}
 
@@ -164,14 +158,12 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	}
 
 	@Override
-	@Nullable
-	public String getMainListenerId() {
+	public @Nullable String getMainListenerId() {
 		return this.mainListenerId;
 	}
 
-	@Nullable
 	@Override
-	public String getId() {
+	public @Nullable String getId() {
 		return this.id;
 	}
 
@@ -185,9 +177,8 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 		this.groupId = groupId;
 	}
 
-	@Nullable
 	@Override
-	public String getGroupId() {
+	public @Nullable String getGroupId() {
 		return this.groupId;
 	}
 
@@ -233,9 +224,8 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	 * @return the topicPartitions for this endpoint.
 	 * @since 2.3
 	 */
-	@Nullable
 	@Override
-	public TopicPartitionOffset[] getTopicPartitionsToAssign() {
+	public TopicPartitionOffset @Nullable [] getTopicPartitionsToAssign() {
 		return this.topicPartitions.toArray(new TopicPartitionOffset[0]);
 	}
 
@@ -254,15 +244,13 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	 * Return the topicPattern for this endpoint.
 	 * @return the topicPattern for this endpoint.
 	 */
-	@Nullable
 	@Override
-	public Pattern getTopicPattern() {
+	public @Nullable Pattern getTopicPattern() {
 		return this.topicPattern;
 	}
 
-	@Nullable
 	@Override
-	public String getGroup() {
+	public @Nullable String getGroup() {
 		return this.group;
 	}
 
@@ -290,8 +278,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	 * @since 2.8
 	 */
 	@Override
-	@Nullable
-	public Boolean getBatchListener() {
+	public @Nullable Boolean getBatchListener() {
 		return this.batchListener;
 	}
 
@@ -313,13 +300,11 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 		this.replyTemplate = replyTemplate;
 	}
 
-	@Nullable
-	protected KafkaTemplate<?, ?> getReplyTemplate() {
+	protected @Nullable KafkaTemplate<?, ?> getReplyTemplate() {
 		return this.replyTemplate;
 	}
 
-	@Nullable
-	protected RecordFilterStrategy<? super K, ? super V> getRecordFilterStrategy() {
+	protected @Nullable RecordFilterStrategy<? super K, ? super V> getRecordFilterStrategy() {
 		return this.recordFilterStrategy;
 	}
 
@@ -344,9 +329,8 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 		this.ackDiscarded = ackDiscarded;
 	}
 
-	@Nullable
 	@Override
-	public String getClientIdPrefix() {
+	public @Nullable String getClientIdPrefix() {
 		return this.clientIdPrefix;
 	}
 
@@ -361,8 +345,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	}
 
 	@Override
-	@Nullable
-	public Integer getConcurrency() {
+	public @Nullable Integer getConcurrency() {
 		return this.concurrency;
 	}
 
@@ -376,8 +359,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	}
 
 	@Override
-	@Nullable
-	public Boolean getAutoStartup() {
+	public @Nullable Boolean getAutoStartup() {
 		return this.autoStartup;
 	}
 
@@ -400,8 +382,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	}
 
 	@Override
-	@Nullable
-	public Properties getConsumerProperties() {
+	public @Nullable Properties getConsumerProperties() {
 		return this.consumerProperties;
 	}
 
@@ -436,9 +417,8 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	}
 
 	@Override
-	@SuppressWarnings("NullAway") // Dataflow analysis limitation
-	public byte[] getListenerInfo() {
-		return this.listenerInfo; // NOSONAR
+	public byte @Nullable [] getListenerInfo() {
+		return this.listenerInfo;
 	}
 
 	/**
@@ -450,8 +430,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 		this.listenerInfo = listenerInfo; // NOSONAR
 	}
 
-	@Nullable
-	protected BatchToRecordAdapter<K, V> getBatchToRecordAdapter() {
+	protected @Nullable BatchToRecordAdapter<K, V> getBatchToRecordAdapter() {
 		return this.batchToRecordAdapter;
 	}
 
@@ -525,7 +504,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	protected abstract MessagingMessageListenerAdapter<K, V> createMessageListener(MessageListenerContainer container,
 			@Nullable MessageConverter messageConverter);
 
-	@SuppressWarnings({"unchecked", "NullAway"})
+	@SuppressWarnings("unchecked")
 	private void setupMessageListener(MessageListenerContainer container,
 			@Nullable MessageConverter messageConverter) {
 
@@ -535,8 +514,6 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 				.acceptIfNotNull(this.correlationHeaderName, adapter::setCorrelationHeaderName);
 		adapter.setSplitIterables(this.splitIterables);
 		Object messageListener = adapter;
-		Assert.state(messageListener != null,
-				() -> "Endpoint [" + this + "] must provide a non null message listener");
 		if (this.recordFilterStrategy != null) {
 			if (isBatchListener()) {
 				if (((MessagingMessageListenerAdapter<K, V>) messageListener).isConsumerRecords()) {

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

@@ -31,10 +31,10 @@ import org.springframework.util.Assert;
  * <p>
  * This should be the default for most users and a good transition paths for those that
  * are used to building such container definitions manually.
- *
+ * <p>
  * This factory is primarily for building containers for {@code KafkaListener} annotated
  * methods but can also be used to create any container.
- *
+ * <p>
  * Only containers for {@code KafkaListener} annotated methods are added to the
  * {@code KafkaListenerEndpointRegistry}.
  *
@@ -62,7 +62,7 @@ public class ConcurrentKafkaListenerContainerFactory<K, V>
 
 	@Override
 	protected ConcurrentMessageListenerContainer<K, V> createContainerInstance(KafkaListenerEndpoint endpoint) {
-		@Nullable TopicPartitionOffset[] topicPartitions = endpoint.getTopicPartitionsToAssign();
+		TopicPartitionOffset[] topicPartitions = endpoint.getTopicPartitionsToAssign();
 		if (topicPartitions != null && topicPartitions.length > 0) {
 			ContainerProperties properties = new ContainerProperties(topicPartitions);
 			return new ConcurrentMessageListenerContainer<>(getConsumerFactory(), properties);

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

@@ -74,8 +74,7 @@ public interface KafkaListenerEndpoint {
 	 * @return the topicPartitions for this endpoint.
 	 * @since 2.3
 	 */
-	@Nullable
-	TopicPartitionOffset[] getTopicPartitionsToAssign();
+	TopicPartitionOffset @Nullable [] getTopicPartitionsToAssign();
 
 	/**
 	 * Return the topicPattern for this endpoint.
@@ -152,8 +151,7 @@ public interface KafkaListenerEndpoint {
 	 * @return the info.
 	 * @since 2.8.4
 	 */
-	@SuppressWarnings("NullAway") // Dataflow analysis limitation
-	default byte[] getListenerInfo() {
+	default byte @Nullable [] getListenerInfo() {
 		return null;
 	}
 

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

@@ -38,21 +38,18 @@ class KafkaListenerEndpointAdapter implements KafkaListenerEndpoint {
 	KafkaListenerEndpointAdapter() {
 	}
 
-	@Nullable
 	@Override
-	public String getId() {
+	public @Nullable String getId() {
 		return null;
 	}
 
-	@Nullable
 	@Override
-	public String getGroupId() {
+	public @Nullable String getGroupId() {
 		return null;
 	}
 
-	@Nullable
 	@Override
-	public String getGroup() {
+	public @Nullable String getGroup() {
 		return null;
 	}
 
@@ -61,33 +58,28 @@ class KafkaListenerEndpointAdapter implements KafkaListenerEndpoint {
 		return Collections.emptyList();
 	}
 
-	@Nullable
 	@Override
-	public TopicPartitionOffset[] getTopicPartitionsToAssign() {
+	public TopicPartitionOffset @Nullable [] getTopicPartitionsToAssign() {
 		return new TopicPartitionOffset[0];
 	}
 
-	@Nullable
 	@Override
-	public Pattern getTopicPattern() {
+	public @Nullable Pattern getTopicPattern() {
 		return null;
 	}
 
-	@Nullable
 	@Override
-	public String getClientIdPrefix() {
+	public @Nullable String getClientIdPrefix() {
 		return null;
 	}
 
-	@Nullable
 	@Override
-	public Integer getConcurrency() {
+	public @Nullable Integer getConcurrency() {
 		return null;
 	}
 
-	@Nullable
 	@Override
-	public Boolean getAutoStartup() { // NOSONAR
+	public @Nullable Boolean getAutoStartup() { // NOSONAR
 		return null; // NOSONAR null check by caller
 	}
 

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

@@ -63,8 +63,7 @@ public class KafkaListenerEndpointRegistrar implements BeanFactoryAware, Initial
 
 	private @Nullable String containerFactoryBeanName;
 
-	@SuppressWarnings("NullAway.Init")
-	private BeanFactory beanFactory;
+	private @Nullable BeanFactory beanFactory;
 
 	private boolean startImmediately;
 
@@ -84,8 +83,7 @@ public class KafkaListenerEndpointRegistrar implements BeanFactoryAware, Initial
 	 * @return the {@link KafkaListenerEndpointRegistry} instance for this
 	 * registrar, may be {@code null}.
 	 */
-	@Nullable
-	public KafkaListenerEndpointRegistry getEndpointRegistry() {
+	public @Nullable KafkaListenerEndpointRegistry getEndpointRegistry() {
 		return this.endpointRegistry;
 	}
 
@@ -130,8 +128,7 @@ public class KafkaListenerEndpointRegistrar implements BeanFactoryAware, Initial
 	 * Return the custom {@link MessageHandlerMethodFactory} to use, if any.
 	 * @return the custom {@link MessageHandlerMethodFactory} to use, if any.
 	 */
-	@Nullable
-	public MessageHandlerMethodFactory getMessageHandlerMethodFactory() {
+	public @Nullable MessageHandlerMethodFactory getMessageHandlerMethodFactory() {
 		return this.messageHandlerMethodFactory;
 	}
 
@@ -173,8 +170,7 @@ public class KafkaListenerEndpointRegistrar implements BeanFactoryAware, Initial
 	 * @return the validator.
 	 * @since 2.2
 	 */
-	@Nullable
-	public Validator getValidator() {
+	public @Nullable Validator getValidator() {
 		return this.validator;
 	}
 
@@ -279,13 +275,13 @@ public class KafkaListenerEndpointRegistrar implements BeanFactoryAware, Initial
 	private record KafkaListenerEndpointDescriptor(KafkaListenerEndpoint endpoint,
 				@Nullable KafkaListenerContainerFactory<?> containerFactory) {
 
-			private KafkaListenerEndpointDescriptor(KafkaListenerEndpoint endpoint,
-					@Nullable KafkaListenerContainerFactory<?> containerFactory) {
-
-				this.endpoint = endpoint;
-				this.containerFactory = containerFactory;
-			}
+		private KafkaListenerEndpointDescriptor(KafkaListenerEndpoint endpoint,
+				@Nullable KafkaListenerContainerFactory<?> containerFactory) {
 
+			this.endpoint = endpoint;
+			this.containerFactory = containerFactory;
 		}
 
+	}
+
 }

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

@@ -21,7 +21,6 @@ import java.util.Properties;
 import org.apache.kafka.streams.KafkaClientSupplier;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.Topology;
-import org.jspecify.annotations.Nullable;
 
 /**
  * Callback interface that can be used to configure {@link KafkaStreams} directly.
@@ -49,7 +48,7 @@ public interface KafkaStreamsCustomizer {
 	 * @since 3.3.0
 	 */
 	default KafkaStreams initKafkaStreams(
-			@Nullable Topology topology,
+			Topology topology,
 			Properties properties,
 			KafkaClientSupplier clientSupplier
 	) {

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

@@ -68,8 +68,7 @@ public class MethodKafkaListenerEndpoint<K, V> extends AbstractKafkaListenerEndp
 	@SuppressWarnings("NullAway.Init")
 	private Method method;
 
-	@SuppressWarnings("NullAway.Init")
-	private MessageHandlerMethodFactory messageHandlerMethodFactory;
+	private @Nullable MessageHandlerMethodFactory messageHandlerMethodFactory;
 
 	private @Nullable KafkaListenerErrorHandler errorHandler;
 
@@ -195,6 +194,8 @@ public class MethodKafkaListenerEndpoint<K, V> extends AbstractKafkaListenerEndp
 	 * @return the handler adapter.
 	 */
 	protected HandlerAdapter configureListenerAdapter(MessagingMessageListenerAdapter<K, V> messageListener) {
+		Assert.state(this.messageHandlerMethodFactory != null,
+				"MessageHandlerMethodFactory must not be null");
 		InvocableHandlerMethod invocableHandlerMethod =
 				this.messageHandlerMethodFactory.createInvocableHandlerMethod(getBean(), getMethod());
 		return new HandlerAdapter(invocableHandlerMethod);
@@ -239,8 +240,7 @@ public class MethodKafkaListenerEndpoint<K, V> extends AbstractKafkaListenerEndp
 		return listener;
 	}
 
-	@SuppressWarnings("null")
-	private String resolve(String value) {
+	private @Nullable String resolve(String value) {
 		BeanExpressionContext beanExpressionContext = getBeanExpressionContext();
 		BeanExpressionResolver resolver = getResolver();
 		if (resolver != null && beanExpressionContext != null) {

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

@@ -87,6 +87,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	private KafkaClientSupplier clientSupplier = new DefaultKafkaClientSupplier();
 
+	@SuppressWarnings("NullAway.Init")
 	private Properties properties;
 
 	private CleanupConfig cleanupConfig;
@@ -115,7 +116,8 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	private volatile boolean running;
 
-	private @Nullable Topology topology;
+	@SuppressWarnings("NullAway.Init")
+	private Topology topology;
 
 	@SuppressWarnings("NullAway.Init")
 	private String beanName;
@@ -219,8 +221,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 	 * @return {@link StreamsUncaughtExceptionHandler}
 	 * @since 2.8.4
 	 */
-	@Nullable
-	public StreamsUncaughtExceptionHandler getStreamsUncaughtExceptionHandler() {
+	public @Nullable StreamsUncaughtExceptionHandler getStreamsUncaughtExceptionHandler() {
 		return this.streamsUncaughtExceptionHandler;
 	}
 
@@ -253,7 +254,6 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 	 * @return {@link Topology} object
 	 * @since 2.4.4
 	 */
-	@Nullable
 	public Topology getTopology() {
 		return this.topology;
 	}
@@ -353,9 +353,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 	@Override
 	public void stop(Runnable callback) {
 		stop();
-		if (callback != null) {
-			callback.run();
-		}
+		callback.run();
 	}
 
 	@Override
@@ -443,11 +441,8 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 		try {
 			this.topology = getObject().build(this.properties);
 			this.infrastructureCustomizer.configureTopology(this.topology);
-			if (this.topology != null) {
-				TopologyDescription description = this.topology.describe();
-				LOGGER.debug(description::toString);
-			}
-
+			TopologyDescription description = this.topology.describe();
+			LOGGER.debug(description::toString);
 		}
 		catch (Exception e) {
 			throw new RuntimeException(e);

spring-kafka/src/main/java/org/springframework/kafka/support/JavaUtils.java

@@ -81,7 +81,7 @@ public final class JavaUtils {
 	 * @param consumer the consumer.
 	 * @return this.
 	 */
-	public JavaUtils acceptIfHasText(String value, Consumer<String> consumer) {
+	public JavaUtils acceptIfHasText(@Nullable String value, Consumer<String> consumer) {
 		if (StringUtils.hasText(value)) {
 			consumer.accept(value);
 		}