jspecify nullability changes in support.converter package

#3762

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

Author: sobychacko

spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java

@@ -79,7 +79,7 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
 
 	private boolean generateTimestamp = false;
 
-	private KafkaHeaderMapper headerMapper;
+	private @Nullable KafkaHeaderMapper headerMapper;
 
 	private boolean rawRecordHeader;
 
@@ -232,15 +232,17 @@ private void addRecordInfo(ConsumerRecord<?, ?> record, Type type, List<Object>
 		}
 	}
 
-	private Object obtainPayload(Type type, ConsumerRecord<?, ?> record, List<ConversionException> conversionFailures) {
+	private @Nullable Object obtainPayload(Type type, ConsumerRecord<?, ?> record, List<ConversionException> conversionFailures) {
 		return this.recordConverter == null || !containerType(type)
 				? extractAndConvertValue(record, type)
 				: convert(record, type, conversionFailures);
 	}
 
 	private Map<String, Object> convertHeaders(Headers headers, List<Map<String, Object>> convertedHeaders) {
 		Map<String, Object> converted = new HashMap<>();
-		this.headerMapper.toHeaders(headers, converted);
+		if (this.headerMapper != null) {
+			this.headerMapper.toHeaders(headers, converted);
+		}
 		convertedHeaders.add(converted);
 		return converted;
 	}
@@ -269,12 +271,17 @@ protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type)
 	 * @param conversionFailures Conversion failures.
 	 * @return the converted payload.
 	 */
