jspecify nullability changes in core and support packages

Kotlin related docs code changes

#3762

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

Author: sobychacko

spring-kafka-docs/src/main/kotlin/org/springframework/kafka/kdocs/requestreply/Application.kt

@@ -63,7 +63,9 @@ class Application {
     }
 
     @Bean
-    fun kafkaTemplate(pf: ProducerFactory<*, *>?) = KafkaTemplate(pf)
+    fun <K : Any, V : Any> kafkaTemplate(pf: ProducerFactory<K, V>?): KafkaTemplate<K, V> {
+        return KafkaTemplate<K, V>(pf!!)
+    }
 
 // tag::beans[]
     @Bean
@@ -75,7 +77,7 @@ class Application {
         replyContainer.containerProperties.groupId = "request.replies"
         val template = ReplyingKafkaTemplate<String, String, String>(pf, replyContainer)
         template.messageConverter = ByteArrayJsonMessageConverter()
-        template.defaultTopic = "requests"
+        template.setDefaultTopic("requests")
         return template
     }
 // end::beans[]

spring-kafka-docs/src/main/kotlin/org/springframework/kafka/kdocs/started/producer/Application.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-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.
@@ -38,9 +38,9 @@ class Application {
     fun topic() = NewTopic("topic1", 10, 1)
 
     @Bean
-    fun runner(template: KafkaTemplate<String?, String?>) =
+    fun runner(template: KafkaTemplate<String, String>) =
         ApplicationRunner { template.send("topic1", "test") }
-
+    
     companion object {
         @JvmStatic
         fun main(args: Array<String>) = runApplication<Application>(*args)

spring-kafka/src/main/java/org/springframework/kafka/core/ConsumerFactory.java

@@ -242,7 +242,7 @@ default void consumerAdded(String id, Consumer<K, V> consumer) {
 		 * period).
 		 * @param consumer the consumer.
 		 */
-		default void consumerRemoved(String id, Consumer<K, V> consumer) {
+		default void consumerRemoved(@Nullable String id, Consumer<K, V> consumer) {
 		}
 
 	}

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java

@@ -24,6 +24,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
@@ -87,15 +88,15 @@ public class DefaultKafkaConsumerFactory<K, V> extends KafkaResourceFactory
 
 	private final List<ConsumerPostProcessor<K, V>> postProcessors = new ArrayList<>();
 
-	private Supplier<Deserializer<K>> keyDeserializerSupplier;
+	private @Nullable Supplier<Deserializer<K>> keyDeserializerSupplier;
 
-	private Supplier<Deserializer<V>> valueDeserializerSupplier;
+	private @Nullable Supplier<Deserializer<V>> valueDeserializerSupplier;
 
 	private String beanName = "not.managed.by.Spring";
 
 	private boolean configureDeserializers = true;
 
-	private ApplicationContext applicationContext;
+	private @Nullable ApplicationContext applicationContext;
 
 	/**
 	 * Construct a factory with the provided configuration.
@@ -245,12 +246,12 @@ public Map<String, Object> getConfigurationProperties() {
 
 	@Override
 	public Deserializer<K> getKeyDeserializer() {
-		return this.keyDeserializerSupplier.get();
+		return Objects.requireNonNull(this.keyDeserializerSupplier).get();
 	}
 
 	@Override
 	public Deserializer<V> getValueDeserializer() {
-		return this.valueDeserializerSupplier.get();
+		return Objects.requireNonNull(this.valueDeserializerSupplier).get();
 	}
 
 	/**
@@ -492,7 +493,7 @@ private Deserializer<V> valueDeserializer(Map<String, Object> configs) {
 
 	protected class ExtendedKafkaConsumer extends KafkaConsumer<K, V> {
 
-		private String idForListeners;
+		private @Nullable String idForListeners;
 
 		protected ExtendedKafkaConsumer(Map<String, Object> configProps) {
 			super(configProps, keyDeserializer(configProps), valueDeserializer(configProps));

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

@@ -142,17 +142,17 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
 
 	private TransactionIdSuffixStrategy transactionIdSuffixStrategy = new DefaultTransactionIdSuffixStrategy(0);
 
-	private Supplier<Serializer<K>> keySerializerSupplier;
+	private @Nullable Supplier<Serializer<K>> keySerializerSupplier;
 
-	private Supplier<Serializer<V>> valueSerializerSupplier;
+	private @Nullable Supplier<Serializer<V>> valueSerializerSupplier;
 
-	private Supplier<Serializer<K>> rawKeySerializerSupplier;
+	private @Nullable Supplier<Serializer<K>> rawKeySerializerSupplier;
 
-	private Supplier<Serializer<V>> rawValueSerializerSupplier;
+	private @Nullable Supplier<Serializer<V>> rawValueSerializerSupplier;
 
 	private Duration physicalCloseTimeout = DEFAULT_PHYSICAL_CLOSE_TIMEOUT;
 
-	private ApplicationContext applicationContext;
+	private @Nullable ApplicationContext applicationContext;
 
 	private String beanName = "not.managed.by.Spring";
 
@@ -162,11 +162,11 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
 
 	private boolean configureSerializers = true;
 
-	private volatile String transactionIdPrefix;
+	private volatile @Nullable String transactionIdPrefix;
 
-	private volatile String clientIdPrefix;
+	private volatile @Nullable String clientIdPrefix;
 
-	private volatile CloseSafeProducer<K, V> producer;
+	private volatile @Nullable CloseSafeProducer<K, V> producer;
 
 	/**
 	 * Construct a factory with the provided configuration.
@@ -267,7 +267,7 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 		}
 	}
 
-	private Supplier<Serializer<K>> keySerializerSupplier(@Nullable Supplier<Serializer<K>> keySerializerSupplier) {
+	private @Nullable Supplier<Serializer<K>> keySerializerSupplier(@Nullable Supplier<Serializer<K>> keySerializerSupplier) {
 		this.rawKeySerializerSupplier = keySerializerSupplier;
 		if (!this.configureSerializers) {
 			return keySerializerSupplier;
@@ -283,7 +283,7 @@ private Supplier<Serializer<K>> keySerializerSupplier(@Nullable Supplier<Seriali
 		};
 	}
 
-	private Supplier<Serializer<V>> valueSerializerSupplier(@Nullable Supplier<Serializer<V>> valueSerializerSupplier) {
+	private @Nullable Supplier<Serializer<V>> valueSerializerSupplier(@Nullable Supplier<Serializer<V>> valueSerializerSupplier) {
 		this.rawValueSerializerSupplier = valueSerializerSupplier;
 		if (!this.configureSerializers) {
 			return valueSerializerSupplier;
@@ -456,21 +456,23 @@ public boolean isProducerPerThread() {
 	@Override
 	@Nullable
 	public Serializer<K> getKeySerializer() {
-		return this.keySerializerSupplier.get();
+		return this.keySerializerSupplier == null ? null : this.keySerializerSupplier.get();
 	}
 
 	@Override
 	@Nullable
 	public Serializer<V> getValueSerializer() {
-		return this.valueSerializerSupplier.get();
+		return this.valueSerializerSupplier == null ? null : this.valueSerializerSupplier.get();
 	}
 
 	@Override
+	@Nullable
 	public Supplier<Serializer<K>> getKeySerializerSupplier() {
 		return this.rawKeySerializerSupplier;
 	}
 
 	@Override
+	@Nullable
 	public Supplier<Serializer<V>> getValueSerializerSupplier() {
 		return this.rawValueSerializerSupplier;
 	}
@@ -546,9 +548,11 @@ public int getPhase() {
 	 *  properties applied
 	 */
 	@Override
-	public ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> overrideProperties) {
+	public ProducerFactory<K, V> copyWithConfigurationOverride(@Nullable Map<String, Object> overrideProperties) {
 		Map<String, Object> producerProperties = new HashMap<>(getConfigurationProperties());
-		producerProperties.putAll(overrideProperties);
+		if (overrideProperties != null) {
+			producerProperties.putAll(overrideProperties);
+		}
 		producerProperties = ensureExistingTransactionIdPrefixInProperties(producerProperties);
 		DefaultKafkaProducerFactory<K, V> newFactory = new DefaultKafkaProducerFactory<>(producerProperties,
 				getKeySerializerSupplier(),
@@ -850,7 +854,7 @@ protected Producer<K, V> createTransactionalProducer() {
 		return createTransactionalProducer(this.transactionIdPrefix);
 	}
 
-	protected Producer<K, V> createTransactionalProducer(String txIdPrefix) {
+	protected Producer<K, V> createTransactionalProducer(@Nullable String txIdPrefix) {
 		BlockingQueue<CloseSafeProducer<K, V>> queue = getCache(txIdPrefix);
 		Assert.notNull(queue, () -> "No cache found for " + txIdPrefix);
 		CloseSafeProducer<K, V> cachedProducer = queue.poll();
@@ -912,7 +916,7 @@ private void removeTransactionProducer(CloseSafeProducer<K, V> producer, Duratio
 		listeners.forEach(listener -> listener.producerRemoved(producer.clientId, producer));
 	}
 
-	private CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix,
+	private CloseSafeProducer<K, V> doCreateTxProducer(@Nullable String prefix, String suffix,
 			BiPredicate<CloseSafeProducer<K, V>, Duration> remover) {
 		Producer<K, V> newProducer = createRawProducer(getTxProducerConfigs(prefix + suffix));
 		try {
@@ -941,7 +945,8 @@ private CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix,
 
 	protected Producer<K, V> createRawProducer(Map<String, Object> rawConfigs) {
 		Producer<K, V> kafkaProducer =
-				new KafkaProducer<>(rawConfigs, this.keySerializerSupplier.get(), this.valueSerializerSupplier.get());
+				new KafkaProducer<>(rawConfigs, this.keySerializerSupplier == null ? null : this.keySerializerSupplier.get(),
+						this.valueSerializerSupplier == null ? null : this.valueSerializerSupplier.get());
 		for (ProducerPostProcessor<K, V> pp : this.postProcessors) {
 			kafkaProducer = pp.apply(kafkaProducer);
 		}
@@ -1033,9 +1038,9 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 
 		private final BiPredicate<CloseSafeProducer<K, V>, Duration> removeProducer;
 
-		final String txIdPrefix; // NOSONAR
+		final @Nullable String txIdPrefix; // NOSONAR
 
-		final String txIdSuffix;  // NOSONAR
+		final @Nullable String txIdSuffix;  // NOSONAR
 
 		final long created; // NOSONAR
 
@@ -1045,7 +1050,7 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 
 		final int epoch; // NOSONAR
 
-		private volatile Exception producerFailed;
+		private volatile @Nullable Exception producerFailed;
 
 		volatile boolean closed; // NOSONAR
 
@@ -1186,8 +1191,11 @@ public void commitTransaction() throws ProducerFencedException {
 		public void abortTransaction() throws ProducerFencedException {
 			LOGGER.debug(() -> toString() + " abortTransaction()");
 			if (this.producerFailed != null) {
-				LOGGER.debug(() -> "abortTransaction ignored - previous txFailed: " + this.producerFailed.getMessage()
-						+ ": " + this);
+				LOGGER.debug(() -> {
+					String message = this.producerFailed == null ? "" : this.producerFailed.getMessage();
+					return "abortTransaction ignored - previous txFailed: " + message
+							+ ": " + this;
+				});
 			}
 			else {
 				try {

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultTransactionIdSuffixStrategy.java

@@ -64,7 +64,7 @@ public DefaultTransactionIdSuffixStrategy(int maxCache) {
 	 * @throws NoProducerAvailableException if caching is enabled and no suffixes are available.
 	 */
 	@Override
-	public String acquireSuffix(String txIdPrefix) {
+	public String acquireSuffix(@Nullable String txIdPrefix) {
 		Assert.notNull(txIdPrefix, "'txIdPrefix' must not be null");
 		BlockingQueue<String> cache = getSuffixCache(txIdPrefix);
 		if (cache == null) {
@@ -79,7 +79,7 @@ public String acquireSuffix(String txIdPrefix) {
 	}
 
 	@Override
-	public void releaseSuffix(String txIdPrefix, String suffix) {
+	public void releaseSuffix(@Nullable String txIdPrefix, @Nullable String suffix) {
 		Assert.notNull(txIdPrefix, "'txIdPrefix' must not be null");
 		Assert.notNull(suffix, "'suffix' must not be null");
 		if (this.maxCache <= 0) {

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java

@@ -102,7 +102,7 @@ public class KafkaAdmin extends KafkaResourceFactory
 
 	private final Map<String, Object> configs;
 
-	private ApplicationContext applicationContext;
+	private @Nullable ApplicationContext applicationContext;
 
 	private Predicate<NewTopic> createOrModifyTopic = nt -> true;
 
@@ -118,7 +118,7 @@ public class KafkaAdmin extends KafkaResourceFactory
 
 	private boolean modifyTopicConfigs;
 
-	private String clusterId;
+	private @Nullable String clusterId;
 
 	/**
 	 * Create an instance with an {@link Admin} based on the supplied
@@ -227,7 +227,7 @@ public void setClusterId(String clusterId) {
 	 * @return the cluster id.
 	 * @since 3.1.8
 	 */
-	public String getClusterId() {
+	public @Nullable String getClusterId() {
 		return this.clusterId;
 	}
 
@@ -321,6 +321,7 @@ private void updateClusterId(Admin adminClient) throws InterruptedException, Exe
 	 * @see #setCreateOrModifyTopic(Predicate)
 	 */
 	protected Collection<NewTopic> newTopics() {
+		Assert.state(this.applicationContext != null, "'applicationContext' cannot be null");
 		Map<String, NewTopic> newTopicsMap = new HashMap<>(
 				this.applicationContext.getBeansOfType(NewTopic.class, false, false));
 		Map<String, NewTopics> wrappers = this.applicationContext.getBeansOfType(NewTopics.class, false, false);
@@ -550,18 +551,20 @@ private Map<String, NewPartitions> checkPartitions(Map<String, NewTopic> topicNa
 		topicInfo.topicNameValues().forEach((n, f) -> {
 			NewTopic topic = topicNameToTopic.get(n);
 			try {
-				TopicDescription topicDescription = f.get(this.operationTimeout, TimeUnit.SECONDS);
-				if (topic.numPartitions() >= 0 && topic.numPartitions() < topicDescription.partitions().size()) {
-					LOGGER.info(() -> String.format(
-						"Topic '%s' exists but has a different partition count: %d not %d", n,
-						topicDescription.partitions().size(), topic.numPartitions()));
-				}
-				else if (topic.numPartitions() > topicDescription.partitions().size()) {
-					LOGGER.info(() -> String.format(
-						"Topic '%s' exists but has a different partition count: %d not %d, increasing "
-						+ "if the broker supports it", n,
-						topicDescription.partitions().size(), topic.numPartitions()));
-					topicsToModify.put(n, NewPartitions.increaseTo(topic.numPartitions()));
+				if (topic != null) {
+					TopicDescription topicDescription = f.get(this.operationTimeout, TimeUnit.SECONDS);
+					if (topic.numPartitions() >= 0 && topic.numPartitions() < topicDescription.partitions().size()) {
+						LOGGER.info(() -> String.format(
+								"Topic '%s' exists but has a different partition count: %d not %d", n,
+								topicDescription.partitions().size(), topic.numPartitions()));
+					}
+					else if (topic.numPartitions() > topicDescription.partitions().size()) {
+						LOGGER.info(() -> String.format(
+								"Topic '%s' exists but has a different partition count: %d not %d, increasing "
+										+ "if the broker supports it", n,
+								topicDescription.partitions().size(), topic.numPartitions()));
+						topicsToModify.put(n, NewPartitions.increaseTo(topic.numPartitions()));
+					}
 				}
 			}
 			catch (@SuppressWarnings("unused") InterruptedException e) {
@@ -594,7 +597,7 @@ private void addTopics(Admin adminClient, List<NewTopic> topicsToAdd) {
 				LOGGER.debug(e.getCause(), "Failed to create topics");
 			}
 			else {
-				LOGGER.error(e.getCause(), "Failed to create topics");
+				LOGGER.error(e.getCause() != null ? e.getCause() : e, "Failed to create topics");
 				throw new KafkaException("Failed to create topics", e.getCause()); // NOSONAR
 			}
 		}
@@ -617,7 +620,7 @@ private void createMissingPartitions(Admin adminClient, Map<String, NewPartition
 				LOGGER.debug(e.getCause(), "Failed to create partitions");
 			}
 			else {
-				LOGGER.error(e.getCause(), "Failed to create partitions");
+				LOGGER.error(e.getCause() != null ? e.getCause() : e, "Failed to create partitions");
 				if (!(e.getCause() instanceof UnsupportedVersionException)) {
 					throw new KafkaException("Failed to create partitions", e.getCause()); // NOSONAR
 				}

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaMetricsSupport.java

@@ -168,7 +168,7 @@ private KafkaClientMetrics createAdminMetrics(AdminClient adminClient, List<Tag>
 	 * @param id the unique identifier for the client to manage in store.
 	 * @param client the Kafka client instance to unbind.
 	 */
-	protected final void unbindClient(String id, C client) {
+	protected final void unbindClient(@Nullable String id, C client) {
 		AutoCloseable removed = (AutoCloseable) this.metrics.remove(id);
 		if (removed != null) {
 			try {

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaResourceFactory.java

@@ -31,7 +31,7 @@
  */
 public abstract class KafkaResourceFactory {
 
-	private Supplier<String> bootstrapServersSupplier;
+	private @Nullable Supplier<String> bootstrapServersSupplier;
 
 	@Nullable
 	protected String getBootstrapServers() {