Further changes for nullability in the config package

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

Author: sobychacko

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 @@ private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> aklEndpoint)
 	 * @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 void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		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 void setMainListenerId(@Nullable String id) {
 	}
 
 	@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 void setGroupId(@Nullable String groupId) {
 		this.groupId = groupId;
 	}
 
-	@Nullable
 	@Override
-	public String getGroupId() {
+	public @Nullable String getGroupId() {
 		return this.groupId;
 	}
 
@@ -233,9 +224,8 @@ public void setTopicPartitions(TopicPartitionOffset... topicPartitions) {
 	 * @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 void setTopicPattern(@Nullable Pattern topicPattern) {
 	 * 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 boolean isBatchListener() {
 	 * @since 2.8
 	 */
 	@Override
-	@Nullable
-	public Boolean getBatchListener() {
+	public @Nullable Boolean getBatchListener() {
 		return this.batchListener;
 	}
 
@@ -313,13 +300,11 @@ public void setReplyTemplate(KafkaTemplate<?, ?> replyTemplate) {
 		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 void setAckDiscarded(boolean ackDiscarded) {
 		this.ackDiscarded = ackDiscarded;
 	}
 
-	@Nullable
 	@Override
-	public String getClientIdPrefix() {
+	public @Nullable String getClientIdPrefix() {
 		return this.clientIdPrefix;
 	}
 
@@ -361,8 +345,7 @@ public void setClientIdPrefix(@Nullable String clientIdPrefix) {
 	}
 
 	@Override
-	@Nullable
-	public Integer getConcurrency() {
+	public @Nullable Integer getConcurrency() {
 		return this.concurrency;
 	}
 
@@ -376,8 +359,7 @@ public void setConcurrency(@Nullable Integer concurrency) {
 	}
 
 	@Override
-	@Nullable
-	public Boolean getAutoStartup() {
+	public @Nullable Boolean getAutoStartup() {
 		return this.autoStartup;
 	}
 
@@ -400,8 +382,7 @@ public void setReplyHeadersConfigurer(ReplyHeadersConfigurer replyHeadersConfigu
 	}
 
 	@Override
-	@Nullable
-	public Properties getConsumerProperties() {
+	public @Nullable Properties getConsumerProperties() {
 		return this.consumerProperties;
 	}
 
@@ -436,9 +417,8 @@ public void setSplitIterables(boolean splitIterables) {
 	}
 
 	@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 void setListenerInfo(byte[] listenerInfo) {  // NOSONAR
 		this.listenerInfo = listenerInfo; // NOSONAR
 	}
 
-	@Nullable
-	protected BatchToRecordAdapter<K, V> getBatchToRecordAdapter() {
+	protected @Nullable BatchToRecordAdapter<K, V> getBatchToRecordAdapter() {
 		return this.batchToRecordAdapter;
 	}
 
@@ -525,7 +504,7 @@ public void setupListenerContainer(MessageListenerContainer listenerContainer,
 	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 @@ private void setupMessageListener(MessageListenerContainer container,
 				.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 @@
  * <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 void setConcurrency(Integer concurrency) {
 
 	@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 @@ void setupListenerContainer(MessageListenerContainer listenerContainer,
 	 * @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 @@ public String getGroup() {
 		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 void setEndpointRegistry(KafkaListenerEndpointRegistry endpointRegistry)
 	 * @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 void setMessageHandlerMethodFactory(MessageHandlerMethodFactory kafkaHand
 	 * 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 void setBeanFactory(BeanFactory beanFactory) {
 	 * @return the validator.
 	 * @since 2.2
 	 */
-	@Nullable
-	public Validator getValidator() {
+	public @Nullable Validator getValidator() {
 		return this.validator;
 	}
 
@@ -279,13 +275,13 @@ public void registerEndpoint(KafkaListenerEndpoint endpoint) {
 	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;
 		}
 
+	}
+
 }