-	protected Object convert(ConsumerRecord<?, ?> record, Type type, List<ConversionException> conversionFailures) {
+	protected @Nullable Object convert(ConsumerRecord<?, ?> record, Type type, List<ConversionException> conversionFailures) {
 		try {
-			Object payload = this.recordConverter
-					.toMessage(record, null, null, ((ParameterizedType) type).getActualTypeArguments()[0]).getPayload();
-			conversionFailures.add(null);
-			return payload;
+			if (this.recordConverter != null) {
+				Object payload = this.recordConverter
+						.toMessage(record, null, null, ((ParameterizedType) type).getActualTypeArguments()[0]).getPayload();
+				conversionFailures.add(null);
+				return payload;
+			}
+			else {
+				return null;
+			}
 		}
 		catch (ConversionException ex) {
 			byte[] original = null;

spring-kafka/src/main/java/org/springframework/kafka/support/converter/ByteArrayJsonMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2024 the original author or authors.
+ * Copyright 2019-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 com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.messaging.Message;
@@ -44,7 +45,7 @@ public ByteArrayJsonMessageConverter(ObjectMapper objectMapper) {
 	}
 
 	@Override
-	protected Object convertPayload(Message<?> message) {
+	protected @Nullable Object convertPayload(Message<?> message) {
 		try {
 			return message.getPayload() instanceof KafkaNull
 					? null

spring-kafka/src/main/java/org/springframework/kafka/support/converter/BytesJsonMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2024 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 com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.kafka.common.utils.Bytes;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.messaging.Message;
@@ -45,7 +46,7 @@ public BytesJsonMessageConverter(ObjectMapper objectMapper) {
 	}
 
 	@Override
-	protected Object convertPayload(Message<?> message) {
+	protected @Nullable Object convertPayload(Message<?> message) {
 		try {
 			return message.getPayload() instanceof KafkaNull
 					? null

spring-kafka/src/main/java/org/springframework/kafka/support/converter/ConversionException.java

@@ -35,18 +35,18 @@
 @SuppressWarnings("serial")
 public class ConversionException extends KafkaException {
 
-	private transient ConsumerRecord<?, ?> record;
+	private transient @Nullable ConsumerRecord<?, ?> record;
 
-	private transient List<ConsumerRecord<?, ?>> records = new ArrayList<>();
+	private transient @Nullable List<ConsumerRecord<?, ?>> records = new ArrayList<>();
 
-	private transient Message<?> message;
+	private transient @Nullable Message<?> message;
 
 	/**
 	 * Construct an instance with the provided properties.
 	 * @param message A text message describing the reason.
 	 * @param cause the cause.
 	 */
-	public ConversionException(String message, Throwable cause) {
+	public ConversionException(String message, @Nullable Throwable cause) {
 		super(message, cause);
 		this.record = null;
 		this.message = null;
@@ -75,7 +75,9 @@ public ConversionException(String message, ConsumerRecord<?, ?> record, Throwabl
 	public ConversionException(String message, List<ConsumerRecord<?, ?>> records, Throwable cause) {
 		super(message, cause);
 		this.record = null;
-		this.records.addAll(records);
+		if (this.records != null) {
+			this.records.addAll(records);
+		}
 		this.message = null;
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/support/converter/JsonMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2024 the original author or authors.
+ * Copyright 2019-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.
@@ -26,6 +26,7 @@
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.apache.kafka.common.utils.Bytes;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.kafka.support.JacksonUtils;
 import org.springframework.kafka.support.KafkaNull;
@@ -91,13 +92,13 @@ protected Headers initialRecordHeaders(Message<?> message) {
 	}
 
 	@Override
-	protected Object convertPayload(Message<?> message) {
+	protected @Nullable Object convertPayload(Message<?> message) {
 		throw new UnsupportedOperationException("Select a subclass that creates a ProducerRecord value "
 				+ "corresponding to the configured Kafka Serializer");
 	}
 
 	@Override
-	protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type) {
+	protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, @Nullable Type type) {
 		Object value = record.value();
 		if (record.value() == null) {
 			return KafkaNull.INSTANCE;
@@ -128,7 +129,7 @@ else if (value instanceof byte[]) {
 		}
 	}
 
-	private JavaType determineJavaType(ConsumerRecord<?, ?> record, Type type) {
+	private JavaType determineJavaType(ConsumerRecord<?, ?> record, @Nullable Type type) {
 		JavaType javaType = this.typeMapper.getTypePrecedence().equals(TypePrecedence.INFERRED) && type != null
 				? TypeFactory.defaultInstance().constructType(type)
 				: this.typeMapper.toJavaType(record.headers());

spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessageConverter.java

@@ -56,7 +56,7 @@ static String getGroupId() {
 	 * @param timestampType the timestamp type.
 	 * @param timestamp the timestamp.
 	 */
-	default void commonHeaders(Acknowledgment acknowledgment, Consumer<?, ?> consumer, Map<String, Object> rawHeaders,
+	default void commonHeaders(@Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer, Map<String, Object> rawHeaders,
 			@Nullable Object theKey, Object topic, Object partition, Object offset,
 			@Nullable Object timestampType, Object timestamp) {
 

spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2024 the original author or authors.
+ * Copyright 2016-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.
@@ -29,6 +29,7 @@
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.support.AbstractKafkaHeaderMapper;
@@ -62,7 +63,7 @@ public class MessagingMessageConverter implements RecordMessageConverter {
 
 	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR
 
-	private final Function<Message<?>, Integer> partitionProvider;
+	private final Function<Message<?>, @Nullable Integer> partitionProvider;
 
 	private boolean generateMessageId = false;
 
@@ -72,7 +73,7 @@ public class MessagingMessageConverter implements RecordMessageConverter {
 
 	private boolean rawRecordHeader;
 
-	private SmartMessageConverter messagingConverter;
+	private @Nullable SmartMessageConverter messagingConverter;
 
 	/**
 	 * Construct an instance that uses the {@link KafkaHeaders#PARTITION} to determine the
@@ -88,7 +89,7 @@ public MessagingMessageConverter() {
 	 * @param partitionProvider the provider.
 	 * @since 3.0.8
 	 */
-	public MessagingMessageConverter(Function<Message<?>, Integer> partitionProvider) {
+	public MessagingMessageConverter(Function<Message<?>, @Nullable Integer> partitionProvider) {
 		Assert.notNull(partitionProvider, "'partitionProvider' cannot be null");
 		if (JacksonPresent.isJackson2Present()) {
 			this.headerMapper = new DefaultKafkaHeaderMapper();
@@ -136,6 +137,7 @@ public void setRawRecordHeader(boolean rawRecordHeader) {
 		this.rawRecordHeader = rawRecordHeader;
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	protected org.springframework.messaging.converter.MessageConverter getMessagingConverter() {
 		return this.messagingConverter;
 	}
@@ -165,16 +167,16 @@ protected org.springframework.messaging.converter.MessageConverter getMessagingC
 	 * @param messagingConverter the converter.
 	 * @since 2.7.1
 	 */
-	public void setMessagingConverter(SmartMessageConverter messagingConverter) {
+	public void setMessagingConverter(@Nullable SmartMessageConverter messagingConverter) {
 		this.messagingConverter = messagingConverter;
 		if (messagingConverter != null && this.headerMapper instanceof AbstractKafkaHeaderMapper) {
 			((AbstractKafkaHeaderMapper) this.headerMapper).addRawMappedHeader(MessageHeaders.CONTENT_TYPE, true);
 		}
 	}
 
 	@Override
-	public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer,
-			Type type) {
+	public Message<?> toMessage(ConsumerRecord<?, ?> record, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer,
+			@Nullable Type type) {
 
 		KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(this.generateMessageId,
 				this.generateTimestamp);
@@ -272,7 +274,7 @@ protected Headers initialRecordHeaders(Message<?> message) {
 	 * @param message the message.
 	 * @return the payload.
 	 */
-	protected Object convertPayload(Message<?> message) {
+	protected @Nullable Object convertPayload(Message<?> message) {
 		Object payload = message.getPayload();
 		if (payload instanceof KafkaNull) {
 			return null;
@@ -289,7 +291,7 @@ protected Object convertPayload(Message<?> message) {
 	 * @param type the required type.
 	 * @return the value.
 	 */
-	protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type) {
+	protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, @Nullable Type type) {
 		return record.value() == null ? KafkaNull.INSTANCE : record.value();
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/support/converter/ProjectingMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2024 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.
@@ -25,6 +25,7 @@
 import com.jayway.jsonpath.spi.mapper.JacksonMappingProvider;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.utils.Bytes;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.data.projection.MethodInterceptorFactory;
@@ -99,12 +100,12 @@ public ProjectingMessageConverter(ObjectMapper mapper, MessagingMessageConverter
 	}
 
 	@Override
-	protected Object convertPayload(Message<?> message) {
+	protected @Nullable Object convertPayload(Message<?> message) {
 		return this.delegate.convertPayload(message);
 	}
 
 	@Override
-	protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type) {
+	protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, @Nullable Type type) {
 		Object value = record.value();
 
 		if (value == null) {

spring-kafka/src/main/java/org/springframework/kafka/support/converter/RecordMessageConverter.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.jspecify.annotations.NonNull;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.messaging.Message;
@@ -43,8 +44,8 @@ public interface RecordMessageConverter extends MessageConverter {
 	 * @return the message.
 	 */
 	@NonNull
-	Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer,
-			Type payloadType);
+	Message<?> toMessage(ConsumerRecord<?, ?> record, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer,
+			@Nullable Type payloadType);
 
 	/**
 	 * Convert a message to a producer record.

spring-kafka/src/main/java/org/springframework/kafka/support/converter/StringJsonMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2024 the original author or authors.
+ * Copyright 2016-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 com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.messaging.Message;
@@ -44,7 +45,7 @@ public StringJsonMessageConverter(ObjectMapper objectMapper) {
 	}
 
 	@Override
-	protected Object convertPayload(Message<?> message) {
+	protected @Nullable Object convertPayload(Message<?> message) {
 		try {
 			return message.getPayload() instanceof KafkaNull
 					? null

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

@@ -1,4 +1,5 @@
 /**
  * Package for kafka converters
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.kafka.support.converter;