jspecify nullability changes for the requestreply package

docs related Kotlin code changes

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

Author: sobychacko

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

@@ -68,33 +68,33 @@ class Application {
 // tag::beans[]
     @Bean
     fun template(
-        pf: ProducerFactory<String?, String>?,
-        factory: ConcurrentKafkaListenerContainerFactory<String?, String?>
-    ): ReplyingKafkaTemplate<String?, String, String?> {
+        pf: ProducerFactory<String, String>,
+        factory: ConcurrentKafkaListenerContainerFactory<String, String>
+    ): ReplyingKafkaTemplate<String, String, String> {
         val replyContainer = factory.createContainer("replies")
         replyContainer.containerProperties.groupId = "request.replies"
-        val template = ReplyingKafkaTemplate(pf, replyContainer)
+        val template = ReplyingKafkaTemplate<String, String, String>(pf, replyContainer)
         template.messageConverter = ByteArrayJsonMessageConverter()
         template.defaultTopic = "requests"
         return template
     }
 // end::beans[]
 
     @Bean
-    fun runner(template: ReplyingKafkaTemplate<String?, String?, String?>): ApplicationRunner {
+    fun runner(template: ReplyingKafkaTemplate<String, String, String>): ApplicationRunner {
         return ApplicationRunner { _ ->
 // tag::sendReceive[]
-            val future1: RequestReplyTypedMessageFuture<String?, String?, Thing> =
+            val future1: RequestReplyTypedMessageFuture<String, String, Thing> =
                 template.sendAndReceive(MessageBuilder.withPayload("getAThing").build(),
                     object : ParameterizedTypeReference<Thing>() {})
-            log.info(future1.sendFuture.get(10, TimeUnit.SECONDS).recordMetadata.toString())
+            log.info(future1.sendFuture?.get(10, TimeUnit.SECONDS)?.recordMetadata.toString())
             val thing = future1.get(10, TimeUnit.SECONDS).payload
             log.info(thing.toString())
 
-            val future2: RequestReplyTypedMessageFuture<String?, String?, List<Thing>> =
+            val future2: RequestReplyTypedMessageFuture<String, String, List<Thing>> =
                 template.sendAndReceive(MessageBuilder.withPayload("getThings").build(),
                     object : ParameterizedTypeReference<List<Thing>>() {})
-            log.info(future2.sendFuture.get(10, TimeUnit.SECONDS).recordMetadata.toString())
+            log.info(future2.sendFuture?.get(10, TimeUnit.SECONDS)?.recordMetadata.toString())
             val things = future2.get(10, TimeUnit.SECONDS).payload
             things.forEach { thing1 -> log.info(thing1.toString()) }
 // end::sendReceive[]

spring-kafka/src/main/java/org/springframework/kafka/requestreply/CorrelationKey.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2022 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.
@@ -18,6 +18,8 @@
 
 import java.util.Arrays;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.util.Assert;
 
 /**
@@ -33,9 +35,9 @@ public final class CorrelationKey {
 
 	private final byte[] correlationId;
 
-	private String asString;
+	private @Nullable String asString;
 
-	private volatile Integer hashCode;
+	private volatile @Nullable Integer hashCode;
 
 	public CorrelationKey(byte[] correlationId) { // NOSONAR array reference
 		Assert.notNull(correlationId, "'correlationId' cannot be null");

spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaOperations.java

@@ -96,7 +96,7 @@ default <P> RequestReplyTypedMessageFuture<K, V, P> sendAndReceive(Message<?> me
 	 * @since 2.7
 	 */
 	default <P> RequestReplyTypedMessageFuture<K, V, P> sendAndReceive(Message<?> message, Duration replyTimeout,
-			ParameterizedTypeReference<P> returnType) {
+			@Nullable ParameterizedTypeReference<P> returnType) {
 
 		throw new UnsupportedOperationException();
 	}

spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java

@@ -113,7 +113,7 @@ public class ReplyingKafkaTemplate<K, V, R> extends KafkaTemplate<K, V> implemen
 
 	private String replyPartitionHeaderName = KafkaHeaders.REPLY_PARTITION;
 
-	private Function<ConsumerRecord<?, ?>, Exception> replyErrorChecker = rec -> null;
+	private Function<ConsumerRecord<?, ?>, @Nullable Exception> replyErrorChecker = rec -> null;
 
 	private CountDownLatch assignLatch = new CountDownLatch(1);
 
@@ -127,6 +127,7 @@ public ReplyingKafkaTemplate(ProducerFactory<K, V> producerFactory,
 		this(producerFactory, replyContainer, false);
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	public ReplyingKafkaTemplate(ProducerFactory<K, V> producerFactory,
 			GenericMessageListenerContainer<K, R> replyContainer, boolean autoFlush) {
 
@@ -290,7 +291,7 @@ public void setReplyPartitionHeaderName(String replyPartitionHeaderName) {
 	 * @param replyErrorChecker the error checker function.
 	 * @since 2.6.7
 	 */
-	public void setReplyErrorChecker(Function<ConsumerRecord<?, ?>, Exception> replyErrorChecker) {
+	public void setReplyErrorChecker(Function<ConsumerRecord<?, ?>, @Nullable Exception> replyErrorChecker) {
 		Assert.notNull(replyErrorChecker, "'replyErrorChecker' cannot be null");
 		this.replyErrorChecker = replyErrorChecker;
 	}
@@ -363,7 +364,7 @@ public RequestReplyMessageFuture<K, V> sendAndReceive(Message<?> message) {
 	}
 
 	@Override
-	public RequestReplyMessageFuture<K, V> sendAndReceive(Message<?> message, Duration replyTimeout) {
+	public RequestReplyMessageFuture<K, V> sendAndReceive(Message<?> message, @Nullable Duration replyTimeout) {
 		return sendAndReceive(message, replyTimeout, null);
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/requestreply/RequestReplyFuture.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2022 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.
@@ -19,6 +19,7 @@
 import java.util.concurrent.CompletableFuture;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.kafka.support.SendResult;
 
@@ -35,7 +36,7 @@
  */
 public class RequestReplyFuture<K, V, R> extends CompletableFuture<ConsumerRecord<K, R>> {
 
-	private volatile CompletableFuture<SendResult<K, V>> sendFuture;
+	private volatile @Nullable CompletableFuture<SendResult<K, V>> sendFuture;
 
 	protected void setSendFuture(CompletableFuture<SendResult<K, V>> sendFuture) {
 		this.sendFuture = sendFuture;
@@ -45,7 +46,7 @@ protected void setSendFuture(CompletableFuture<SendResult<K, V>> sendFuture) {
 	 * Return the send future.
 	 * @return the send future.
 	 */
-	public CompletableFuture<SendResult<K, V>> getSendFuture() {
+	public @Nullable CompletableFuture<SendResult<K, V>> getSendFuture() {
 		return this.sendFuture;
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/requestreply/RequestReplyMessageFuture.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2022 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.
@@ -18,6 +18,8 @@
 
 import java.util.concurrent.CompletableFuture;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.kafka.support.SendResult;
 import org.springframework.messaging.Message;
 
@@ -33,17 +35,17 @@
  */
 public class RequestReplyMessageFuture<K, V> extends CompletableFuture<Message<?>> {
 
-	private final CompletableFuture<SendResult<K, V>> sendFuture; // NOSONAR
+	private final @Nullable CompletableFuture<SendResult<K, V>> sendFuture; // NOSONAR
 
-	RequestReplyMessageFuture(CompletableFuture<SendResult<K, V>> sendFuture) {
+	RequestReplyMessageFuture(@Nullable CompletableFuture<SendResult<K, V>> sendFuture) {
 		this.sendFuture = sendFuture;
 	}
 
 	/**
 	 * Return the send future.
 	 * @return the send future.
 	 */
-	public CompletableFuture<SendResult<K, V>> getSendFuture() {
+	public @Nullable CompletableFuture<SendResult<K, V>> getSendFuture() {
 		return this.sendFuture;
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/requestreply/RequestReplyTypedMessageFuture.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2022 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.
@@ -21,6 +21,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.kafka.support.SendResult;
 import org.springframework.messaging.Message;
 
@@ -37,7 +39,7 @@
  */
 public class RequestReplyTypedMessageFuture<K, V, P> extends RequestReplyMessageFuture<K, V> {
 
-	RequestReplyTypedMessageFuture(CompletableFuture<SendResult<K, V>> sendFuture) {
+	RequestReplyTypedMessageFuture(@Nullable CompletableFuture<SendResult<K, V>> sendFuture) {
 		super(sendFuture);
 	}
 

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

@@ -1,4 +1,5 @@
 /**
  * Provides classes for request/reply semantics.
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.kafka.requestreply;