diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/AbstractJavaTypeMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/AbstractJavaTypeMapper.java
index 877538a326..c24411c01f 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/AbstractJavaTypeMapper.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/AbstractJavaTypeMapper.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-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.
@@ -24,6 +24,7 @@
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeader;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.messaging.converter.MessageConversionException;
@@ -82,7 +83,7 @@ public abstract class AbstractJavaTypeMapper implements BeanClassLoaderAware {
 
 	private String keyClassIdFieldName = DEFAULT_KEY_CLASSID_FIELD_NAME;
 
-	private ClassLoader classLoader = ClassUtils.getDefaultClassLoader();
+	private @Nullable ClassLoader classLoader = ClassUtils.getDefaultClassLoader();
 
 	public String getClassIdFieldName() {
 		return this.classIdFieldName;
@@ -133,7 +134,7 @@ public void setBeanClassLoader(ClassLoader classLoader) {
 		this.classLoader = classLoader;
 	}
 
-	protected ClassLoader getClassLoader() {
+	protected @Nullable ClassLoader getClassLoader() {
 		return this.classLoader;
 	}
 
@@ -155,7 +156,7 @@ protected String retrieveHeader(Headers headers, String headerName) {
 		return classId;
 	}
 
-	protected String retrieveHeaderAsString(Headers headers, String headerName) {
+	protected @Nullable String retrieveHeaderAsString(Headers headers, String headerName) {
 		Header header = headers.lastHeader(headerName);
 		if (header != null) {
 			String classId = null;
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/ClassMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/ClassMapper.java
index 03e46a354c..28bc5ab502 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/ClassMapper.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/ClassMapper.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-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.
@@ -17,6 +17,7 @@
 package org.springframework.kafka.support.mapping;
 
 import org.apache.kafka.common.header.Headers;
+import org.jspecify.annotations.Nullable;
 
 /**
  * Strategy for setting metadata on messages such that one can create the class
@@ -32,6 +33,7 @@ public interface ClassMapper {
 
 	void fromClass(Class<?> clazz, Headers headers);
 
+	@Nullable
 	Class<?> toClass(Headers headers);
 
 }
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/DefaultJackson2JavaTypeMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/DefaultJackson2JavaTypeMapper.java
index 5a9b64df27..ee38e6d229 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/DefaultJackson2JavaTypeMapper.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/DefaultJackson2JavaTypeMapper.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2024 the original author or authors.
+ * Copyright 2017-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.
@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.type.TypeFactory;
 import org.apache.kafka.common.header.Headers;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.util.Assert;
@@ -89,7 +90,7 @@ public void addTrustedPackages(String... packagesToTrust) {
 	}
 
 	@Override
-	public JavaType toJavaType(Headers headers) {
+	public @Nullable JavaType toJavaType(Headers headers) {
 		String typeIdHeader = retrieveHeaderAsString(headers, getClassIdFieldName());
 
 		if (typeIdHeader != null) {
@@ -181,8 +182,9 @@ public void fromClass(Class<?> clazz, Headers headers) {
 	}
 
 	@Override
-	public Class<?> toClass(Headers headers) {
-		return toJavaType(headers).getRawClass();
+	public @Nullable Class<?> toClass(Headers headers) {
+		JavaType javaType = toJavaType(headers);
+		return javaType == null ? null : javaType.getRawClass();
 	}
 
 	@Override
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/Jackson2JavaTypeMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/Jackson2JavaTypeMapper.java
index 91cd1960da..979dc81a00 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/Jackson2JavaTypeMapper.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/Jackson2JavaTypeMapper.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2024 the original author or authors.
+ * Copyright 2017-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.databind.JavaType;
 import org.apache.kafka.common.header.Headers;
+import org.jspecify.annotations.Nullable;
 
 /**
  * Strategy for setting metadata on messages such that one can create the class that needs
@@ -52,6 +53,7 @@ enum TypePrecedence {
 
 	void fromJavaType(JavaType javaType, Headers headers);
 
+	@Nullable
 	JavaType toJavaType(Headers headers);
 
 	TypePrecedence getTypePrecedence();
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/package-info.java b/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/package-info.java
index 70a31bb5ff..261ad63c57 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/package-info.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/mapping/package-info.java
@@ -1,4 +1,5 @@
 /**
  * Provides classes related to type mapping.
  */
-package org.springframework.kafka.support.mapping;
\ No newline at end of file
+@org.jspecify.annotations.NullMarked
+package org.springframework.kafka.support.mapping;
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTopicSerialization.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTopicSerialization.java
index 8660aa35ce..5819aa2653 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTopicSerialization.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTopicSerialization.java
@@ -84,7 +84,7 @@ public abstract class DelegatingByTopicSerialization<T extends Closeable> implem
 
 	private final Set<String> patterns = ConcurrentHashMap.newKeySet();
 
-	private T defaultDelegate;
+	private @Nullable T defaultDelegate;
 
 	private boolean forKeys;
 
@@ -93,7 +93,7 @@ public abstract class DelegatingByTopicSerialization<T extends Closeable> implem
 	public DelegatingByTopicSerialization() {
 	}
 
-	public DelegatingByTopicSerialization(Map<Pattern, T> delegates, T defaultDelegate) {
+	public DelegatingByTopicSerialization(Map<Pattern, T> delegates, @Nullable T defaultDelegate) {
 		Assert.notNull(delegates, "'delegates' cannot be null");
 		Assert.notNull(defaultDelegate, "'defaultDelegate' cannot be null");
 		this.delegates.putAll(delegates);
@@ -260,7 +260,7 @@ else if (key instanceof String regex) {
 		}
 	}
 
-	protected T instantiateAndConfigure(Map<String, ?> configs, boolean isKey, Map<Pattern, T> delegates2,
+	protected @Nullable T instantiateAndConfigure(Map<String, ?> configs, boolean isKey, Map<Pattern, T> delegates2,
 			@Nullable Pattern pattern, Class<?> clazz) {
 
 		if (pattern != null && !this.patterns.add(pattern.pattern())) {
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTopicSerializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTopicSerializer.java
index 161382fa68..d843bcf3cc 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTopicSerializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTopicSerializer.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2022 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.
@@ -71,7 +71,7 @@ public byte[] serialize(String topic, Object data) {
 		throw new UnsupportedOperationException();
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({"unchecked", "NullAway"}) // Dataflow analysis limitation
 	@Override
 	public byte[] serialize(String topic, Headers headers, Object data) {
 		if (data == null) {
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTypeSerializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTypeSerializer.java
index 1a29210ee7..2c0cb21de7 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTypeSerializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTypeSerializer.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2024 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.
@@ -81,6 +81,7 @@ public void configure(Map<String, ?> configs, boolean isKey) {
 		this.delegates.values().forEach(del -> del.configure(configs, isKey));
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	@Override
 	public byte[] serialize(String topic, Object data) {
 		if (data == null) {
@@ -90,6 +91,7 @@ public byte[] serialize(String topic, Object data) {
 		return delegate.serialize(topic, data);
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	@Override
 	public byte[] serialize(String topic, Headers headers, Object data) {
 		if (data == null) {
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingDeserializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingDeserializer.java
index b9bfcf3fd8..e52f52bad9 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingDeserializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingDeserializer.java
@@ -178,7 +178,7 @@ public Object deserialize(String topic, Headers headers, ByteBuffer data) {
 		return deserializer == null ? data : deserializer.deserialize(topic, headers, data);
 	}
 
-	private Deserializer<?> getDeserializerByHeaders(Headers headers) {
+	private @Nullable Deserializer<?> getDeserializerByHeaders(Headers headers) {
 		byte[] value = null;
 		String selectorKey = selectorKey();
 		Header header = headers.lastHeader(selectorKey);
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingSerializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingSerializer.java
index 42c6184ce7..f705f40fa7 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingSerializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingSerializer.java
@@ -187,6 +187,7 @@ public byte[] serialize(String topic, Object data) {
 		throw new UnsupportedOperationException();
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	@Override
 	public byte[] serialize(String topic, Headers headers, Object data) {
 		if (data == null) {
@@ -229,7 +230,7 @@ private String selectorKey() {
 	/*
 	 * Package for testing.
 	 */
-	@Nullable
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	byte[] trySerdes(Object data) {
 		try {
 			Serde<? extends Object> serdeFrom = Serdes.serdeFrom(data.getClass());
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer.java
index eb73ce5e4d..e241b36d93 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer.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.
@@ -21,6 +21,7 @@
 
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.serialization.Deserializer;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -66,13 +67,13 @@ public class ErrorHandlingDeserializer<T> implements Deserializer<T> {
 	 */
 	public static final String VALIDATOR_CLASS = "spring.deserializer.validator.class";
 
-	private Deserializer<T> delegate;
+	private @Nullable Deserializer<T> delegate;
 
 	private boolean isForKey;
 
-	private Function<FailedDeserializationInfo, T> failedDeserializationFunction;
+	private @Nullable Function<FailedDeserializationInfo, T> failedDeserializationFunction;
 
-	private Validator validator;
+	private @Nullable Validator validator;
 
 	public ErrorHandlingDeserializer() {
 	}
@@ -194,9 +195,9 @@ private void setupValidator(Map<String, ?> configs) {
 	}
 
 	@Override
-	public T deserialize(String topic, byte[] data) {
+	public @Nullable T deserialize(String topic, byte[] data) {
 		try {
-			return validate(this.delegate.deserialize(topic, data));
+			return this.delegate == null ? null : validate(this.delegate.deserialize(topic, data));
 		}
 		catch (Exception e) {
 			return recoverFromSupplier(topic, null, data, e);
@@ -204,7 +205,7 @@ public T deserialize(String topic, byte[] data) {
 	}
 
 	@Override
-	public T deserialize(String topic, Headers headers, byte[] data) {
+	public @Nullable T deserialize(String topic, Headers headers, byte[] data) {
 		try {
 			if (this.isForKey) {
 				headers.remove(SerializationUtils.KEY_DESERIALIZER_EXCEPTION_HEADER);
@@ -212,7 +213,7 @@ public T deserialize(String topic, Headers headers, byte[] data) {
 			else {
 				headers.remove(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER);
 			}
-			return validate(this.delegate.deserialize(topic, headers, data));
+			return this.delegate == null ? null : validate(this.delegate.deserialize(topic, headers, data));
 		}
 		catch (Exception e) {
 			SerializationUtils.deserializationException(headers, data, e, this.isForKey);
@@ -228,7 +229,7 @@ private T validate(T deserialized) {
 		return deserialized;
 	}
 
-	private T recoverFromSupplier(String topic, Headers headers, byte[] data, Exception exception) {
+	private @Nullable T recoverFromSupplier(String topic, @Nullable Headers headers, byte[] data, Exception exception) {
 		if (this.failedDeserializationFunction != null) {
 			FailedDeserializationInfo failedDeserializationInfo =
 					new FailedDeserializationInfo(topic, headers, data, this.isForKey, exception);
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/FailedDeserializationInfo.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/FailedDeserializationInfo.java
index fc30d961b5..62b1c4bce6 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/FailedDeserializationInfo.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/FailedDeserializationInfo.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 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.
@@ -19,6 +19,7 @@
 import java.util.Arrays;
 
 import org.apache.kafka.common.header.Headers;
+import org.jspecify.annotations.Nullable;
 
 /**
  * Class containing all the contextual information around a deserialization error.
@@ -32,7 +33,7 @@ public class FailedDeserializationInfo {
 
 	private final String topic;
 
-	private final Headers headers;
+	private final @Nullable Headers headers;
 
 	private final byte[] data;
 
@@ -48,7 +49,7 @@ public class FailedDeserializationInfo {
 	 * @param isForKey true for a key deserializer, false otherwise.
 	 * @param exception exception causing the deserialization error.
 	 */
-	public FailedDeserializationInfo(String topic, Headers headers, byte[] data, boolean isForKey,
+	public FailedDeserializationInfo(String topic, @Nullable Headers headers, byte[] data, boolean isForKey,
 			Exception exception) {
 
 		this.topic = topic;
@@ -62,7 +63,7 @@ public String getTopic() {
 		return this.topic;
 	}
 
-	public Headers getHeaders() {
+	public @Nullable Headers getHeaders() {
 		return this.headers;
 	}
 
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java
index ed36816da5..b2f22a3b86 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java
@@ -129,11 +129,11 @@ public class JsonDeserializer<T> implements Deserializer<T> {
 
 	protected final ObjectMapper objectMapper; // NOSONAR
 
-	protected JavaType targetType; // NOSONAR
+	protected @Nullable JavaType targetType; // NOSONAR
 
 	protected Jackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper(); // NOSONAR
 
-	private ObjectReader reader;
+	private @Nullable ObjectReader reader;
 
 	private boolean typeMapperExplicitlySet = false;
 
@@ -141,7 +141,7 @@ public class JsonDeserializer<T> implements Deserializer<T> {
 
 	private boolean useTypeHeaders = true;
 
-	private JsonTypeResolver typeResolver;
+	private @Nullable JsonTypeResolver typeResolver;
 
 	private boolean setterCalled;
 
@@ -211,7 +211,7 @@ public JsonDeserializer(@Nullable Class<? super T> targetType, boolean useHeader
 	 * type if not.
 	 * @since 2.3
 	 */
-	public JsonDeserializer(TypeReference<? super T> targetType, boolean useHeadersIfPresent) {
+	public JsonDeserializer(@Nullable TypeReference<? super T> targetType, boolean useHeadersIfPresent) {
 		this(targetType, JacksonUtils.enhancedObjectMapper(), useHeadersIfPresent);
 	}
 
@@ -223,7 +223,7 @@ public JsonDeserializer(TypeReference<? super T> targetType, boolean useHeadersI
 	 * type if not.
 	 * @since 2.3
 	 */
-	public JsonDeserializer(JavaType targetType, boolean useHeadersIfPresent) {
+	public JsonDeserializer(@Nullable JavaType targetType, boolean useHeadersIfPresent) {
 		this(targetType, JacksonUtils.enhancedObjectMapper(), useHeadersIfPresent);
 	}
 
@@ -250,7 +250,7 @@ public JsonDeserializer(TypeReference<? super T> targetType, ObjectMapper object
 	 * @param targetType the target java type to use if no type info headers are present.
 	 * @param objectMapper the mapper. type if not.
 	 */
-	public JsonDeserializer(JavaType targetType, ObjectMapper objectMapper) {
+	public JsonDeserializer(@Nullable JavaType targetType, ObjectMapper objectMapper) {
 		this(targetType, objectMapper, true);
 	}
 
@@ -291,7 +291,7 @@ public JsonDeserializer(@Nullable Class<? super T> targetType, ObjectMapper obje
 	 * type if not.
 	 * @since 2.3
 	 */
-	public JsonDeserializer(TypeReference<? super T> targetType, ObjectMapper objectMapper,
+	public JsonDeserializer(@Nullable TypeReference<? super T> targetType, ObjectMapper objectMapper,
 			boolean useHeadersIfPresent) {
 
 		this(targetType != null ? TypeFactory.defaultInstance().constructType(targetType) : null,
@@ -442,6 +442,7 @@ private boolean configsHasOurKeys(Map<String, ?> configs) {
 		return false;
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	private Map<String, Class<?>> createMappings(Map<String, ?> configs) {
 		Map<String, Class<?>> mappings =
 				JsonSerializer.createMappings(configs.get(JsonSerializer.TYPE_MAPPINGS).toString());
@@ -562,7 +563,7 @@ private void addTargetPackageToTrusted() {
 		}
 	}
 
-	private String getTargetPackageName() {
+	private @Nullable String getTargetPackageName() {
 		if (this.targetType != null) {
 			return ClassUtils.getPackageName(this.targetType.getRawClass()).replaceFirst("\\[L", "");
 		}
@@ -574,7 +575,7 @@ private void doAddTrustedPackages(String... packages) {
 	}
 
 	@Override
-	public T deserialize(String topic, Headers headers, byte[] data) {
+	public @Nullable T deserialize(String topic, Headers headers, byte[] data) {
 		if (data == null) {
 			return null;
 		}
@@ -605,7 +606,7 @@ public T deserialize(String topic, Headers headers, byte[] data) {
 	}
 
 	@Override
-	public T deserialize(String topic, @Nullable byte[] data) {
+	public @Nullable T deserialize(String topic, byte[] data) {
 		if (data == null) {
 			return null;
 		}
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java
index edfd1b658e..997f2bcf6f 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java
@@ -101,7 +101,7 @@ public JsonSerializer(TypeReference<? super T> targetType, ObjectMapper objectMa
 		this(targetType == null ? null : objectMapper.constructType(targetType.getType()), objectMapper);
 	}
 
-	public JsonSerializer(JavaType targetType, ObjectMapper objectMapper) {
+	public JsonSerializer(@Nullable JavaType targetType, ObjectMapper objectMapper) {
 		Assert.notNull(objectMapper, "'objectMapper' must not be null.");
 		this.objectMapper = objectMapper;
 		this.writer = objectMapper.writerFor(targetType);
@@ -201,8 +201,8 @@ protected static Map<String, Class<?>> createMappings(String mappings) {
 		return mappingsMap;
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	@Override
-	@Nullable
 	public byte[] serialize(String topic, Headers headers, @Nullable T data) {
 		if (data == null) {
 			return null;
@@ -213,8 +213,8 @@ public byte[] serialize(String topic, Headers headers, @Nullable T data) {
 		return serialize(topic, data);
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	@Override
-	@Nullable
 	public byte[] serialize(String topic, @Nullable T data) {
 		if (data == null) {
 			return null;
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonTypeResolver.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonTypeResolver.java
index dad78f8f43..47cfa85ecd 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonTypeResolver.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonTypeResolver.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 com.fasterxml.jackson.databind.JavaType;
 import org.apache.kafka.common.header.Headers;
+import org.jspecify.annotations.Nullable;
 
 /**
  * Determine the {@link JavaType} from the topic/data/headers.
@@ -36,6 +37,6 @@ public interface JsonTypeResolver {
 	 * @param headers the headers.
 	 * @return the type.
 	 */
-	JavaType resolveType(String topic, byte[] data, Headers headers);
+	JavaType resolveType(String topic, byte[] data, @Nullable Headers headers);
 
 }
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ParseStringDeserializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ParseStringDeserializer.java
index 98d3b789b6..cf8a167590 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ParseStringDeserializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ParseStringDeserializer.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.
@@ -26,6 +26,7 @@
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.utils.Utils;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.util.Assert;
 
@@ -105,7 +106,7 @@ public T deserialize(String topic, byte[] data) {
 	}
 
 	@Override
-	public T deserialize(String topic, Headers headers, byte[] data) {
+	public T deserialize(String topic, @Nullable Headers headers, byte[] data) {
 		return this.parser.apply(data == null ? null : new String(data, this.charset), headers);
 	}
 
@@ -115,7 +116,7 @@ public T deserialize(String topic, Headers headers, ByteBuffer data) {
 		return this.parser.apply(value, headers);
 	}
 
-	private String deserialize(ByteBuffer data) {
+	private @Nullable String deserialize(@Nullable ByteBuffer data) {
 		if (data == null) {
 			return null;
 		}
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/StringOrBytesSerializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/StringOrBytesSerializer.java
index e1d6d5059b..8dc3f56062 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/StringOrBytesSerializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/StringOrBytesSerializer.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 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.
@@ -39,6 +39,7 @@ public void configure(Map<String, ?> configs, boolean isKey) {
 		this.stringSerializer.configure(configs, isKey);
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	@Override
 	public byte[] serialize(String topic, Object data) {
 		if (data instanceof byte[]) {
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ToStringSerializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ToStringSerializer.java
index 6b8d8df05e..f33ef51d21 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ToStringSerializer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ToStringSerializer.java
@@ -86,7 +86,7 @@ public byte[] serialize(String topic, @Nullable T data) {
 	}
 
 	@Override
-	@Nullable
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	public byte[] serialize(String topic, @Nullable Headers headers, @Nullable T data) {
 		if (data == null) {
 			return null;
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/package-info.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/package-info.java
new file mode 100644
index 0000000000..42618506db
--- /dev/null
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/package-info.java
@@ -0,0 +1,5 @@
+/**
+ * Provides classes related to type mapping.
+ */
+@org.jspecify.annotations.NullMarked
+package org.springframework.kafka.support.serializer;