Add JSpecify nullability annotations to kafka.event and security.jaas packages

Add nullability annotations and make relevant fields final in spring-kafka event classes
and security.jaas classes for improved type safety.

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

Author: sobychacko

spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerPausedEvent.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.
@@ -19,6 +19,7 @@
 import java.util.Collection;
 
 import org.apache.kafka.common.TopicPartition;
+import org.jspecify.annotations.Nullable;
 
 /**
  * An event published when a consumer is paused.
@@ -31,7 +32,7 @@ public class ConsumerPausedEvent extends KafkaEvent {
 
 	private static final long serialVersionUID = 1L;
 
-	private final String reason;
+	private final @Nullable String reason;
 
 	private transient Collection<TopicPartition> partitions;
 

spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerStoppingEvent.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.
@@ -35,9 +35,9 @@ public class ConsumerStoppingEvent extends KafkaEvent {
 
 	private static final long serialVersionUID = 1L;
 
-	private transient Consumer<?, ?> consumer;
+	private transient final Consumer<?, ?> consumer;
 
-	private transient Collection<TopicPartition> partitions;
+	private transient final Collection<TopicPartition> partitions;
 
 	/**
 	 * Construct an instance with the provided source, consumer and partitions.

spring-kafka/src/main/java/org/springframework/kafka/event/KafkaEvent.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2024 the original author or authors.
+ * Copyright 2015-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,7 +29,7 @@ public abstract class KafkaEvent extends ApplicationEvent {
 
 	private static final long serialVersionUID = 1L;
 
-	private transient Object container;
+	private transient final Object container;
 
 	public KafkaEvent(Object source, Object container) {
 		super(source);

spring-kafka/src/main/java/org/springframework/kafka/event/ListenerContainerIdleEvent.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2023 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.
@@ -23,6 +23,7 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
+import org.jspecify.annotations.Nullable;
 
 /**
  * An event that is emitted when a container is idle if the container
@@ -41,9 +42,9 @@ public class ListenerContainerIdleEvent extends KafkaEvent {
 
 	private final boolean paused;
 
-	private transient List<TopicPartition> topicPartitions;
+	private transient final @Nullable List<TopicPartition> topicPartitions;
 
-	private transient Consumer<?, ?> consumer;
+	private transient final Consumer<?, ?> consumer;
 
 	/**
 	 * Construct an instance with the provided arguments.
@@ -58,7 +59,7 @@ public class ListenerContainerIdleEvent extends KafkaEvent {
 	 */
 	public ListenerContainerIdleEvent(Object source, Object container,
 			long idleTime, String id,
-			Collection<TopicPartition> topicPartitions, Consumer<?, ?> consumer, boolean paused) {
+			@Nullable Collection<TopicPartition> topicPartitions, Consumer<?, ?> consumer, boolean paused) {
 
 		super(source, container);
 		this.idleTime = idleTime;
@@ -80,7 +81,7 @@ public long getIdleTime() {
 	 * The TopicPartitions the container is listening to.
 	 * @return the TopicPartition list.
 	 */
-	public Collection<TopicPartition> getTopicPartitions() {
+	public @Nullable Collection<TopicPartition> getTopicPartitions() {
 		return this.topicPartitions == null ? null : Collections.unmodifiableList(this.topicPartitions);
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/event/ListenerContainerNoLongerIdleEvent.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.
@@ -23,6 +23,7 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
+import org.jspecify.annotations.Nullable;
 
 /**
  * An event that is emitted when a container is no longer idle if configured to publish
@@ -39,9 +40,9 @@ public class ListenerContainerNoLongerIdleEvent extends KafkaEvent {
 
 	private final String listenerId;
 
-	private transient List<TopicPartition> topicPartitions;
+	private transient final @Nullable List<TopicPartition> topicPartitions;
 
-	private transient Consumer<?, ?> consumer;
+	private transient final @Nullable Consumer<?, ?> consumer;
 
 	/**
 	 * Construct an instance with the provided arguments.
@@ -53,7 +54,7 @@ public class ListenerContainerNoLongerIdleEvent extends KafkaEvent {
 	 * @param consumer the consumer.
 	 */
 	public ListenerContainerNoLongerIdleEvent(Object source, Object container, long idleTime, String id,
-			Collection<TopicPartition> topicPartitions, Consumer<?, ?> consumer) {
+			@Nullable Collection<TopicPartition> topicPartitions, Consumer<?, ?> consumer) {
 
 		super(source, container);
 		this.idleTime = idleTime;
@@ -66,7 +67,7 @@ public ListenerContainerNoLongerIdleEvent(Object source, Object container, long
 	 * The TopicPartitions the container is listening to.
 	 * @return the TopicPartition list.
 	 */
-	public Collection<TopicPartition> getTopicPartitions() {
+	public @Nullable Collection<TopicPartition> getTopicPartitions() {
 		return this.topicPartitions == null ? null : Collections.unmodifiableList(this.topicPartitions);
 	}
 
@@ -91,7 +92,7 @@ public String getListenerId() {
 	 * Allows the listener to resume a paused consumer.
 	 * @return the consumer.
 	 */
-	public Consumer<?, ?> getConsumer() {
+	public @Nullable Consumer<?, ?> getConsumer() {
 		return this.consumer;
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/event/ListenerContainerPartitionIdleEvent.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.
@@ -39,7 +39,7 @@ public class ListenerContainerPartitionIdleEvent extends KafkaEvent {
 
 	private final boolean paused;
 
-	private transient Consumer<?, ?> consumer;
+	private transient final Consumer<?, ?> consumer;
 
 	/**
 	 * Construct an instance with the provided arguments.

spring-kafka/src/main/java/org/springframework/kafka/event/ListenerContainerPartitionNoLongerIdleEvent.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.
@@ -37,7 +37,7 @@ public class ListenerContainerPartitionNoLongerIdleEvent extends KafkaEvent {
 
 	private final TopicPartition topicPartition;
 
-	private transient Consumer<?, ?> consumer;
+	private transient final Consumer<?, ?> consumer;
 
 	/**
 	 * Construct an instance with the provided arguments.

spring-kafka/src/main/java/org/springframework/kafka/event/NonResponsiveConsumerEvent.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2022 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 org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
+import org.jspecify.annotations.Nullable;
 
 /**
  * An event that is emitted when a consumer is not responding to the poll; with early
@@ -40,9 +41,9 @@ public class NonResponsiveConsumerEvent extends KafkaEvent {
 
 	private final String listenerId;
 
-	private final List<TopicPartition> topicPartitions;
+	private final @Nullable List<TopicPartition> topicPartitions;
 
-	private transient Consumer<?, ?> consumer;
+	private transient final Consumer<?, ?> consumer;
 
 	/**
 	 * Construct an instance with the provided properties.
@@ -56,7 +57,7 @@ public class NonResponsiveConsumerEvent extends KafkaEvent {
 	 */
 	public NonResponsiveConsumerEvent(Object source, Object container,
 			long timeSinceLastPoll, String id,
-			Collection<TopicPartition> topicPartitions, Consumer<?, ?> consumer) {
+			@Nullable Collection<TopicPartition> topicPartitions, Consumer<?, ?> consumer) {
 
 		super(source, container);
 		this.timeSinceLastPoll = timeSinceLastPoll;
@@ -77,7 +78,7 @@ public long getTimeSinceLastPoll() {
 	 * The TopicPartitions the container is listening to.
 	 * @return the TopicPartition list.
 	 */
-	public Collection<TopicPartition> getTopicPartitions() {
+	public @Nullable Collection<TopicPartition> getTopicPartitions() {
 		return this.topicPartitions == null ? null : Collections.unmodifiableList(this.topicPartitions);
 	}
 

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

@@ -1,4 +1,5 @@
 /**
  * Application Events.
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.kafka.event;

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

@@ -1,4 +1,5 @@
 /**
  * Base package for kafka
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.kafka;

spring-kafka/src/main/java/org/springframework/kafka/security/jaas/KafkaJaasLoginModuleInitializer.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.
@@ -24,6 +24,7 @@
 import javax.security.auth.login.Configuration;
 
 import org.apache.kafka.common.security.JaasUtils;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.SmartInitializingSingleton;
@@ -124,7 +125,7 @@ public void setOptions(Map<String, String> options) {
 	public void afterSingletonsInstantiated() {
 		// only use programmatic support if a file is not set via system property
 		if (this.ignoreJavaLoginConfigParamSystemProperty) {
-			Map<String, AppConfigurationEntry[]> configurationEntries = new HashMap<>();
+			Map<String, @Nullable AppConfigurationEntry[]> configurationEntries = new HashMap<>();
 			AppConfigurationEntry kafkaClientConfigurationEntry = new AppConfigurationEntry(
 					this.loginModule,
 					this.controlFlag,
@@ -145,20 +146,21 @@ public void destroy() {
 
 	private static class InternalConfiguration extends Configuration {
 
-		private final Map<String, AppConfigurationEntry[]> configurationEntries;
+		private final Map<String, @Nullable AppConfigurationEntry[]> configurationEntries;
 
 		private final Configuration delegate;
 
-		InternalConfiguration(Map<String, AppConfigurationEntry[]> configurationEntries, Configuration delegate) {
+		InternalConfiguration(Map<String, @Nullable AppConfigurationEntry[]> configurationEntries, Configuration delegate) {
 			Assert.notNull(configurationEntries, "'configurationEntries' cannot be null");
 			Assert.notEmpty(configurationEntries, "'configurationEntries' cannot be empty");
 			this.configurationEntries = configurationEntries;
 			this.delegate = delegate;
 		}
 
 		@Override
-		public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
-			AppConfigurationEntry[] conf = this.delegate == null ? null : this.delegate.getAppConfigurationEntry(name);
+		@SuppressWarnings("NullAway") // Overridden method does not define nullness
+		public @Nullable AppConfigurationEntry[] getAppConfigurationEntry(String name) {
+			AppConfigurationEntry[] conf = this.delegate.getAppConfigurationEntry(name);
 			return conf != null ? conf : this.configurationEntries.get(name);
 		}
 

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

@@ -1,4 +1,5 @@
 /**
  * Provides classes related to jaas.
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.kafka.security.jaas;