Add nullability annotations to module/spring-boot-kafka

See gh-46587

Author: mhalbritter

module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/ConcurrentKafkaListenerContainerFactoryConfigurer.java

@@ -19,6 +19,8 @@
 import java.time.Duration;
 import java.util.function.Function;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.boot.kafka.autoconfigure.KafkaProperties.Listener;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
@@ -36,6 +38,7 @@
 import org.springframework.kafka.support.converter.BatchMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
+import org.springframework.util.Assert;
 
 /**
  * Configure {@link ConcurrentKafkaListenerContainerFactory} with sensible defaults tuned
@@ -53,133 +56,133 @@
  */
 public class ConcurrentKafkaListenerContainerFactoryConfigurer {
 
-	private KafkaProperties properties;
+	private @Nullable KafkaProperties properties;
 
-	private BatchMessageConverter batchMessageConverter;
+	private @Nullable BatchMessageConverter batchMessageConverter;
 
-	private RecordMessageConverter recordMessageConverter;
+	private @Nullable RecordMessageConverter recordMessageConverter;
 
-	private RecordFilterStrategy<Object, Object> recordFilterStrategy;
+	private @Nullable RecordFilterStrategy<Object, Object> recordFilterStrategy;
 
-	private KafkaTemplate<Object, Object> replyTemplate;
+	private @Nullable KafkaTemplate<Object, Object> replyTemplate;
 
-	private KafkaAwareTransactionManager<Object, Object> transactionManager;
+	private @Nullable KafkaAwareTransactionManager<Object, Object> transactionManager;
 
-	private ConsumerAwareRebalanceListener rebalanceListener;
+	private @Nullable ConsumerAwareRebalanceListener rebalanceListener;
 
-	private CommonErrorHandler commonErrorHandler;
+	private @Nullable CommonErrorHandler commonErrorHandler;
 
-	private AfterRollbackProcessor<Object, Object> afterRollbackProcessor;
+	private @Nullable AfterRollbackProcessor<Object, Object> afterRollbackProcessor;
 
-	private RecordInterceptor<Object, Object> recordInterceptor;
+	private @Nullable RecordInterceptor<Object, Object> recordInterceptor;
 
-	private BatchInterceptor<Object, Object> batchInterceptor;
+	private @Nullable BatchInterceptor<Object, Object> batchInterceptor;
 
-	private Function<MessageListenerContainer, String> threadNameSupplier;
+	private @Nullable Function<MessageListenerContainer, String> threadNameSupplier;
 
-	private SimpleAsyncTaskExecutor listenerTaskExecutor;
+	private @Nullable SimpleAsyncTaskExecutor listenerTaskExecutor;
 
 	/**
 	 * Set the {@link KafkaProperties} to use.
 	 * @param properties the properties
 	 */
-	void setKafkaProperties(KafkaProperties properties) {
+	void setKafkaProperties(@Nullable KafkaProperties properties) {
 		this.properties = properties;
 	}
 
 	/**
 	 * Set the {@link BatchMessageConverter} to use.
 	 * @param batchMessageConverter the message converter
 	 */
-	void setBatchMessageConverter(BatchMessageConverter batchMessageConverter) {
+	void setBatchMessageConverter(@Nullable BatchMessageConverter batchMessageConverter) {
 		this.batchMessageConverter = batchMessageConverter;
 	}
 
 	/**
 	 * Set the {@link RecordMessageConverter} to use.
 	 * @param recordMessageConverter the message converter
 	 */
-	void setRecordMessageConverter(RecordMessageConverter recordMessageConverter) {
+	void setRecordMessageConverter(@Nullable RecordMessageConverter recordMessageConverter) {
 		this.recordMessageConverter = recordMessageConverter;
 	}
 
 	/**
 	 * Set the {@link RecordFilterStrategy} to use to filter incoming records.
 	 * @param recordFilterStrategy the record filter strategy
 	 */
-	void setRecordFilterStrategy(RecordFilterStrategy<Object, Object> recordFilterStrategy) {
+	void setRecordFilterStrategy(@Nullable RecordFilterStrategy<Object, Object> recordFilterStrategy) {
 		this.recordFilterStrategy = recordFilterStrategy;
 	}
 
 	/**
 	 * Set the {@link KafkaTemplate} to use to send replies.
 	 * @param replyTemplate the reply template
 	 */
-	void setReplyTemplate(KafkaTemplate<Object, Object> replyTemplate) {
+	void setReplyTemplate(@Nullable KafkaTemplate<Object, Object> replyTemplate) {
 		this.replyTemplate = replyTemplate;
 	}
 
 	/**
 	 * Set the {@link KafkaAwareTransactionManager} to use.
 	 * @param transactionManager the transaction manager
 	 */
-	void setTransactionManager(KafkaAwareTransactionManager<Object, Object> transactionManager) {
+	void setTransactionManager(@Nullable KafkaAwareTransactionManager<Object, Object> transactionManager) {
 		this.transactionManager = transactionManager;
 	}
 
 	/**
 	 * Set the {@link ConsumerAwareRebalanceListener} to use.
 	 * @param rebalanceListener the rebalance listener.
 	 */
-	void setRebalanceListener(ConsumerAwareRebalanceListener rebalanceListener) {
+	void setRebalanceListener(@Nullable ConsumerAwareRebalanceListener rebalanceListener) {
 		this.rebalanceListener = rebalanceListener;
 	}
 
 	/**
 	 * Set the {@link CommonErrorHandler} to use.
 	 * @param commonErrorHandler the error handler.
 	 */
-	public void setCommonErrorHandler(CommonErrorHandler commonErrorHandler) {
+	public void setCommonErrorHandler(@Nullable CommonErrorHandler commonErrorHandler) {
 		this.commonErrorHandler = commonErrorHandler;
 	}
 
 	/**
 	 * Set the {@link AfterRollbackProcessor} to use.
 	 * @param afterRollbackProcessor the after rollback processor
 	 */
-	void setAfterRollbackProcessor(AfterRollbackProcessor<Object, Object> afterRollbackProcessor) {
+	void setAfterRollbackProcessor(@Nullable AfterRollbackProcessor<Object, Object> afterRollbackProcessor) {
 		this.afterRollbackProcessor = afterRollbackProcessor;
 	}
 
 	/**
 	 * Set the {@link RecordInterceptor} to use.
 	 * @param recordInterceptor the record interceptor.
 	 */
-	void setRecordInterceptor(RecordInterceptor<Object, Object> recordInterceptor) {
+	void setRecordInterceptor(@Nullable RecordInterceptor<Object, Object> recordInterceptor) {
 		this.recordInterceptor = recordInterceptor;
 	}
 
 	/**
 	 * Set the {@link BatchInterceptor} to use.
 	 * @param batchInterceptor the batch interceptor.
 	 */
-	void setBatchInterceptor(BatchInterceptor<Object, Object> batchInterceptor) {
+	void setBatchInterceptor(@Nullable BatchInterceptor<Object, Object> batchInterceptor) {
 		this.batchInterceptor = batchInterceptor;
 	}
 
 	/**
 	 * Set the thread name supplier to use.
 	 * @param threadNameSupplier the thread name supplier to use
 	 */
-	void setThreadNameSupplier(Function<MessageListenerContainer, String> threadNameSupplier) {
+	void setThreadNameSupplier(@Nullable Function<MessageListenerContainer, String> threadNameSupplier) {
 		this.threadNameSupplier = threadNameSupplier;
 	}
 
 	/**
 	 * Set the executor for threads that poll the consumer.
 	 * @param listenerTaskExecutor task executor
 	 */
-	void setListenerTaskExecutor(SimpleAsyncTaskExecutor listenerTaskExecutor) {
+	void setListenerTaskExecutor(@Nullable SimpleAsyncTaskExecutor listenerTaskExecutor) {
 		this.listenerTaskExecutor = listenerTaskExecutor;
 	}
 
@@ -199,6 +202,7 @@ public void configure(ConcurrentKafkaListenerContainerFactory<Object, Object> li
 
 	private void configureListenerFactory(ConcurrentKafkaListenerContainerFactory<Object, Object> factory) {
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+		Assert.state(this.properties != null, "'properties' must not be null");
 		Listener properties = this.properties.getListener();
 		map.from(properties::getConcurrency).to(factory::setConcurrency);
 		map.from(properties::isAutoStartup).to(factory::setAutoStartup);
@@ -219,6 +223,7 @@ private void configureListenerFactory(ConcurrentKafkaListenerContainerFactory<Ob
 
 	private void configureContainer(ContainerProperties container) {
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+		Assert.state(this.properties != null, "'properties' must not be null");
 		Listener properties = this.properties.getListener();
 		map.from(properties::getAckMode).to(container::setAckMode);
 		map.from(properties::getAsyncAcks).to(container::setAsyncAcks);

module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaAnnotationDrivenConfiguration.java

@@ -18,6 +18,8 @@
 
 import java.util.function.Function;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -62,27 +64,27 @@ class KafkaAnnotationDrivenConfiguration {
 
 	private final KafkaProperties properties;
 
-	private final RecordMessageConverter recordMessageConverter;
+	private final @Nullable RecordMessageConverter recordMessageConverter;
 
-	private final RecordFilterStrategy<Object, Object> recordFilterStrategy;
+	private final @Nullable RecordFilterStrategy<Object, Object> recordFilterStrategy;
 
 	private final BatchMessageConverter batchMessageConverter;
 
-	private final KafkaTemplate<Object, Object> kafkaTemplate;
+	private final @Nullable KafkaTemplate<Object, Object> kafkaTemplate;
 
-	private final KafkaAwareTransactionManager<Object, Object> transactionManager;
+	private final @Nullable KafkaAwareTransactionManager<Object, Object> transactionManager;
 
-	private final ConsumerAwareRebalanceListener rebalanceListener;
+	private final @Nullable ConsumerAwareRebalanceListener rebalanceListener;
 
-	private final CommonErrorHandler commonErrorHandler;
+	private final @Nullable CommonErrorHandler commonErrorHandler;
 
-	private final AfterRollbackProcessor<Object, Object> afterRollbackProcessor;
+	private final @Nullable AfterRollbackProcessor<Object, Object> afterRollbackProcessor;
 
-	private final RecordInterceptor<Object, Object> recordInterceptor;
+	private final @Nullable RecordInterceptor<Object, Object> recordInterceptor;
 
-	private final BatchInterceptor<Object, Object> batchInterceptor;
+	private final @Nullable BatchInterceptor<Object, Object> batchInterceptor;
 
-	private final Function<MessageListenerContainer, String> threadNameSupplier;
+	private final @Nullable Function<MessageListenerContainer, String> threadNameSupplier;
 
 	KafkaAnnotationDrivenConfiguration(KafkaProperties properties,
 			ObjectProvider<RecordMessageConverter> recordMessageConverter,

module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaAutoConfiguration.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.config.SslConfigs;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.aot.hint.MemberCategory;
 import org.springframework.aot.hint.RuntimeHints;
@@ -240,14 +241,14 @@ private static void setBackOffPolicy(RetryTopicConfigurationBuilder builder, Bac
 		}
 	}
 
-	static void applySslBundle(Map<String, Object> properties, SslBundle sslBundle) {
+	static void applySslBundle(Map<String, Object> properties, @Nullable SslBundle sslBundle) {
 		if (sslBundle != null) {
 			properties.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, SslBundleSslEngineFactory.class);
 			properties.put(SslBundle.class.getName(), sslBundle);
 		}
 	}
 
-	static void applySecurityProtocol(Map<String, Object> properties, String securityProtocol) {
+	static void applySecurityProtocol(Map<String, Object> properties, @Nullable String securityProtocol) {
 		if (StringUtils.hasLength(securityProtocol)) {
 			properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
 		}
@@ -256,7 +257,7 @@ static void applySecurityProtocol(Map<String, Object> properties, String securit
 	static class KafkaRuntimeHints implements RuntimeHintsRegistrar {
 
 		@Override
-		public void registerHints(RuntimeHints hints, ClassLoader classLoader) {
+		public void registerHints(RuntimeHints hints, @Nullable ClassLoader classLoader) {
 			hints.reflection().registerType(SslBundleSslEngineFactory.class, MemberCategory.INVOKE_PUBLIC_CONSTRUCTORS);
 		}
 

module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaConnectionDetails.java

@@ -18,6 +18,8 @@
 
 import java.util.List;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.boot.autoconfigure.service.connection.ConnectionDetails;
 import org.springframework.boot.ssl.SslBundle;
 
@@ -41,15 +43,15 @@ public interface KafkaConnectionDetails extends ConnectionDetails {
 	 * Returns the SSL bundle.
 	 * @return the SSL bundle
 	 */
-	default SslBundle getSslBundle() {
+	default @Nullable SslBundle getSslBundle() {
 		return null;
 	}
 
 	/**
 	 * Returns the security protocol.
 	 * @return the security protocol
 	 */
-	default String getSecurityProtocol() {
+	default @Nullable String getSecurityProtocol() {
 		return null;
 	}
 
@@ -117,20 +119,21 @@ static Configuration of(List<String> bootstrapServers, SslBundle sslBundle) {
 		 * @param securityProtocol the security protocol
 		 * @return the configuration
 		 */
-		static Configuration of(List<String> bootstrapServers, SslBundle sslBundle, String securityProtocol) {
+		static Configuration of(List<String> bootstrapServers, @Nullable SslBundle sslBundle,
+				@Nullable String securityProtocol) {
 			return new Configuration() {
 				@Override
 				public List<String> getBootstrapServers() {
 					return bootstrapServers;
 				}
 
 				@Override
-				public SslBundle getSslBundle() {
+				public @Nullable SslBundle getSslBundle() {
 					return sslBundle;
 				}
 
 				@Override
-				public String getSecurityProtocol() {
+				public @Nullable String getSecurityProtocol() {
 					return securityProtocol;
 				}
 			};
@@ -146,15 +149,15 @@ public String getSecurityProtocol() {
 		 * Returns the SSL bundle.
 		 * @return the SSL bundle
 		 */
-		default SslBundle getSslBundle() {
+		default @Nullable SslBundle getSslBundle() {
 			return null;
 		}
 
 		/**
 		 * Returns the security protocol.
 		 * @return the security protocol
 		 */
-		default String getSecurityProtocol() {
+		default @Nullable String getSecurityProtocol() {
 			return null;
 		}