Apply Nullability to spring-integration-kafka module

Related to: #10083

Signed-off-by: Jiandong Ma <[email protected]>

Author: mjd507

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 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,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.support.KafkaHeaders;
@@ -42,7 +44,7 @@ public abstract class AbstractKafkaChannel extends AbstractMessageChannel {
 
 	protected final String topic; // NOSONAR final
 
-	private String groupId;
+	private @Nullable String groupId;
 
 	/**
 	 * Construct an instance with the provided {@link KafkaOperations} and topic.
@@ -64,7 +66,7 @@ public void setGroupId(String groupId) {
 		this.groupId = groupId;
 	}
 
-	protected String getGroupId() {
+	protected @Nullable String getGroupId() {
 		return this.groupId;
 	}
 
@@ -82,7 +84,8 @@ protected boolean doSend(Message<?> message, long timeout) {
 			return false;
 		}
 		catch (ExecutionException e) {
-			this.logger.error(e.getCause(), () -> "Interrupted while waiting for send result for: " + message);
+			Throwable cause = e.getCause() != null ? e.getCause() : e;
+			this.logger.error(cause, () -> "Interrupted while waiting for send result for: " + message);
 			return false;
 		}
 		catch (TimeoutException e) {

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 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,12 +20,13 @@
 import java.util.Deque;
 import java.util.List;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.integration.channel.ExecutorChannelInterceptorAware;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource;
 import org.springframework.integration.support.management.metrics.CounterFacade;
 import org.springframework.integration.support.management.metrics.MetricsCaptor;
 import org.springframework.kafka.core.KafkaOperations;
-import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
@@ -47,7 +48,7 @@ public class PollableKafkaChannel extends AbstractKafkaChannel
 
 	private final KafkaMessageSource<?, ?> source;
 
-	private CounterFacade receiveCounter;
+	private @Nullable CounterFacade receiveCounter;
 
 	private volatile int executorInterceptorsSize;
 
@@ -197,8 +198,8 @@ public boolean hasExecutorInterceptors() {
 
 	private static String topic(KafkaMessageSource<?, ?> source) {
 		Assert.notNull(source, "'source' cannot be null");
-		String[] topics = source.getConsumerProperties().getTopics();
-		Assert.isTrue(topics != null && topics.length == 1, "Only one topic is allowed");
+		@Nullable String @Nullable [] topics = source.getConsumerProperties().getTopics();
+		Assert.isTrue(topics != null && topics.length == 1 && topics[0] != null, "Only one topic is allowed");
 		return topics[0];
 	}
 

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 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.
@@ -18,6 +18,7 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
@@ -56,8 +57,10 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su
 
 	private final IntegrationRecordMessageListener recordListener = new IntegrationRecordMessageListener();
 
+	@SuppressWarnings("NullAway.Init")
 	private MessageDispatcher dispatcher;
 
+	@SuppressWarnings("NullAway.Init")
 	private MessageListenerContainer container;
 
 	private boolean autoStartup = true;
@@ -183,8 +186,8 @@ private class IntegrationRecordMessageListener extends RecordMessagingMessageLis
 		}
 
 		@Override
-		public void onMessage(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment,
-				Consumer<?, ?> consumer) {
+		public void onMessage(ConsumerRecord<Object, Object> record, @Nullable Acknowledgment acknowledgment,
+				@Nullable Consumer<?, ?> consumer) {
 
 			SubscribableKafkaChannel.this.dispatcher.dispatch(toMessagingMessage(record, acknowledgment, consumer));
 		}

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/package-info.java

@@ -1,4 +1,5 @@
 /**
  * Provides classes related to message channel implementations for Apache Kafka.
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.kafka.channel;

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.kafka.config.xml;
 
+import org.jspecify.annotations.Nullable;
 import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
@@ -38,7 +39,7 @@
 public class KafkaChannelParser extends AbstractChannelParser {
 
 	@Override
-	protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
+	protected @Nullable BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder builder;
 		String factory = element.getAttribute("container-factory");
 		boolean hasFactory = StringUtils.hasText(factory);

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/package-info.java

@@ -1,4 +1,5 @@
 /**
  * Provides parser classes to provide Xml namespace support for the Apache Kafka components.
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.kafka.config.xml;

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2023 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.
@@ -16,9 +16,10 @@
 
 package org.springframework.integration.kafka.dsl;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.integration.dsl.MessageChannelSpec;
 import org.springframework.integration.kafka.channel.AbstractKafkaChannel;
-import org.springframework.lang.Nullable;
 
 /**
  *

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2023 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.
@@ -20,6 +20,7 @@
 
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.core.task.AsyncTaskExecutor;
 import org.springframework.integration.dsl.IntegrationComponentSpec;
@@ -28,7 +29,6 @@
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.support.TopicPartitionOffset;
-import org.springframework.lang.Nullable;
 
 /**
  * A helper class in the Builder pattern style to delegate options to the

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2023 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.
@@ -16,12 +16,13 @@
 
 package org.springframework.integration.kafka.dsl;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.integration.dsl.IntegrationComponentSpec;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.support.ProducerListener;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
-import org.springframework.lang.Nullable;
 
 /**
  * An {@link IntegrationComponentSpec} implementation for the {@link KafkaTemplate}.

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/package-info.java

@@ -1,6 +1,5 @@
 /**
  * Provides Spring Integration Java DSL Components support for Apache Kafka.
  */
-@org.springframework.lang.NonNullApi
-@org.springframework.lang.NonNullFields
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.kafka.dsl;