Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

jspecify nullability changes for the listener package. (#3775)

Browse Source 
https://github.com/spring-projects/spring-kafka/issues/3762

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
This commit is contained in:
Soby Chacko
2025-03-01 14:35:23 -05:00
committed by GitHub
parent 2af76c0bf9
commit cfd7c98e15
55 changed files with 367 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-kafka-docs/src/main/kotlin/org/springframework/kafka/kdocs/dynamic/Application.kt
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022-2024 the original author or authors.
* Copyright 2022-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.
@@ -60,8 +60,8 @@ private fun createContainer(
factory: ConcurrentKafkaListenerContainerFactory<String, String>, topic: String, group: String
): ConcurrentMessageListenerContainer<String, String> {
val container = factory.createContainer(topic)
container.containerProperties.messageListener = MyListener()
container.containerProperties.groupId = group
container.containerProperties.setMessageListener(MyListener())
container.containerProperties.setGroupId(group)
container.beanName = group
container.start()
return container
@@ -104,9 +104,10 @@ fun pojo(id: String, topic: String): MyPojo {
// tag::listener[]
class MyListener : MessageListener<String?, String?> {
class MyListener : MessageListener<String, String> {
override fun onMessage(data: ConsumerRecord<String, String>) {
override fun onMessage(data: ConsumerRecord<String?, String?>) {
// ...
}

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

@@ -74,7 +74,7 @@ class Application {
factory: ConcurrentKafkaListenerContainerFactory<String, String>
): ReplyingKafkaTemplate<String, String, String> {
val replyContainer = factory.createContainer("replies")
replyContainer.containerProperties.groupId = "request.replies"
replyContainer.containerProperties.setGroupId("request.replies")
val template = ReplyingKafkaTemplate<String, String, String>(pf, replyContainer)
template.messageConverter = ByteArrayJsonMessageConverter()
template.setDefaultTopic("requests")

1
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java
View File

@@ -321,6 +321,7 @@ public interface KafkaOperations<K, V> {
*/
interface OperationsCallback<K, V, T> {
@Nullable
T doInOperations(KafkaOperations<K, V> operations);
}

2
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java
View File

@@ -660,7 +660,7 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
}
@Override
public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
public <T> @Nullable T executeInTransaction(OperationsCallback<K, V, T> callback) {
Assert.notNull(callback, "'callback' cannot be null");
Assert.state(this.transactional, "Producer factory does not support transactions");
Thread currentThread = Thread.currentThread();

7
spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerStoppingEvent.java
View File

@@ -20,6 +20,7 @@ import java.util.Collection;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.jspecify.annotations.Nullable;
/**
* An event published when a consumer is stopped. While it is best practice to use
@@ -37,7 +38,7 @@ public class ConsumerStoppingEvent extends KafkaEvent {
private transient final Consumer<?, ?> consumer;
private transient final Collection<TopicPartition> partitions;
private transient final @Nullable Collection<TopicPartition> partitions;
/**
* Construct an instance with the provided source, consumer and partitions.
@@ -48,7 +49,7 @@ public class ConsumerStoppingEvent extends KafkaEvent {
* @since 2.2.1
*/
public ConsumerStoppingEvent(Object source, Object container,
Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
Consumer<?, ?> consumer, @Nullable Collection<TopicPartition> partitions) {
super(source, container);
this.consumer = consumer;
this.partitions = partitions;
@@ -58,7 +59,7 @@ public class ConsumerStoppingEvent extends KafkaEvent {
return this.consumer;
}
public Collection<TopicPartition> getPartitions() {
public @Nullable Collection<TopicPartition> getPartitions() {
return this.partitions;
}

30
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java
View File

@@ -65,21 +65,23 @@ public abstract class AbstractConsumerSeekAware implements ConsumerSeekAware {
}
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
partitions.forEach(tp -> {
List<ConsumerSeekCallback> removedCallbacks = this.topicToCallbacks.remove(tp);
if (removedCallbacks != null && !removedCallbacks.isEmpty()) {
removedCallbacks.forEach(cb -> {
List<TopicPartition> topics = this.callbackToTopics.get(cb);
if (topics != null) {
topics.remove(tp);
if (topics.isEmpty()) {
this.callbackToTopics.remove(cb);
public void onPartitionsRevoked(@Nullable Collection<TopicPartition> partitions) {
if (partitions != null) {
partitions.forEach(tp -> {
List<ConsumerSeekCallback> removedCallbacks = this.topicToCallbacks.remove(tp);
if (removedCallbacks != null && !removedCallbacks.isEmpty()) {
removedCallbacks.forEach(cb -> {
List<TopicPartition> topics = this.callbackToTopics.get(cb);
if (topics != null) {
topics.remove(tp);
if (topics.isEmpty()) {
this.callbackToTopics.remove(cb);
}
}
}
});
}
});
});
}
});
}
}
@Override

12
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractKafkaBackOffManagerFactory.java
View File

@@ -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.
@@ -16,6 +16,10 @@
package org.springframework.kafka.listener;
import java.util.Objects;
import org.jspecify.annotations.Nullable;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.kafka.config.KafkaListenerConfigUtils;
@@ -32,9 +36,9 @@ import org.springframework.util.Assert;
public abstract class AbstractKafkaBackOffManagerFactory
implements KafkaBackOffManagerFactory, ApplicationContextAware {
private ApplicationContext applicationContext;
private @Nullable ApplicationContext applicationContext;
private ListenerContainerRegistry listenerContainerRegistry;
private @Nullable ListenerContainerRegistry listenerContainerRegistry;
/**
* Creates an instance that will retrieve the {@link ListenerContainerRegistry} from
@@ -83,7 +87,7 @@ public abstract class AbstractKafkaBackOffManagerFactory
}
protected <T> T getBean(String beanName, Class<T> beanClass) {
return this.applicationContext.getBean(beanName, beanClass);
return Objects.requireNonNull(this.applicationContext).getBean(beanName, beanClass);
}
@Override

30
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java
View File

@@ -101,9 +101,9 @@ public abstract class AbstractMessageListenerContainer<K, V>
@NonNull
private String beanName = "noBeanNameSet";
private ApplicationEventPublisher applicationEventPublisher;
private @Nullable ApplicationEventPublisher applicationEventPublisher;
private CommonErrorHandler commonErrorHandler;
private @Nullable CommonErrorHandler commonErrorHandler;
private boolean autoStartup = true;
@@ -114,15 +114,16 @@ public abstract class AbstractMessageListenerContainer<K, V>
private int topicCheckTimeout = DEFAULT_TOPIC_CHECK_TIMEOUT;
private RecordInterceptor<K, V> recordInterceptor;
private @Nullable RecordInterceptor<K, V> recordInterceptor;
private BatchInterceptor<K, V> batchInterceptor;
private @Nullable BatchInterceptor<K, V> batchInterceptor;
private boolean interceptBeforeTx = true;
@SuppressWarnings("NullAway.Init")
private byte[] listenerInfo;
private ApplicationContext applicationContext;
private @Nullable ApplicationContext applicationContext;
private volatile boolean running = false;
@@ -149,13 +150,13 @@ public abstract class AbstractMessageListenerContainer<K, V>
* @param containerProperties the properties.
*/
@SuppressWarnings("unchecked")
protected AbstractMessageListenerContainer(ConsumerFactory<? super K, ? super V> consumerFactory,
protected AbstractMessageListenerContainer(@Nullable ConsumerFactory<? super K, ? super V> consumerFactory,
ContainerProperties containerProperties) {
Assert.notNull(containerProperties, "'containerProperties' cannot be null");
Assert.notNull(consumerFactory, "'consumerFactory' cannot be null");
this.consumerFactory = (ConsumerFactory<K, V>) consumerFactory;
String[] topics = containerProperties.getTopics();
@Nullable String @Nullable [] topics = containerProperties.getTopics();
if (topics != null) {
this.containerProperties = new ContainerProperties(topics);
}
@@ -165,7 +166,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
this.containerProperties = new ContainerProperties(topicPattern);
}
else {
TopicPartitionOffset[] topicPartitions = containerProperties.getTopicPartitions();
@Nullable TopicPartitionOffset @Nullable [] topicPartitions = containerProperties.getTopicPartitions();
if (topicPartitions != null) {
this.containerProperties = new ContainerProperties(topicPartitions);
}
@@ -370,8 +371,8 @@ public abstract class AbstractMessageListenerContainer<K, V>
return this.mainListenerId;
}
@Nullable
@Override
@SuppressWarnings("NullAway") // Dataflow analysis limitation
public byte[] getListenerInfo() {
return this.listenerInfo != null ? Arrays.copyOf(this.listenerInfo, this.listenerInfo.length) : null;
}
@@ -382,6 +383,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
* @param listenerInfo the info.
* @since 2.8.4
*/
@SuppressWarnings("NullAway") // Dataflow analysis limitation
public void setListenerInfo(@Nullable byte[] listenerInfo) {
this.listenerInfo = listenerInfo != null ? Arrays.copyOf(listenerInfo, listenerInfo.length) : null;
}
@@ -458,7 +460,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
this.kafkaAdmin = kafkaAdmin;
}
protected RecordInterceptor<K, V> getRecordInterceptor() {
protected @Nullable RecordInterceptor<K, V> getRecordInterceptor() {
return this.recordInterceptor;
}
@@ -469,11 +471,11 @@ public abstract class AbstractMessageListenerContainer<K, V>
* @since 2.2.7
* @see #setInterceptBeforeTx(boolean)
*/
public void setRecordInterceptor(RecordInterceptor<K, V> recordInterceptor) {
public void setRecordInterceptor(@Nullable RecordInterceptor<K, V> recordInterceptor) {
this.recordInterceptor = recordInterceptor;
}
protected BatchInterceptor<K, V> getBatchInterceptor() {
protected @Nullable BatchInterceptor<K, V> getBatchInterceptor() {
return this.batchInterceptor;
}
@@ -483,7 +485,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
* @since 2.6.6
* @see #setInterceptBeforeTx(boolean)
*/
public void setBatchInterceptor(BatchInterceptor<K, V> batchInterceptor) {
public void setBatchInterceptor(@Nullable BatchInterceptor<K, V> batchInterceptor) {
this.batchInterceptor = batchInterceptor;
}
@@ -541,7 +543,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
List<String> missing = null;
try (AdminClient client = AdminClient.create(configs)) { // NOSONAR - false positive null check
if (client != null) {
String[] topics = this.containerProperties.getTopics();
@Nullable String @Nullable[] topics = this.containerProperties.getTopics();
if (topics == null) {
topics = Arrays.stream(this.containerProperties.getTopicPartitions())
.map(TopicPartitionOffset::getTopic)

2
spring-kafka/src/main/java/org/springframework/kafka/listener/AcknowledgingConsumerAwareMessageListener.java
View File

@@ -47,6 +47,6 @@ public interface AcknowledgingConsumerAwareMessageListener<K, V> extends Message
}
@Override
void onMessage(ConsumerRecord<K, V> data, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer);
void onMessage(ConsumerRecord<K, V> data, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer);
}

2
spring-kafka/src/main/java/org/springframework/kafka/listener/BackOffHandler.java
View File

@@ -35,7 +35,7 @@ public interface BackOffHandler {
* @param exception the exception.
* @param nextBackOff the next back off.
*/
default void onNextBackOff(@Nullable MessageListenerContainer container, Exception exception, long nextBackOff) {
default void onNextBackOff(@Nullable MessageListenerContainer container, @Nullable Exception exception, long nextBackOff) {
throw new UnsupportedOperationException();
}

2
spring-kafka/src/main/java/org/springframework/kafka/listener/BatchAcknowledgingConsumerAwareMessageListener.java
View File

@@ -51,6 +51,6 @@ public interface BatchAcknowledgingConsumerAwareMessageListener<K, V> extends Ba
}
@Override
void onMessage(List<ConsumerRecord<K, V>> data, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer);
void onMessage(List<ConsumerRecord<K, V>> data, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer);
}

5
spring-kafka/src/main/java/org/springframework/kafka/listener/BatchAcknowledgingMessageListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2019 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.
@@ -19,6 +19,7 @@ package org.springframework.kafka.listener;
import java.util.List;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jspecify.annotations.Nullable;
import org.springframework.kafka.support.Acknowledgment;
@@ -49,6 +50,6 @@ public interface BatchAcknowledgingMessageListener<K, V> extends BatchMessageLis
}
@Override
void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment);
void onMessage(List<ConsumerRecord<K, V>> data, @Nullable Acknowledgment acknowledgment);
}

5
spring-kafka/src/main/java/org/springframework/kafka/listener/BatchConsumerAwareMessageListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 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.
@@ -20,6 +20,7 @@ import java.util.List;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jspecify.annotations.Nullable;
/**
* Listener for handling a batch of incoming Kafka messages; the list
@@ -47,6 +48,6 @@ public interface BatchConsumerAwareMessageListener<K, V> extends BatchMessageLis
}
@Override
void onMessage(List<ConsumerRecord<K, V>> data, Consumer<?, ?> consumer);
void onMessage(List<ConsumerRecord<K, V>> data, @Nullable Consumer<?, ?> consumer);
}

2
spring-kafka/src/main/java/org/springframework/kafka/listener/BatchListenerFailedException.java
View File

@@ -36,7 +36,7 @@ public class BatchListenerFailedException extends KafkaException {
private final int index;
private transient ConsumerRecord<?, ?> record;
private transient @Nullable ConsumerRecord<?, ?> record;
/**
* Construct an instance with the provided properties.

5
spring-kafka/src/main/java/org/springframework/kafka/listener/CompositeBatchInterceptor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2021 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.
@@ -22,6 +22,7 @@ import java.util.Collection;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.jspecify.annotations.Nullable;
import org.springframework.util.Assert;
@@ -53,7 +54,7 @@ public class CompositeBatchInterceptor<K, V> implements BatchInterceptor<K, V> {
}
@Override
public ConsumerRecords<K, V> intercept(ConsumerRecords<K, V> records, Consumer<K, V> consumer) {
public @Nullable ConsumerRecords<K, V> intercept(ConsumerRecords<K, V> records, Consumer<K, V> consumer) {
ConsumerRecords<K, V> recordsToIntercept = records;
for (BatchInterceptor<K, V> delegate : this.delegates) {
recordsToIntercept = delegate.intercept(recordsToIntercept, consumer);

16
spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
View File

@@ -73,7 +73,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
private boolean alwaysClientIdSuffix = true;
private volatile Reason reason;
private volatile @Nullable Reason reason;
/**
* Construct an instance with the supplied configuration properties.
@@ -82,7 +82,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
* @param consumerFactory the consumer factory.
* @param containerProperties the container properties.
*/
public ConcurrentMessageListenerContainer(ConsumerFactory<? super K, ? super V> consumerFactory,
public ConcurrentMessageListenerContainer(@Nullable ConsumerFactory<? super K, ? super V> consumerFactory,
ContainerProperties containerProperties) {
super(consumerFactory, containerProperties);
@@ -244,7 +244,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
if (!isRunning()) {
checkTopics();
ContainerProperties containerProperties = getContainerProperties();
TopicPartitionOffset[] topicPartitions = containerProperties.getTopicPartitions();
@Nullable TopicPartitionOffset @Nullable [] topicPartitions = containerProperties.getTopicPartitions();
if (topicPartitions != null && this.concurrency > topicPartitions.length) {
this.logger.warn(() -> "When specific partitions are provided, the concurrency must be less than or "
+ "equal to the number of partitions; reduced from " + this.concurrency + " to "
@@ -302,7 +302,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
}
private KafkaMessageListenerContainer<K, V> constructContainer(ContainerProperties containerProperties,
@Nullable TopicPartitionOffset[] topicPartitions, int i) {
@Nullable TopicPartitionOffset @Nullable [] topicPartitions, int i) {
KafkaMessageListenerContainer<K, V> container;
if (topicPartitions == null) {
@@ -315,9 +315,8 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
return container;
}
@Nullable
private TopicPartitionOffset[] partitionSubset(ContainerProperties containerProperties, int index) {
TopicPartitionOffset[] topicPartitions = containerProperties.getTopicPartitions();
private @Nullable TopicPartitionOffset @Nullable [] partitionSubset(ContainerProperties containerProperties, int index) {
@Nullable TopicPartitionOffset @Nullable [] topicPartitions = containerProperties.getTopicPartitions();
if (topicPartitions == null) {
return null;
}
@@ -434,7 +433,8 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
}
}
private void publishConcurrentContainerStoppedEvent(Reason reason) {
@SuppressWarnings("NullAway") // Dataflow analysis limitation
private void publishConcurrentContainerStoppedEvent(@Nullable Reason reason) {
ApplicationEventPublisher eventPublisher = getApplicationEventPublisher();
if (eventPublisher != null) {
eventPublisher.publishEvent(new ConcurrentContainerStoppedEvent(this, reason));

5
spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareListenerErrorHandler.java
View File

@@ -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.
@@ -17,6 +17,7 @@
package org.springframework.kafka.listener;
import org.apache.kafka.clients.consumer.Consumer;
import org.jspecify.annotations.Nullable;
import org.springframework.messaging.Message;
@@ -38,6 +39,6 @@ public interface ConsumerAwareListenerErrorHandler extends KafkaListenerErrorHan
}
@Override
Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer);
Object handleError(Message<?> message, ListenerExecutionFailedException exception, @Nullable Consumer<?, ?> consumer);
}

5
spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareMessageListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 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 @@ package org.springframework.kafka.listener;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jspecify.annotations.Nullable;
/**
* Listener for handling individual incoming Kafka messages.
@@ -43,6 +44,6 @@ public interface ConsumerAwareMessageListener<K, V> extends MessageListener<K, V
}
@Override
void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer);
void onMessage(ConsumerRecord<K, V> data, @Nullable Consumer<?, ?> consumer);
}

2
spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareRecordRecoverer.java
View File

@@ -43,6 +43,6 @@ public interface ConsumerAwareRecordRecoverer extends ConsumerRecordRecoverer {
* @param exception the exception.
* @since 2.7
*/
void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consumer, Exception exception);
void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consumer, @Nullable Exception exception);
}

24
spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerProperties.java
View File

@@ -51,17 +51,17 @@ public class ConsumerProperties {
/**
* Topic names.
*/
private final String[] topics;
private final @Nullable String @Nullable [] topics;
/**
* Topic pattern.
*/
private final Pattern topicPattern;
private final @Nullable Pattern topicPattern;
/**
* Topics/partitions/initial offsets.
*/
private final TopicPartitionOffset[] topicPartitions;
private final @Nullable TopicPartitionOffset @Nullable [] topicPartitions;
/**
* The max time to block in the consumer waiting for records.
@@ -71,7 +71,7 @@ public class ConsumerProperties {
/**
* Override the group id.
*/
private String groupId;
private @Nullable String groupId;
/**
* Override the client id.
@@ -81,21 +81,21 @@ public class ConsumerProperties {
/**
* A user defined {@link ConsumerRebalanceListener} implementation.
*/
private ConsumerRebalanceListener consumerRebalanceListener;
private @Nullable ConsumerRebalanceListener consumerRebalanceListener;
private Duration syncCommitTimeout;
private @Nullable Duration syncCommitTimeout;
/**
* The commit callback; by default a simple logging callback is used to log
* success at DEBUG level and failures at ERROR level.
*/
private OffsetCommitCallback commitCallback;
private @Nullable OffsetCommitCallback commitCallback;
/**
* A provider for {@link OffsetAndMetadata}; by default, the provider creates an offset and metadata with
* empty metadata. The provider gives a way to customize the metadata.
*/
private OffsetAndMetadataProvider offsetAndMetadataProvider;
private @Nullable OffsetAndMetadataProvider offsetAndMetadataProvider;
/**
* Whether or not to call consumer.commitSync() or commitAsync() when the
@@ -107,7 +107,7 @@ public class ConsumerProperties {
private Properties kafkaConsumerProperties = new Properties();
private Duration authExceptionRetryInterval;
private @Nullable Duration authExceptionRetryInterval;
private int commitRetries = DEFAULT_COMMIT_RETRIES;
@@ -137,7 +137,7 @@ public class ConsumerProperties {
* @param topicPattern the pattern.
* @see org.apache.kafka.clients.CommonClientConfigs#METADATA_MAX_AGE_CONFIG
*/
public ConsumerProperties(Pattern topicPattern) {
public ConsumerProperties(@Nullable Pattern topicPattern) {
this.topics = null;
this.topicPattern = topicPattern;
this.topicPartitions = null;
@@ -160,7 +160,7 @@ public class ConsumerProperties {
* @return the topics.
*/
@Nullable
public String[] getTopics() {
public String @Nullable [] getTopics() {
return this.topics != null
? Arrays.copyOf(this.topics, this.topics.length)
: null;
@@ -181,7 +181,7 @@ public class ConsumerProperties {
* @since 2.5
*/
@Nullable
public TopicPartitionOffset[] getTopicPartitions() {
public TopicPartitionOffset @Nullable [] getTopicPartitions() {
return this.topicPartitions != null
? Arrays.copyOf(this.topicPartitions, this.topicPartitions.length)
: null;

2
spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java
View File

@@ -60,7 +60,7 @@ public interface ConsumerSeekAware {
* @param partitions the partitions that have been revoked.
* @since 2.3
*/
default void onPartitionsRevoked(Collection<TopicPartition> partitions) {
default void onPartitionsRevoked(@Nullable Collection<TopicPartition> partitions) {
}
/**

27
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerGroupSequencer.java
View File

@@ -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.
@@ -19,8 +19,10 @@ package org.springframework.kafka.listener;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.Objects;
import org.apache.commons.logging.LogFactory;
import org.jspecify.annotations.Nullable;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
@@ -56,13 +58,13 @@ public class ContainerGroupSequencer implements ApplicationContextAware,
private final TaskExecutor executor = new SimpleAsyncTaskExecutor("container-group-sequencer-");
private ApplicationContext applicationContext;
private @Nullable ApplicationContext applicationContext;
private boolean stopLastGroupWhenIdle;
private Iterator<ContainerGroup> iterator;
private @Nullable Iterator<ContainerGroup> iterator;
private ContainerGroup currentGroup;
private @Nullable ContainerGroup currentGroup;
private boolean autoStartup = true;
@@ -136,7 +138,7 @@ public class ContainerGroupSequencer implements ApplicationContextAware,
MessageListenerContainer parent = event.getContainer(MessageListenerContainer.class);
MessageListenerContainer container = (MessageListenerContainer) event.getSource();
boolean inCurrentGroup = this.currentGroup != null && this.currentGroup.contains(parent);
if (this.running && inCurrentGroup && (this.iterator.hasNext() || this.stopLastGroupWhenIdle)) {
if (this.running && inCurrentGroup && (Objects.requireNonNull(this.iterator).hasNext() || this.stopLastGroupWhenIdle)) {
this.executor.execute(() -> {
LOGGER.debug(() -> "Stopping: " + container);
container.stop(() -> {
@@ -157,9 +159,16 @@ public class ContainerGroupSequencer implements ApplicationContextAware,
LOGGER.debug(() -> "Stopping: " + parent);
parent.stop(() -> {
if (this.currentGroup != null) {
LOGGER.debug(() -> "Checking group: " + this.currentGroup.toString());
LOGGER.debug(() -> {
if (this.currentGroup != null) {
return "Checking group: " + this.currentGroup.toString();
}
else {
return "Current group is null";
}
});
if (this.currentGroup.allStopped()) {
if (this.iterator.hasNext()) {
if (Objects.requireNonNull(this.iterator).hasNext()) {
this.currentGroup = this.iterator.next();
LOGGER.debug(() -> "Starting next group: " + this.currentGroup);
this.currentGroup.start();
@@ -185,7 +194,7 @@ public class ContainerGroupSequencer implements ApplicationContextAware,
public void initialize() {
this.groups.clear();
for (String group : this.groupNames) {
this.groups.add(this.applicationContext.getBean(group + ".group", ContainerGroup.class));
this.groups.add(Objects.requireNonNull(this.applicationContext).getBean(group + ".group", ContainerGroup.class));
}
if (!this.groups.isEmpty()) {
this.iterator = this.groups.iterator();
@@ -194,7 +203,7 @@ public class ContainerGroupSequencer implements ApplicationContextAware,
Collection<String> ids = grp.getListenerIds();
ids.stream().forEach(id -> {
MessageListenerContainer container = this.registry.getListenerContainer(id);
if (container.getContainerProperties().getIdleEventInterval() == null) {
if (Objects.requireNonNull(container).getContainerProperties().getIdleEventInterval() == null) {
container.getContainerProperties().setIdleEventInterval(this.defaultIdleEventInterval);
container.setAutoStartup(false);
}

6
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerPartitionPausingBackOffManagerFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022 the original author or authors.
* Copyright 2022-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.
@@ -16,6 +16,8 @@
package org.springframework.kafka.listener;
import org.jspecify.annotations.Nullable;
import org.springframework.context.ApplicationContext;
import org.springframework.util.Assert;
@@ -28,7 +30,7 @@ import org.springframework.util.Assert;
*/
public class ContainerPartitionPausingBackOffManagerFactory extends AbstractKafkaBackOffManagerFactory {
private BackOffHandler backOffHandler;
private @Nullable BackOffHandler backOffHandler;
/**
* Construct an instance with the provided properties.

2
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerPausingBackOffHandler.java
View File

@@ -43,7 +43,7 @@ public class ContainerPausingBackOffHandler implements BackOffHandler {
}
@Override
public void onNextBackOff(@Nullable MessageListenerContainer container, Exception exception, long nextBackOff) {
public void onNextBackOff(@Nullable MessageListenerContainer container, @Nullable Exception exception, long nextBackOff) {
if (container == null) {
this.defaultBackOffHandler.onNextBackOff(container, exception, nextBackOff); // NOSONAR
}

34
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java
View File

@@ -239,12 +239,12 @@ public class ContainerProperties extends ConsumerProperties {
* The message listener; must be a {@link org.springframework.kafka.listener.MessageListener}
* or {@link org.springframework.kafka.listener.AcknowledgingMessageListener}.
*/
private Object messageListener;
private @Nullable Object messageListener;
/**
* The executor for threads that poll the consumer.
*/
private AsyncTaskExecutor listenerTaskExecutor;
private @Nullable AsyncTaskExecutor listenerTaskExecutor;
/**
* The timeout for shutting down the container. This is the maximum amount of
@@ -253,22 +253,22 @@ public class ContainerProperties extends ConsumerProperties {
*/
private long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT;
private Long idleEventInterval;
private @Nullable Long idleEventInterval;
private Long idlePartitionEventInterval;
private @Nullable Long idlePartitionEventInterval;
private double idleBeforeDataMultiplier = DEFAULT_IDLE_BEFORE_DATA_MULTIPLIER;
@Deprecated(since = "3.2")
private PlatformTransactionManager transactionManager;
private @Nullable PlatformTransactionManager transactionManager;
private KafkaAwareTransactionManager<?, ?> kafkaAwareTransactionManager;
private @Nullable KafkaAwareTransactionManager<?, ?> kafkaAwareTransactionManager;
private boolean batchRecoverAfterRollback = false;
private int monitorInterval = DEFAULT_MONITOR_INTERVAL;
private TaskScheduler scheduler;
private @Nullable TaskScheduler scheduler;
private float noPollThreshold = DEFAULT_NO_POLL_THRESHOLD;
@@ -286,7 +286,7 @@ public class ContainerProperties extends ConsumerProperties {
private Duration consumerStartTimeout = DEFAULT_CONSUMER_START_TIMEOUT;
private Boolean subBatchPerPartition;
private @Nullable Boolean subBatchPerPartition;
private AssignmentCommitOption assignmentCommitOption = AssignmentCommitOption.LATEST_ONLY_NO_TX;
@@ -294,7 +294,7 @@ public class ContainerProperties extends ConsumerProperties {
private EOSMode eosMode = EOSMode.V2;
private TransactionDefinition transactionDefinition;
private @Nullable TransactionDefinition transactionDefinition;
private boolean stopContainerWhenFenced;
@@ -304,7 +304,7 @@ public class ContainerProperties extends ConsumerProperties {
private boolean pauseImmediate;
private KafkaListenerObservationConvention observationConvention;
private @Nullable KafkaListenerObservationConvention observationConvention;
private Duration pollTimeoutWhilePaused = DEFAULT_PAUSED_POLL_TIMEOUT;
@@ -327,7 +327,7 @@ public class ContainerProperties extends ConsumerProperties {
* @param topicPattern the pattern.
* @see org.apache.kafka.clients.CommonClientConfigs#METADATA_MAX_AGE_CONFIG
*/
public ContainerProperties(Pattern topicPattern) {
public ContainerProperties(@Nullable Pattern topicPattern) {
super(topicPattern);
}
@@ -486,7 +486,7 @@ public class ContainerProperties extends ConsumerProperties {
return this.ackTime;
}
public Object getMessageListener() {
public @Nullable Object getMessageListener() {
return this.messageListener;
}
@@ -1028,14 +1028,16 @@ public class ContainerProperties extends ConsumerProperties {
this.adviceChain.forEach(advised::addAdvice);
}
else {
ProxyFactory pf = new ProxyFactory(this.messageListener);
this.adviceChain.forEach(pf::addAdvice);
this.messageListener = pf.getProxy();
if (this.messageListener != null) {
ProxyFactory pf = new ProxyFactory(this.messageListener);
this.adviceChain.forEach(pf::addAdvice);
this.messageListener = pf.getProxy();
}
}
}
}
public KafkaListenerObservationConvention getObservationConvention() {
public @Nullable KafkaListenerObservationConvention getObservationConvention() {
return this.observationConvention;
}

54
spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java
View File

@@ -25,6 +25,7 @@ import java.util.Collections;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -87,11 +88,11 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
private final BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver;
private final Function<ProducerRecord<?, ?>, KafkaOperations<?, ?>> templateResolver;
private final Function<ProducerRecord<?, ?>, @Nullable KafkaOperations<?, ?>> templateResolver;
private final EnumSet<HeaderNames.HeadersToAdd> whichHeaders = EnumSet.allOf(HeaderNames.HeadersToAdd.class);
private HeaderNames headerNames = getHeaderNames();
private @Nullable HeaderNames headerNames = getHeaderNames();
private boolean retainExceptionHeader;
@@ -174,7 +175,7 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
* template from the map values iterator will be used.
* @param templates the {@link KafkaOperations}s to use for publishing.
*/
public DeadLetterPublishingRecoverer(Map<Class<?>, KafkaOperations<? extends Object, ? extends Object>> templates) {
public DeadLetterPublishingRecoverer(Map<Class<?>, @Nullable KafkaOperations<? extends Object, ? extends Object>> templates) {
this(templates, DEFAULT_DESTINATION_RESOLVER);
}
@@ -192,7 +193,7 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
* @param destinationResolver the resolving function.
*/
@SuppressWarnings("unchecked")
public DeadLetterPublishingRecoverer(Map<Class<?>, KafkaOperations<? extends Object, ? extends Object>> templates,
public DeadLetterPublishingRecoverer(Map<Class<?>, @Nullable KafkaOperations<? extends Object, ? extends Object>> templates,
BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver) {
Assert.isTrue(!ObjectUtils.isEmpty(templates), "At least one template is required");
@@ -205,7 +206,7 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
Boolean tx = this.transactional;
Assert.isTrue(templates.values()
.stream()
.map(t -> t.isTransactional())
.map(t -> Objects.requireNonNull(t).isTransactional())
.allMatch(t -> t.equals(tx)), "All templates must have the same setting for transactional");
this.destinationResolver = destinationResolver;
}
@@ -222,7 +223,7 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
* @param destinationResolver the resolving function.
* @since 3.0.9
*/
public DeadLetterPublishingRecoverer(Function<ProducerRecord<?, ?>, KafkaOperations<?, ?>> templateResolver,
public DeadLetterPublishingRecoverer(Function<ProducerRecord<?, ?>, @Nullable KafkaOperations<?, ?>> templateResolver,
BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver) {
this(templateResolver, false, destinationResolver);
}
@@ -240,7 +241,7 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
* @param destinationResolver the resolving function.
* @since 2.7
*/
public DeadLetterPublishingRecoverer(Function<ProducerRecord<?, ?>, KafkaOperations<?, ?>> templateResolver,
public DeadLetterPublishingRecoverer(Function<ProducerRecord<?, ?>, @Nullable KafkaOperations<?, ?>> templateResolver,
boolean transactional,
BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver) {
@@ -498,7 +499,7 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
}
}
@SuppressWarnings("unchecked")
@SuppressWarnings({"unchecked", "NullAway"})
@Override
public void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consumer, Exception exception) {
TopicPartition tp = this.destinationResolver.apply(record, exception);
@@ -626,8 +627,8 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
}
@SuppressWarnings("unchecked")
private KafkaOperations<Object, Object> findTemplateForValue(@Nullable Object value,
Map<Class<?>, KafkaOperations<?, ?>> templates) {
private @Nullable KafkaOperations<Object, Object> findTemplateForValue(@Nullable Object value,
Map<Class<?>, @Nullable KafkaOperations<?, ?>> templates) {
if (value == null) {
KafkaOperations<?, ?> operations = templates.get(Void.class);
@@ -667,6 +668,7 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
* @return the producer record to send.
* @see KafkaHeaders
*/
@SuppressWarnings("NullAway") // Dataflow analysis limitation
protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord<?, ?> record,
TopicPartition topicPartition, Headers headers, @Nullable byte[] key, @Nullable byte[] value) {
@@ -778,7 +780,7 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
}
private void maybeAddOriginalHeaders(Headers kafkaHeaders, ConsumerRecord<?, ?> record, Exception ex) {
maybeAddHeader(kafkaHeaders, this.headerNames.original.topicHeader,
maybeAddHeader(kafkaHeaders, Objects.requireNonNull(this.headerNames).original.topicHeader,
() -> record.topic().getBytes(StandardCharsets.UTF_8), HeaderNames.HeadersToAdd.TOPIC);
maybeAddHeader(kafkaHeaders, this.headerNames.original.partitionHeader,
() -> ByteBuffer.allocate(Integer.BYTES).putInt(record.partition()).array(),
@@ -837,13 +839,13 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
}
@Nullable
private String buildMessage(Exception exception, Throwable cause) {
private String buildMessage(Exception exception, @Nullable Throwable cause) {
String message = exception.getMessage();
if (!exception.equals(cause)) {
if (message != null) {
message = message + "; ";
}
String causeMsg = cause.getMessage();
String causeMsg = Objects.requireNonNull(cause).getMessage();
if (causeMsg != null) {
if (message != null) {
message = message + causeMsg;
@@ -1195,17 +1197,17 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
*/
public class Original {
private String offsetHeader;
private @Nullable String offsetHeader;
private String timestampHeader;
private @Nullable String timestampHeader;
private String timestampTypeHeader;
private @Nullable String timestampTypeHeader;
private String topicHeader;
private @Nullable String topicHeader;
private String partitionHeader;
private @Nullable String partitionHeader;
private String consumerGroupHeader;
private @Nullable String consumerGroupHeader;
/**
* Sets the name of the header that will be used to store the offset
@@ -1317,19 +1319,19 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
*/
public class ExceptionInfo {
private String keyExceptionFqcn;
private @Nullable String keyExceptionFqcn;
private String exceptionFqcn;
private @Nullable String exceptionFqcn;
private String exceptionCauseFqcn;
private @Nullable String exceptionCauseFqcn;
private String keyExceptionMessage;
private @Nullable String keyExceptionMessage;
private String exceptionMessage;
private @Nullable String exceptionMessage;
private String keyExceptionStacktrace;
private @Nullable String keyExceptionStacktrace;
private String exceptionStacktrace;
private @Nullable String exceptionStacktrace;
/**
* Sets the name of the header that will be used to store the keyExceptionFqcn

5
spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java
View File

@@ -20,6 +20,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;
@@ -65,7 +66,7 @@ public class DefaultAfterRollbackProcessor<K, V> extends FailedRecordProcessor
private final BackOff backOff;
private final KafkaOperations<?, ?> kafkaTemplate;
private final @Nullable KafkaOperations<?, ?> kafkaTemplate;
private final BiConsumer<ConsumerRecords<?, ?>, Exception> recoverer;
@@ -169,7 +170,7 @@ public class DefaultAfterRollbackProcessor<K, V> extends FailedRecordProcessor
if (SeekUtils.doSeeks((List) records, consumer, exception, recoverable,
getFailureTracker(), container, this.logger)
&& isCommitRecovered() && this.kafkaTemplate.isTransactional()) {
&& isCommitRecovered() && Objects.requireNonNull(this.kafkaTemplate).isTransactional()) {
ConsumerRecord<K, V> skipped = records.get(0);
this.kafkaTemplate.sendOffsetsToTransaction(
Collections.singletonMap(new TopicPartition(skipped.topic(), skipped.partition()),

2
spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultBackOffHandler.java
View File

@@ -30,7 +30,7 @@ import org.jspecify.annotations.Nullable;
public class DefaultBackOffHandler implements BackOffHandler {
@Override
public void onNextBackOff(@Nullable MessageListenerContainer container, Exception exception, long nextBackOff) {
public void onNextBackOff(@Nullable MessageListenerContainer container, @Nullable Exception exception, long nextBackOff) {
try {
if (container == null) {
Thread.sleep(nextBackOff);

5
spring-kafka/src/main/java/org/springframework/kafka/listener/DeliveryAttemptAwareRetryListener.java
View File

@@ -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.
@@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.jspecify.annotations.Nullable;
import org.springframework.kafka.support.KafkaHeaders;
@@ -37,7 +38,7 @@ import org.springframework.kafka.support.KafkaHeaders;
public class DeliveryAttemptAwareRetryListener implements RetryListener {
@Override
public void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt) {
public void failedDelivery(ConsumerRecord<?, ?> record, @Nullable Exception ex, int deliveryAttempt) {
// Pass
}

8
spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java
View File

@@ -19,6 +19,7 @@ package org.springframework.kafka.listener;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
import java.util.function.BiConsumer;
@@ -152,7 +153,8 @@ public final class ErrorHandlingUtils {
logger.debug(ex, () -> "Retry failed for: " + toLog);
recoveryException = ex;
Exception newException = unwrapIfNeeded(ex);
if (reClassifyOnExceptionChange && !newException.getClass().equals(lastException.getClass())
if (reClassifyOnExceptionChange && !Objects.requireNonNull(newException).getClass()
.equals(Objects.requireNonNull(lastException).getClass())
&& !classifier.classify(newException)) {
break;
@@ -204,7 +206,7 @@ public final class ErrorHandlingUtils {
* @return the unwrapped cause or cause of cause.
* @since 2.8.11
*/
public static Exception unwrapIfNeeded(Exception exception) {
public static @Nullable Exception unwrapIfNeeded(@Nullable Exception exception) {
Exception theEx = exception;
if (theEx instanceof TimestampedException && theEx.getCause() instanceof Exception cause) {
theEx = cause;
@@ -222,7 +224,7 @@ public final class ErrorHandlingUtils {
* @return the root cause.
* @since 3.0.7
*/
public static Exception findRootCause(Exception exception) {
public static @Nullable Exception findRootCause(@Nullable Exception exception) {
Exception realException = exception;
while ((realException instanceof ListenerExecutionFailedException
|| realException instanceof TimestampedException)

2
spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java
View File

@@ -41,7 +41,7 @@ public abstract class FailedRecordProcessor extends ExceptionClassifier implemen
private static final BackOff NO_RETRIES_OR_DELAY_BACKOFF = new FixedBackOff(0L, 0L);
private final BiFunction<ConsumerRecord<?, ?>, Exception, BackOff> noRetriesForClassified =
private final BiFunction<ConsumerRecord<?, ?>, @Nullable Exception, BackOff> noRetriesForClassified =
(rec, ex) -> {
Exception theEx = ErrorHandlingUtils.unwrapIfNeeded(ex);
if (!getClassifier().classify(theEx) || theEx instanceof KafkaBackoffException) {

34
spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordTracker.java
View File

@@ -21,6 +21,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
@@ -57,7 +58,7 @@ class FailedRecordTracker implements RecoveryStrategy {
private final BackOff backOff;
private BiFunction<ConsumerRecord<?, ?>, Exception, BackOff> backOffFunction;
private @Nullable BiFunction<ConsumerRecord<?, ?>, @Nullable Exception, BackOff> backOffFunction;
private final BackOffHandler backOffHandler;
@@ -71,6 +72,7 @@ class FailedRecordTracker implements RecoveryStrategy {
this(recoverer, backOff, null, logger);
}
@SuppressWarnings("NullAway") // Dataflow analysis limitation
FailedRecordTracker(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff,
@Nullable BackOffHandler backOffHandler, LogAccessor logger) {
@@ -111,7 +113,7 @@ class FailedRecordTracker implements RecoveryStrategy {
* @param backOffFunction the function.
* @since 2.6
*/
public void setBackOffFunction(@Nullable BiFunction<ConsumerRecord<?, ?>, Exception, BackOff> backOffFunction) {
public void setBackOffFunction(@Nullable BiFunction<ConsumerRecord<?, ?>, @Nullable Exception, BackOff> backOffFunction) {
this.backOffFunction = backOffFunction;
}
@@ -165,7 +167,7 @@ class FailedRecordTracker implements RecoveryStrategy {
}
@Override
public boolean recovered(ConsumerRecord<?, ?> record, Exception exception,
public boolean recovered(ConsumerRecord<?, ?> record, @Nullable Exception exception,
@Nullable MessageListenerContainer container,
@Nullable Consumer<?, ?> consumer) throws InterruptedException {
@@ -194,14 +196,14 @@ class FailedRecordTracker implements RecoveryStrategy {
}
}
private FailedRecord getFailedRecordInstance(ConsumerRecord<?, ?> record, Exception exception,
private FailedRecord getFailedRecordInstance(ConsumerRecord<?, ?> record, @Nullable Exception exception,
Map<TopicPartition, FailedRecord> map, TopicPartition topicPartition) {
Exception realException = ErrorHandlingUtils.findRootCause(exception);
FailedRecord failedRecord = map.get(topicPartition);
if (failedRecord == null || failedRecord.getOffset() != record.offset()
|| (this.resetStateOnExceptionChange
&& !realException.getClass().isInstance(failedRecord.getLastException()))) {
&& !Objects.requireNonNull(realException).getClass().isInstance(failedRecord.getLastException()))) {
failedRecord = new FailedRecord(record.offset(), determineBackOff(record, realException).start());
map.put(topicPartition, failedRecord);
@@ -213,7 +215,7 @@ class FailedRecordTracker implements RecoveryStrategy {
return failedRecord;
}
private BackOff determineBackOff(ConsumerRecord<?, ?> record, Exception exception) {
private BackOff determineBackOff(ConsumerRecord<?, ?> record, @Nullable Exception exception) {
if (this.backOffFunction == null) {
return this.backOff;
}
@@ -221,8 +223,8 @@ class FailedRecordTracker implements RecoveryStrategy {
return backOffToUse != null ? backOffToUse : this.backOff;
}
private void attemptRecovery(ConsumerRecord<?, ?> record, Exception exception, @Nullable TopicPartition tp,
Consumer<?, ?> consumer) {
private void attemptRecovery(ConsumerRecord<?, ?> record, @Nullable Exception exception, @Nullable TopicPartition tp,
@Nullable Consumer<?, ?> consumer) {
try {
this.recoverer.accept(record, consumer, exception);
@@ -231,7 +233,10 @@ class FailedRecordTracker implements RecoveryStrategy {
catch (RuntimeException e) {
this.retryListeners.forEach(rl -> rl.recoveryFailed(record, exception, e));
if (tp != null && this.resetStateOnRecoveryFailure) {
this.failures.get(Thread.currentThread()).remove(tp);
Map<TopicPartition, FailedRecord> topicPartitionFailedRecordMap = this.failures.get(Thread.currentThread());
if (topicPartitionFailedRecordMap != null) {
topicPartitionFailedRecordMap.remove(tp);
}
}
throw e;
}
@@ -257,7 +262,11 @@ class FailedRecordTracker implements RecoveryStrategy {
return 1;
}
FailedRecord failedRecord = map.get(topicPartitionOffset.getTopicPartition());
if (failedRecord == null || failedRecord.getOffset() != topicPartitionOffset.getOffset()) {
if (failedRecord == null) {
return 1;
}
Long offsetValue = topicPartitionOffset.getOffset();
if (offsetValue != null && failedRecord.getOffset() != offsetValue) {
return 1;
}
return failedRecord.getDeliveryAttempts().get() + 1;
@@ -271,7 +280,7 @@ class FailedRecordTracker implements RecoveryStrategy {
private final AtomicInteger deliveryAttempts = new AtomicInteger(1);
private Exception lastException;
private @Nullable Exception lastException;
FailedRecord(long offset, BackOffExecution backOffExecution) {
this.offset = offset;
@@ -290,11 +299,12 @@ class FailedRecordTracker implements RecoveryStrategy {
return this.deliveryAttempts;
}
@Nullable
Exception getLastException() {
return this.lastException;
}
void setLastException(Exception lastException) {
void setLastException(@Nullable Exception lastException) {
this.lastException = lastException;
}

2
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerBackoffManager.java
View File

@@ -62,7 +62,7 @@ public interface KafkaConsumerBackoffManager {
/**
* The consumer of the message, if present.
*/
private final Consumer<?, ?> consumerForTimingAdjustment;
private final @Nullable Consumer<?, ?> consumerForTimingAdjustment;
Context(long dueTimestamp, TopicPartition topicPartition, String listenerId,
@Nullable Consumer<?, ?> consumerForTimingAdjustment) {

4
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaListenerErrorHandler.java
View File

@@ -57,7 +57,7 @@ public interface KafkaListenerErrorHandler {
* {@code @SendTo} annotation.
*/
default Object handleError(Message<?> message, ListenerExecutionFailedException exception,
Consumer<?, ?> consumer) {
@Nullable Consumer<?, ?> consumer) {
return handleError(message, exception);
}
@@ -74,7 +74,7 @@ public interface KafkaListenerErrorHandler {
*/
@Nullable
default Object handleError(Message<?> message, ListenerExecutionFailedException exception,
Consumer<?, ?> consumer, @Nullable Acknowledgment ack) {
@Nullable Consumer<?, ?> consumer, @Nullable Acknowledgment ack) {
return handleError(message, exception, consumer);
}

213
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -187,15 +187,17 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private final AbstractMessageListenerContainer<K, V> thisOrParentContainer;
private final TopicPartitionOffset[] topicPartitions;
private final @Nullable TopicPartitionOffset @Nullable [] topicPartitions;
private String clientIdSuffix;
private @Nullable String clientIdSuffix;
private Runnable emergencyStop = () -> stopAbnormally(() -> {
});
@SuppressWarnings("NullAway.Init")
private volatile ListenerConsumer listenerConsumer;
@SuppressWarnings("NullAway.Init")
private volatile CompletableFuture<Void> listenerConsumerFuture;
private volatile CountDownLatch startLatch = new CountDownLatch(1);
@@ -234,7 +236,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
*/
KafkaMessageListenerContainer(@Nullable AbstractMessageListenerContainer<K, V> container,
ConsumerFactory<? super K, ? super V> consumerFactory,
ContainerProperties containerProperties, @Nullable TopicPartitionOffset... topicPartitions) {
ContainerProperties containerProperties, @Nullable TopicPartitionOffset @Nullable ... topicPartitions) {
super(consumerFactory, containerProperties);
Assert.notNull(consumerFactory, "A ConsumerFactory must be provided");
@@ -373,6 +375,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
containerProperties.setListenerTaskExecutor(consumerExecutor);
}
GenericMessageListener<?> listener = (GenericMessageListener<?>) messageListener;
Assert.state(listener != null, "'messageListener' cannot be null");
ListenerType listenerType = determineListenerType(listener);
ObservationRegistry observationRegistry = containerProperties.getObservationRegistry();
if (observationRegistry.isNoop()) {
@@ -635,11 +638,11 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private final GenericMessageListener<?> genericListener;
private final ConsumerSeekAware consumerSeekAwareListener;
private final @Nullable ConsumerSeekAware consumerSeekAwareListener;
private final MessageListener<K, V> listener;
private final @Nullable MessageListener<K, V> listener;
private final BatchMessageListener<K, V> batchListener;
private final @Nullable BatchMessageListener<K, V> batchListener;
private final ListenerType listenerType;
@@ -675,22 +678,22 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private final BlockingQueue<TopicPartitionOffset> seeks = new LinkedBlockingQueue<>();
private final CommonErrorHandler commonErrorHandler;
private final @Nullable CommonErrorHandler commonErrorHandler;
@Deprecated(since = "3.2", forRemoval = true)
@SuppressWarnings("removal")
private final PlatformTransactionManager transactionManager =
private final @Nullable PlatformTransactionManager transactionManager =
this.containerProperties.getKafkaAwareTransactionManager() != null ?
this.containerProperties.getKafkaAwareTransactionManager() :
this.containerProperties.getTransactionManager();
private final KafkaAwareTransactionManager<?, ?> kafkaTxManager =
private final @Nullable KafkaAwareTransactionManager<?, ?> kafkaTxManager =
this.transactionManager instanceof KafkaAwareTransactionManager<?, ?> kafkaAwareTransactionManager ?
kafkaAwareTransactionManager : null;
private final TransactionTemplate transactionTemplate;
private final @Nullable TransactionTemplate transactionTemplate;
private final String consumerGroupId = KafkaMessageListenerContainer.this.getGroupId();
private final @Nullable String consumerGroupId = KafkaMessageListenerContainer.this.getGroupId();
private final TaskScheduler taskScheduler;
@@ -709,39 +712,39 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private final boolean syncCommits = this.containerProperties.isSyncCommits();
private final Duration syncCommitTimeout;
private final @Nullable Duration syncCommitTimeout;
private final RecordInterceptor<K, V> recordInterceptor =
private final @Nullable RecordInterceptor<K, V> recordInterceptor =
!isInterceptBeforeTx() || this.transactionManager == null
? getRecordInterceptor()
: null;
private final RecordInterceptor<K, V> earlyRecordInterceptor =
private final @Nullable RecordInterceptor<K, V> earlyRecordInterceptor =
isInterceptBeforeTx() && this.transactionManager != null
? getRecordInterceptor()
: null;
private final RecordInterceptor<K, V> commonRecordInterceptor = getRecordInterceptor();
private final @Nullable RecordInterceptor<K, V> commonRecordInterceptor = getRecordInterceptor();
private final BatchInterceptor<K, V> batchInterceptor =
private final @Nullable BatchInterceptor<K, V> batchInterceptor =
!isInterceptBeforeTx() || this.transactionManager == null
? getBatchInterceptor()
: null;
private final BatchInterceptor<K, V> earlyBatchInterceptor =
private final @Nullable BatchInterceptor<K, V> earlyBatchInterceptor =
isInterceptBeforeTx() && this.transactionManager != null
? getBatchInterceptor()
: null;
private final BatchInterceptor<K, V> commonBatchInterceptor = getBatchInterceptor();
private final @Nullable BatchInterceptor<K, V> commonBatchInterceptor = getBatchInterceptor();
private final ThreadStateProcessor pollThreadStateProcessor;
private final @Nullable ThreadStateProcessor pollThreadStateProcessor;
private final ConsumerSeekCallback seekCallback = new InitialOrIdleSeekCallback();
private final long maxPollInterval;
private final MicrometerHolder micrometerHolder;
private final @Nullable MicrometerHolder micrometerHolder;
private final boolean observationEnabled;
@@ -749,20 +752,20 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private final boolean subBatchPerPartition = this.containerProperties.isSubBatchPerPartition();
private final Duration authExceptionRetryInterval =
private final @Nullable Duration authExceptionRetryInterval =
this.containerProperties.getAuthExceptionRetryInterval();
private final AssignmentCommitOption autoCommitOption = this.containerProperties.getAssignmentCommitOption();
private final boolean commitCurrentOnAssignment;
private final DeliveryAttemptAware deliveryAttemptAware;
private final @Nullable DeliveryAttemptAware deliveryAttemptAware;
private final EOSMode eosMode = this.containerProperties.getEosMode();
private final Map<TopicPartition, OffsetAndMetadata> commitsDuringRebalance = new HashMap<>();
private final String clientId;
private final @Nullable String clientId;
private final boolean fixTxOffsets = this.containerProperties.isFixTxOffsets();
@@ -770,9 +773,9 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private final Set<TopicPartition> pausedPartitions = new HashSet<>();
private final Map<TopicPartition, List<Long>> offsetsInThisBatch;
private final @Nullable Map<TopicPartition, List<Long>> offsetsInThisBatch;
private final Map<TopicPartition, List<ConsumerRecord<K, V>>> deferredOffsets;
private final @Nullable Map<TopicPartition, List<ConsumerRecord<K, V>>> deferredOffsets;
private final Map<TopicPartition, Long> lastReceivePartition;
@@ -793,15 +796,15 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
@Nullable
private final KafkaAdmin kafkaAdmin;
private final Object bootstrapServers;
private final @Nullable Object bootstrapServers;
@Nullable
private final Function<ConsumerRecord<?, ?>, Map<String, String>> micrometerTagsProvider =
this.containerProperties.getMicrometerTagsProvider();
private String clusterId;
private @Nullable String clusterId;
private Map<TopicPartition, OffsetMetadata> definedPartitions;
private @Nullable Map<TopicPartition, OffsetMetadata> definedPartitions;
private int count;
@@ -821,11 +824,11 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private int nackIndex;
private Iterator<TopicPartition> batchIterator;
private @Nullable Iterator<TopicPartition> batchIterator;
private ConsumerRecords<K, V> lastBatch;
private @Nullable ConsumerRecords<K, V> lastBatch;
private Producer<?, ?> producer;
private @Nullable Producer<?, ?> producer;
private boolean wasIdle;
@@ -835,7 +838,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private boolean receivedSome;
private ConsumerRecords<K, V> remainingRecords;
private @Nullable ConsumerRecords<K, V> remainingRecords;
private boolean pauseForPending;
@@ -843,7 +846,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private volatile boolean consumerPaused;
private volatile Thread consumerThread;
private volatile @Nullable Thread consumerThread;
private volatile long lastPoll = System.currentTimeMillis();
@@ -969,7 +972,10 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
this.kafkaAdmin = obtainAdmin();
if (isListenerAdapterObservationAware()) {
((RecordMessagingMessageListenerAdapter<?, ?>) this.listener).setObservationRegistry(observationRegistry);
RecordMessagingMessageListenerAdapter<?, ?> recordMessagingMessageListenerAdapter = (RecordMessagingMessageListenerAdapter<?, ?>) this.listener;
if (recordMessagingMessageListenerAdapter != null) {
recordMessagingMessageListenerAdapter.setObservationRegistry(observationRegistry);
}
}
}
@@ -1003,7 +1009,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
KafkaAdmin admin = applicationContext.getBeanProvider(KafkaAdmin.class).getIfUnique();
if (admin != null) {
Map<String, Object> props = new HashMap<>(admin.getConfigurationProperties());
if (!props.get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG).equals(this.bootstrapServers)) {
Object bootstrapServer = props.get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG);
if (bootstrapServer != null && !bootstrapServer.equals(this.bootstrapServers)) {
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
int opTo = admin.getOperationTimeout();
admin = new KafkaAdmin(props);
@@ -1047,11 +1054,12 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
return common;
}
@Nullable
String getClientId() {
return this.clientId;
}
private String determineClientId() {
private @Nullable String determineClientId() {
Map<MetricName, ? extends Metric> metrics = this.consumer.metrics();
Iterator<MetricName> metricIterator = metrics.keySet().iterator();
if (metricIterator.hasNext()) {
@@ -1087,7 +1095,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
}
else {
if (this.commonErrorHandler.deliveryAttemptHeader()) {
if (Objects.requireNonNull(this.commonErrorHandler).deliveryAttemptHeader()) {
aware = this.commonErrorHandler;
}
}
@@ -1115,6 +1123,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
return !this.autoCommit && resetLatest && latestOnlyOption;
}
@SuppressWarnings("NullAway") // Dataflow analysis limitation
private long obtainMaxPollInterval(Properties consumerProperties) {
Object timeout = consumerProperties.get(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG);
if (timeout == null) {
@@ -1139,7 +1148,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
+ ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG
+ "'; using Kafka default.");
}
return (int) CONSUMER_CONFIG_DEFAULTS.get(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG);
Object maxPollIntervalMs = CONSUMER_CONFIG_DEFAULTS.get(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG);
return maxPollIntervalMs == null ? null : (int) maxPollIntervalMs;
}
}
@@ -1195,7 +1205,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
return isAutoCommit;
}
private Duration determineSyncCommitTimeout() {
private @Nullable Duration determineSyncCommitTimeout() {
Duration syncTimeout = this.containerProperties.getSyncCommitTimeout();
if (syncTimeout != null) {
return syncTimeout;
@@ -1224,8 +1234,9 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
+ ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG
+ "'; defaulting to Kafka default for sync commit timeouts");
}
return Duration
.ofMillis((int) CONSUMER_CONFIG_DEFAULTS.get(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG));
Object defaultApiTimeout = CONSUMER_CONFIG_DEFAULTS.get(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG);
return defaultApiTimeout == null ? null : Duration
.ofMillis((int) defaultApiTimeout);
}
}
}
@@ -1275,7 +1286,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
if (KafkaUtils.MICROMETER_PRESENT && this.containerProperties.isMicrometerEnabled()
&& !this.observationEnabled) {
Function<Object, Map<String, String>> mergedProvider =
Function<@Nullable Object, Map<String, String>> mergedProvider =
cr -> this.containerProperties.getMicrometerTags();
if (this.micrometerTagsProvider != null) {
mergedProvider = cr -> {
@@ -1297,7 +1308,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void seekPartitions(Collection<TopicPartition> partitions, boolean idle) {
this.consumerSeekAwareListener.registerSeekCallback(this);
Objects.requireNonNull(this.consumerSeekAwareListener).registerSeekCallback(this);
Map<TopicPartition, Long> current = new HashMap<>();
for (TopicPartition topicPartition : partitions) {
current.put(topicPartition, ListenerConsumer.this.consumer.position(topicPartition));
@@ -1512,7 +1523,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
for (ConsumerRecord<K, V> kvConsumerRecord : pending) {
records.add(kvConsumerRecord);
}
this.commonErrorHandler.handleRemaining(cfe, records, this.consumer,
Objects.requireNonNull(this.commonErrorHandler).handleRemaining(cfe, records, this.consumer,
KafkaMessageListenerContainer.this.thisOrParentContainer);
}
}
@@ -1626,7 +1637,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
commitOffsets(toFix);
}
else {
this.transactionTemplate.executeWithoutResult(status -> {
Objects.requireNonNull(this.transactionTemplate).executeWithoutResult(status -> {
doSendOffsets(getTxProducer(), toFix);
});
}
@@ -1657,7 +1668,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
}
TopicPartition next = this.batchIterator.next();
List<ConsumerRecord<K, V>> subBatch = this.lastBatch.records(next);
List<ConsumerRecord<K, V>> subBatch = Objects.requireNonNull(this.lastBatch).records(next);
records = new ConsumerRecords<>(Collections.singletonMap(next, subBatch));
if (!this.batchIterator.hasNext()) {
this.batchIterator = null;
@@ -1705,7 +1716,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private synchronized void captureOffsets(ConsumerRecords<K, V> records) {
if (this.offsetsInThisBatch != null && records.count() > 0) {
this.offsetsInThisBatch.clear();
this.deferredOffsets.clear();
Objects.requireNonNull(this.deferredOffsets).clear();
records.partitions().forEach(part -> {
LinkedList<Long> offs = new LinkedList<>();
this.offsetsInThisBatch.put(part, offs);
@@ -1844,7 +1855,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void doResumeConsumerIfNeccessary() {
if (this.pausedForAsyncAcks && this.offsetsInThisBatch.isEmpty()) {
if (this.pausedForAsyncAcks && Objects.requireNonNull(this.offsetsInThisBatch).isEmpty()) {
this.pausedForAsyncAcks = false;
this.logger.debug("Resuming after manual async acks cleared");
}
@@ -2097,13 +2108,13 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private synchronized void ackInOrder(ConsumerRecord<K, V> cRecord) {
TopicPartition part = new TopicPartition(cRecord.topic(), cRecord.partition());
List<Long> offs = this.offsetsInThisBatch.get(part);
List<Long> offs = Objects.requireNonNull(this.offsetsInThisBatch).get(part);
if (!ObjectUtils.isEmpty(offs)) {
List<ConsumerRecord<K, V>> deferred = this.deferredOffsets.get(part);
List<ConsumerRecord<K, V>> deferred = Objects.requireNonNull(this.deferredOffsets).get(part);
if (offs.get(0) == cRecord.offset()) {
offs.remove(0);
ConsumerRecord<K, V> recordToAck = cRecord;
if (!deferred.isEmpty()) {
if (!CollectionUtils.isEmpty(deferred)) {
deferred.sort((a, b) -> Long.compare(a.offset(), b.offset()));
while (!ObjectUtils.isEmpty(deferred) && deferred.get(0).offset() == recordToAck.offset() + 1) {
recordToAck = deferred.remove(0);
@@ -2121,7 +2132,9 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
+ "; you are acknowledging a stale record: " + KafkaUtils.format(cRecord));
}
else {
if (deferred != null) {
deferred.add(cRecord);
}
}
}
else {
@@ -2158,7 +2171,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
if (records == null || records.count() == 0) {
return;
}
List<ConsumerRecord<K, V>> recordList = null;
List<ConsumerRecord<K, V>> recordList = new ArrayList<>();
if (!this.wantsFullRecords) {
recordList = createRecordList(records);
}
@@ -2173,10 +2186,10 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void invokeBatchListenerInTx(final ConsumerRecords<K, V> records,
@Nullable final List<ConsumerRecord<K, V>> recordList) {
final List<ConsumerRecord<K, V>> recordList) {
try {
this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
Objects.requireNonNull(this.transactionTemplate).execute(new TransactionCallbackWithoutResult() {
@Override
public void doInTransactionWithoutResult(TransactionStatus s) {
@@ -2205,7 +2218,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void batchRollback(final ConsumerRecords<K, V> records,
@Nullable final List<ConsumerRecord<K, V>> recordList, RuntimeException e) {
final List<ConsumerRecord<K, V>> recordList, RuntimeException e) {
@SuppressWarnings(UNCHECKED)
AfterRollbackProcessor<K, V> afterRollbackProcessorToUse =
@@ -2298,7 +2311,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void commitOffsetsIfNeededAfterHandlingError(final ConsumerRecords<K, V> records) {
if ((!this.autoCommit && this.commonErrorHandler.isAckAfterHandle() && this.consumerGroupId != null)
if ((!this.autoCommit && Objects.requireNonNull(this.commonErrorHandler).isAckAfterHandle() && this.consumerGroupId != null)
|| this.producer != null) {
if (this.remainingRecords != null) {
ConsumerRecord<K, V> firstUncommitted = this.remainingRecords.iterator().next();
@@ -2345,7 +2358,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void successTimer(@Nullable Object sample, @Nullable ConsumerRecord<?, ?> record) {
if (sample != null) {
if (sample != null && this.micrometerHolder != null) {
if (this.micrometerTagsProvider == null || record == null) {
this.micrometerHolder.success(sample);
}
@@ -2362,11 +2375,13 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
? exception.getCause().getClass().getSimpleName()
: exception.getClass().getSimpleName();
if (this.micrometerTagsProvider == null || record == null) {
this.micrometerHolder.failure(sample, exceptionName);
}
else {
this.micrometerHolder.failure(sample, exceptionName, record);
if (this.micrometerHolder != null) {
if (this.micrometerTagsProvider == null || record == null) {
this.micrometerHolder.failure(sample, exceptionName);
}
else {
this.micrometerHolder.failure(sample, exceptionName, record);
}
}
}
}
@@ -2409,7 +2424,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void invokeBatchOnMessageWithRecordsOrList(final ConsumerRecords<K, V> recordsArg,
@Nullable List<ConsumerRecord<K, V>> recordListArg) {
List<ConsumerRecord<K, V>> recordListArg) {
ConsumerRecords<K, V> records = recordsArg;
List<ConsumerRecord<K, V>> recordList = recordListArg;
@@ -2430,7 +2445,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
Object sample = startMicrometerSample();
try {
if (this.wantsFullRecords) {
this.batchListener.onMessage(records, // NOSONAR
Objects.requireNonNull(this.batchListener).onMessage(records, // NOSONAR
this.isAnyManualAck
? new ConsumerBatchAcknowledgment(records, recordList)
: null,
@@ -2451,22 +2466,22 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void doInvokeBatchOnMessage(final ConsumerRecords<K, V> records,
@Nullable List<ConsumerRecord<K, V>> recordList) {
List<ConsumerRecord<K, V>> recordList) {
try {
switch (this.listenerType) {
case ACKNOWLEDGING_CONSUMER_AWARE ->
this.batchListener.onMessage(recordList,
Objects.requireNonNull(this.batchListener).onMessage(recordList,
this.isAnyManualAck
? new ConsumerBatchAcknowledgment(records, recordList)
: null, this.consumer);
case ACKNOWLEDGING ->
this.batchListener.onMessage(recordList,
Objects.requireNonNull(this.batchListener).onMessage(recordList,
this.isAnyManualAck
? new ConsumerBatchAcknowledgment(records, recordList)
: null);
case CONSUMER_AWARE -> this.batchListener.onMessage(recordList, this.consumer);
case SIMPLE -> this.batchListener.onMessage(recordList);
case CONSUMER_AWARE -> Objects.requireNonNull(this.batchListener).onMessage(recordList, this.consumer);
case SIMPLE -> Objects.requireNonNull(this.batchListener).onMessage(recordList);
}
}
catch (Exception ex) { // NOSONAR
@@ -2475,9 +2490,9 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void invokeBatchErrorHandler(final ConsumerRecords<K, V> records,
@Nullable List<ConsumerRecord<K, V>> list, RuntimeException rte) {
List<ConsumerRecord<K, V>> list, RuntimeException rte) {
if (this.commonErrorHandler.seeksAfterHandling() || this.transactionManager != null
if (Objects.requireNonNull(this.commonErrorHandler).seeksAfterHandling() || this.transactionManager != null
|| rte instanceof CommitFailedException) {
this.commonErrorHandler.handleBatch(rte, records, this.consumer,
@@ -2547,7 +2562,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void invokeInTransaction(Iterator<ConsumerRecord<K, V>> iterator, final ConsumerRecord<K, V> cRecord) {
this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
Objects.requireNonNull(this.transactionTemplate).execute(new TransactionCallbackWithoutResult() {
@Override
public void doInTransactionWithoutResult(TransactionStatus s) {
@@ -2746,8 +2761,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
@SuppressWarnings(RAWTYPES)
private Producer<?, ?> getTxProducer() {
return ((KafkaResourceHolder) TransactionSynchronizationManager
.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
return ((KafkaResourceHolder) Objects.requireNonNull(TransactionSynchronizationManager
.getResource(Objects.requireNonNull(ListenerConsumer.this.kafkaTxManager).getProducerFactory())))
.getProducer(); // NOSONAR
}
@@ -2818,7 +2833,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
private void commitOffsetsIfNeededAfterHandlingError(final ConsumerRecord<K, V> cRecord) {
if ((!this.autoCommit && this.commonErrorHandler.isAckAfterHandle() && this.consumerGroupId != null)
if ((!this.autoCommit && Objects.requireNonNull(this.commonErrorHandler).isAckAfterHandle() && this.consumerGroupId != null)
|| this.producer != null) {
if (this.remainingRecords == null
|| !cRecord.equals(this.remainingRecords.iterator().next())) {
@@ -2884,17 +2899,17 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
try {
switch (this.listenerType) {
case ACKNOWLEDGING_CONSUMER_AWARE ->
this.listener.onMessage(cRecord,
Objects.requireNonNull(this.listener).onMessage(cRecord,
this.isAnyManualAck
? new ConsumerAcknowledgment(cRecord)
: null, this.consumer);
case CONSUMER_AWARE -> this.listener.onMessage(cRecord, this.consumer);
case CONSUMER_AWARE -> Objects.requireNonNull(this.listener).onMessage(cRecord, this.consumer);
case ACKNOWLEDGING ->
this.listener.onMessage(cRecord,
Objects.requireNonNull(this.listener).onMessage(cRecord,
this.isAnyManualAck
? new ConsumerAcknowledgment(cRecord)
: null);
case SIMPLE -> this.listener.onMessage(cRecord);
case SIMPLE -> Objects.requireNonNull(this.listener).onMessage(cRecord);
}
}
catch (Exception ex) { // NOSONAR
@@ -2906,7 +2921,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private void invokeErrorHandlerBySingleRecord(FailedRecordTuple<K, V> failedRecordTuple) {
final ConsumerRecord<K, V> cRecord = failedRecordTuple.record;
RuntimeException rte = failedRecordTuple.ex;
if (this.commonErrorHandler.seeksAfterHandling() || rte instanceof CommitFailedException) {
if (Objects.requireNonNull(this.commonErrorHandler).seeksAfterHandling() || rte instanceof CommitFailedException) {
try {
if (this.producer == null) {
processCommits();
@@ -2944,7 +2959,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private void invokeErrorHandler(final ConsumerRecord<K, V> cRecord,
Iterator<ConsumerRecord<K, V>> iterator, RuntimeException rte) {
if (this.commonErrorHandler.seeksAfterHandling() || rte instanceof CommitFailedException) {
if (Objects.requireNonNull(this.commonErrorHandler).seeksAfterHandling() || rte instanceof CommitFailedException) {
try {
if (this.producer == null) {
processCommits();
@@ -2990,7 +3005,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private RuntimeException decorateException(Exception ex) {
Exception toHandle = ex;
if (toHandle instanceof ListenerExecutionFailedException) {
toHandle = new ListenerExecutionFailedException(toHandle.getMessage(), this.consumerGroupId,
String message = toHandle.getMessage() == null ? "Error occurred" : toHandle.getMessage();
toHandle = new ListenerExecutionFailedException(message, this.consumerGroupId,
toHandle.getCause()); // NOSONAR restored below
fixStackTrace(ex, toHandle);
}
@@ -3073,11 +3089,11 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
doSendOffsets(this.producer, commits);
}
private void doSendOffsets(Producer<?, ?> prod, Map<TopicPartition, OffsetAndMetadata> commits) {
private void doSendOffsets(@Nullable Producer<?, ?> prod, Map<TopicPartition, OffsetAndMetadata> commits) {
if (CollectionUtils.isEmpty(commits)) {
return;
}
prod.sendOffsetsToTransaction(commits, this.consumer.groupMetadata());
Objects.requireNonNull(prod).sendOffsetsToTransaction(commits, this.consumer.groupMetadata());
if (this.fixTxOffsets) {
this.lastCommits.putAll(commits);
}
@@ -3153,12 +3169,15 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
Function<Long, Long> offsetComputeFunction = offset.getOffsetComputeFunction();
if (position == null) {
if (offset.isRelativeToCurrent()) {
whereTo += this.consumer.position(topicPartition);
long topicPartitionPosition = this.consumer.position(topicPartition);
Assert.state(whereTo != null, "Current offset must not be null");
whereTo += topicPartitionPosition;
whereTo = Math.max(whereTo, 0);
}
else if (offsetComputeFunction != null) {
whereTo = offsetComputeFunction.apply(this.consumer.position(topicPartition));
}
Assert.state(whereTo != null, "offset to seek cannot be null");
this.consumer.seek(topicPartition, whereTo);
}
else if (SeekPosition.TIMESTAMP.equals(position)) {
@@ -3260,7 +3279,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
doInitialSeeks(partitions, beginnings, ends);
if (this.consumerSeekAwareListener != null) {
this.consumerSeekAwareListener.onPartitionsAssigned(this.definedPartitions.keySet().stream()
this.consumerSeekAwareListener.onPartitionsAssigned(Objects.requireNonNull(this.definedPartitions).keySet().stream()
.map(tp -> new SimpleEntry<>(tp, this.consumer.position(tp)))
.collect(Collectors.toMap(SimpleEntry::getKey, SimpleEntry::getValue)),
this.seekCallback);
@@ -3475,7 +3494,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
@Override
public String getGroupId() {
public @Nullable String getGroupId() {
return this.consumerGroupId;
}
@@ -3541,7 +3560,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private final ConsumerRecords<K, V> records;
private final List<ConsumerRecord<K, V>> recordList;
private final @Nullable List<ConsumerRecord<K, V>> recordList;
private volatile boolean acked;
@@ -3566,7 +3585,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
for (TopicPartition topicPartition : this.records.partitions()) {
if (offs != null) {
offs.remove(topicPartition);
deferred.remove(topicPartition);
Objects.requireNonNull(deferred).remove(topicPartition);
}
}
processAcks(this.records);
@@ -3642,10 +3661,10 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
private class ListenerConsumerRebalanceListener implements ConsumerRebalanceListener {
private final ConsumerRebalanceListener userListener = getContainerProperties()
private final @Nullable ConsumerRebalanceListener userListener = getContainerProperties()
.getConsumerRebalanceListener();
private final ConsumerAwareRebalanceListener consumerAwareListener =
private final @Nullable ConsumerAwareRebalanceListener consumerAwareListener =
this.userListener instanceof ConsumerAwareRebalanceListener carl ? carl : null;
private final Collection<TopicPartition> revoked = new LinkedList<>();
@@ -3662,7 +3681,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
partitions);
}
else {
this.userListener.onPartitionsRevoked(partitions);
Objects.requireNonNull(this.userListener).onPartitionsRevoked(partitions);
}
try {
// Wait until now to commit, in case the user listener added acks
@@ -3691,7 +3710,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
if (pendingOffsets != null) {
partitions.forEach(tp -> {
pendingOffsets.remove(tp);
ListenerConsumer.this.deferredOffsets.remove(tp);
Objects.requireNonNull(ListenerConsumer.this.deferredOffsets).remove(tp);
});
if (pendingOffsets.isEmpty()) {
ListenerConsumer.this.consumerPaused = false;
@@ -3732,7 +3751,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
this.consumerAwareListener.onPartitionsAssigned(ListenerConsumer.this.consumer, partitions);
}
else {
this.userListener.onPartitionsAssigned(partitions);
Objects.requireNonNull(this.userListener).onPartitionsAssigned(partitions);
}
if (!ListenerConsumer.this.firstPoll && ListenerConsumer.this.definedPartitions == null
&& ListenerConsumer.this.consumerSeekAwareListener != null) {
@@ -3858,7 +3877,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
this.consumerAwareListener.onPartitionsLost(ListenerConsumer.this.consumer, partitions);
}
else {
this.userListener.onPartitionsLost(partitions);
Objects.requireNonNull(this.userListener).onPartitionsLost(partitions);
}
onPartitionsRevoked(partitions);
}
@@ -3995,12 +4014,12 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
* @param relativeToCurrent relative to current.
* @param seekPosition seek position strategy.
*/
private record OffsetMetadata(Long offset, boolean relativeToCurrent, SeekPosition seekPosition) {
private record OffsetMetadata(@Nullable Long offset, boolean relativeToCurrent, @Nullable SeekPosition seekPosition) {
}
private class StopCallback implements BiConsumer<Object, Throwable> {
private final Runnable callback;
private final @Nullable Runnable callback;
StopCallback(@Nullable Runnable callback) {
this.callback = callback;

3
spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerContainerPauseService.java
View File

@@ -19,6 +19,7 @@ package org.springframework.kafka.listener;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.util.Objects;
import java.util.Optional;
import org.apache.commons.logging.LogFactory;
@@ -140,7 +141,7 @@ public class ListenerContainerPauseService {
* Callers must ensure this.registry is not null before calling.
*/
private Optional<MessageListenerContainer> getListenerContainer(String listenerId) {
MessageListenerContainer messageListenerContainer = this.registry.getListenerContainer(listenerId); // NOSONAR
MessageListenerContainer messageListenerContainer = Objects.requireNonNull(this.registry).getListenerContainer(listenerId); // NOSONAR
if (messageListenerContainer == null) {
LOGGER.warn(() -> "MessageListenerContainer " + listenerId + " does not exists");
}

2
spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerExecutionFailedException.java
View File

@@ -28,7 +28,7 @@ import org.springframework.kafka.KafkaException;
@SuppressWarnings("serial")
public class ListenerExecutionFailedException extends KafkaException {
private final String groupId;
private final @Nullable String groupId;
/**
* Construct an instance with the provided properties.

13
spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java
View File

@@ -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.
@@ -17,9 +17,11 @@
package org.springframework.kafka.listener;
import java.util.Map;
import java.util.Objects;
import java.util.function.Supplier;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.jspecify.annotations.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.backoff.BackOff;
@@ -124,7 +126,7 @@ public final class ListenerUtils {
* @since 3.1
*/
public static void unrecoverableBackOff(BackOff backOff, Map<Thread, BackOffExecution> executions,
Map<Thread, Long> lastIntervals, MessageListenerContainer container) throws InterruptedException {
Map<Thread, Long> lastIntervals, @Nullable MessageListenerContainer container) throws InterruptedException {
Thread currentThread = Thread.currentThread();
Long interval = nextBackOff(backOff, executions);
@@ -158,8 +160,8 @@ public final class ListenerUtils {
* @throws InterruptedException if the thread is interrupted.
* @since 2.7
*/
public static void stoppableSleep(MessageListenerContainer container, long interval) throws InterruptedException {
conditionalSleep(container::isRunning, interval);
public static void stoppableSleep(@Nullable MessageListenerContainer container, long interval) throws InterruptedException {
conditionalSleep(Objects.requireNonNull(container)::isRunning, interval);
}
/**
@@ -188,8 +190,9 @@ public final class ListenerUtils {
* @return an offset and metadata.
* @since 2.8.6
*/
public static OffsetAndMetadata createOffsetAndMetadata(MessageListenerContainer container,
public static OffsetAndMetadata createOffsetAndMetadata(@Nullable MessageListenerContainer container,
long offset) {
Assert.state(container != null, "Container cannot be null");
final OffsetAndMetadataProvider metadataProvider = container.getContainerProperties()
.getOffsetAndMetadataProvider();
if (metadataProvider != null) {

2
spring-kafka/src/main/java/org/springframework/kafka/listener/ManualAckListenerErrorHandler.java
View File

@@ -38,7 +38,7 @@ public interface ManualAckListenerErrorHandler extends KafkaListenerErrorHandler
}
@Override
Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer,
Object handleError(Message<?> message, ListenerExecutionFailedException exception, @Nullable Consumer<?, ?> consumer,
@Nullable Acknowledgment ack);
}

2
spring-kafka/src/main/java/org/springframework/kafka/listener/RecoveryStrategy.java
View File

@@ -39,7 +39,7 @@ public interface RecoveryStrategy {
* @return true to skip.
* @throws InterruptedException if the thread is interrupted.
*/
boolean recovered(ConsumerRecord<?, ?> record, Exception ex, @Nullable MessageListenerContainer container,
boolean recovered(ConsumerRecord<?, ?> record, @Nullable Exception ex, @Nullable MessageListenerContainer container,
@Nullable Consumer<?, ?> consumer) throws InterruptedException;
}

9
spring-kafka/src/main/java/org/springframework/kafka/listener/RetryListener.java
View File

@@ -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,7 @@ package org.springframework.kafka.listener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.jspecify.annotations.Nullable;
/**
* A listener for retry activity.
@@ -35,14 +36,14 @@ public interface RetryListener {
* @param ex the exception.
* @param deliveryAttempt the delivery attempt.
*/
void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt);
void failedDelivery(ConsumerRecord<?, ?> record, @Nullable Exception ex, int deliveryAttempt);
/**
* Called after a failing record was successfully recovered.
* @param record the record.
* @param ex the exception.
*/
default void recovered(ConsumerRecord<?, ?> record, Exception ex) {
default void recovered(ConsumerRecord<?, ?> record, @Nullable Exception ex) {
}
/**
@@ -51,7 +52,7 @@ public interface RetryListener {
* @param original the original exception causing the recovery attempt.
* @param failure the exception thrown by the recoverer.
*/
default void recoveryFailed(ConsumerRecord<?, ?> record, Exception original, Exception failure) {
default void recoveryFailed(ConsumerRecord<?, ?> record, @Nullable Exception original, Exception failure) {
}
/**

4
spring-kafka/src/main/java/org/springframework/kafka/listener/SeekUtils.java
View File

@@ -77,7 +77,7 @@ public final class SeekUtils {
* @param logger a {@link LogAccessor} for seek errors.
* @return true if the failed record was skipped.
*/
public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, Exception exception,
public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, @Nullable Exception exception,
boolean recoverable, BiPredicate<ConsumerRecord<?, ?>, Exception> skipper, LogAccessor logger) {
return doSeeks(records, consumer, exception, recoverable, (rec, ex, cont, cons) -> skipper.test(rec, ex), null,
@@ -95,7 +95,7 @@ public final class SeekUtils {
* @param logger a {@link LogAccessor} for seek errors.
* @return true if the failed record was skipped.
*/
public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, Exception exception,
public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, @Nullable Exception exception,
boolean recoverable, RecoveryStrategy recovery, @Nullable MessageListenerContainer container,
LogAccessor logger) {

2
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractDelegatingMessageListenerAdapter.java
View File

@@ -80,7 +80,7 @@ public abstract class AbstractDelegatingMessageListenerAdapter<T>
}
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
public void onPartitionsRevoked(@Nullable Collection<TopicPartition> partitions) {
if (this.seekAware != null) {
this.seekAware.onPartitionsRevoked(partitions);
}

4
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java
View File

@@ -139,7 +139,7 @@ public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessage
*/
@Override
public void onMessage(List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer) {
@Nullable Consumer<?, ?> consumer) {
Message<?> message;
if (!isConsumerRecordList()) {
@@ -170,7 +170,7 @@ public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessage
@SuppressWarnings({ "unchecked", "rawtypes" })
protected Message<?> toMessagingMessage(List records, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer) {
@Nullable Consumer<?, ?> consumer) {
return getBatchMessageConverter().toMessage(records, acknowledgment, consumer, getType());
}

4
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchToRecordAdapter.java
View File

@@ -49,7 +49,7 @@ public interface BatchToRecordAdapter<K, V> {
* @param callback the callback.
*/
void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment ack,
Consumer<?, ?> consumer, Callback<K, V> callback);
@Nullable Consumer<?, ?> consumer, Callback<K, V> callback);
/**
* A callback for each message.
@@ -67,7 +67,7 @@ public interface BatchToRecordAdapter<K, V> {
* @param consumer the consumer.
* @param message the message.
*/
void invoke(ConsumerRecord<K, V> record, @Nullable Acknowledgment ack, Consumer<?, ?> consumer,
void invoke(ConsumerRecord<K, V> record, @Nullable Acknowledgment ack, @Nullable Consumer<?, ?> consumer,
Message<?> message);
}

2
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java
View File

@@ -107,7 +107,7 @@ public class ConvertingMessageListener<V> implements DelegatingMessageListener<M
@Override
@SuppressWarnings("unchecked")
public void onMessage(ConsumerRecord receivedRecord, @Nullable Acknowledgment acknowledgment, Consumer consumer) {
public void onMessage(ConsumerRecord receivedRecord, @Nullable Acknowledgment acknowledgment, @Nullable Consumer consumer) {
ConsumerRecord convertedConsumerRecord = convertConsumerRecord(receivedRecord);
if (this.delegate instanceof AcknowledgingConsumerAwareMessageListener) {
this.delegate.onMessage(convertedConsumerRecord, acknowledgment, consumer);

2
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DefaultBatchToRecordAdapter.java
View File

@@ -64,7 +64,7 @@ public class DefaultBatchToRecordAdapter<K, V> implements BatchToRecordAdapter<K
@Override
public void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment ack,
Consumer<?, ?> consumer, Callback<K, V> callback) {
@Nullable Consumer<?, ?> consumer, Callback<K, V> callback) {
for (int i = 0; i < messages.size(); i++) {
Message<?> message = messages.get(i);

14
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java
View File

@@ -392,7 +392,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
public void onPartitionsRevoked(@Nullable Collection<TopicPartition> partitions) {
if (this.bean instanceof ConsumerSeekAware csa) {
csa.onPartitionsRevoked(partitions);
}
@@ -406,12 +406,12 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
protected Message<?> toMessagingMessage(ConsumerRecord<K, V> cRecord, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer) {
@Nullable Consumer<?, ?> consumer) {
return getMessageConverter().toMessage(cRecord, acknowledgment, consumer, getType());
}
protected void invoke(Object records, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer,
protected void invoke(Object records, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer,
final Message<?> message) {
Throwable listenerError = null;
@@ -455,7 +455,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
*/
@Nullable
protected final Object invokeHandler(Object data, @Nullable Acknowledgment acknowledgment, Message<?> message,
Consumer<?, ?> consumer) {
@Nullable Consumer<?, ?> consumer) {
Acknowledgment ack = acknowledgment;
if (ack == null && this.noOpAck) {
@@ -511,7 +511,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
* {@code o.s.messaging.Message<?>}; may be null
*/
protected void handleResult(Object resultArg, Object request, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer, @Nullable Message<?> source) {
@Nullable Consumer<?, ?> consumer, @Nullable Message<?> source) {
final Observation observation = getCurrentObservation();
this.logger.debug(() -> "Listener method returned result [" + resultArg
+ "] - generating response message for it");
@@ -724,7 +724,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
@SuppressWarnings("NullAway") // Dataflow analysis limitation
protected void asyncFailure(Object request, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer,
protected void asyncFailure(Object request, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer,
@Nullable Throwable t, @Nullable Message<?> source) {
try {
@@ -749,7 +749,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
return request instanceof ConsumerRecord && exception instanceof RuntimeException;
}
protected void handleException(Object records, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer,
protected void handleException(Object records, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer,
@Nullable Message<?> message, ListenerExecutionFailedException e) {
if (this.errorHandler != null) {

2
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java
View File

@@ -70,7 +70,7 @@ public class RecordMessagingMessageListenerAdapter<K, V> extends MessagingMessag
*/
@Override
public void onMessage(ConsumerRecord<K, V> record, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer) {
@Nullable Consumer<?, ?> consumer) {
Message<?> message;
if (isConversionNeeded()) {

2
spring-kafka/src/main/java/org/springframework/kafka/listener/package-info.java
View File

@@ -1,5 +1,5 @@
/**
* Package for kafka listeners
*/
@org.springframework.lang.NonNullApi
@org.jspecify.annotations.NullMarked
package org.springframework.kafka.listener;

9
spring-kafka/src/main/java/org/springframework/kafka/requestreply/AggregatingReplyingKafkaTemplate.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2023 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.
@@ -34,6 +34,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.jspecify.annotations.Nullable;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.BatchConsumerAwareMessageListener;
@@ -125,7 +126,7 @@ public class AggregatingReplyingKafkaTemplate<K, V, R>
}
@Override
public void onMessage(List<ConsumerRecord<K, Collection<ConsumerRecord<K, R>>>> data, Consumer<?, ?> consumer) {
public void onMessage(List<ConsumerRecord<K, Collection<ConsumerRecord<K, R>>>> data, @Nullable Consumer<?, ?> consumer) {
List<ConsumerRecord<K, Collection<ConsumerRecord<K, R>>>> completed = new ArrayList<>();
String correlationHeaderName = getCorrelationHeaderName();
data.forEach(record -> {
@@ -191,11 +192,11 @@ public class AggregatingReplyingKafkaTemplate<K, V, R>
}
}
private void checkOffsetsAndCommitIfNecessary(List<ConsumerRecord<K, R>> list, Consumer<?, ?> consumer) {
private void checkOffsetsAndCommitIfNecessary(List<ConsumerRecord<K, R>> list, @Nullable Consumer<?, ?> consumer) {
list.forEach(record -> this.offsets.compute(
new TopicPartition(record.topic(), record.partition()),
(k, v) -> v == null ? record.offset() + 1 : Math.max(v, record.offset() + 1)));
if (this.pending.isEmpty() && !this.offsets.isEmpty()) {
if (this.pending.isEmpty() && !this.offsets.isEmpty() && consumer != null) {
consumer.commitSync(this.offsets.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey,
entry -> new OffsetAndMetadata(entry.getValue()))),

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

@@ -219,7 +219,7 @@ public class ReplyingKafkaTemplate<K, V, R> extends KafkaTemplate<K, V> implemen
* Return the topics/partitions assigned to the replying listener container.
* @return the topics/partitions.
*/
public Collection<TopicPartition> getAssignedReplyTopicPartitions() {
public @Nullable Collection<TopicPartition> getAssignedReplyTopicPartitions() {
return this.replyContainer.getAssignedPartitions();
}

2
spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessageConverter.java
View File

@@ -46,7 +46,7 @@ public interface BatchMessageConverter extends MessageConverter {
*/
@NonNull
Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type payloadType);
@Nullable Consumer<?, ?> consumer, Type payloadType);
/**
* Convert a message to a producer record.

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

@@ -148,7 +148,7 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
@Override // NOSONAR
public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {
@Nullable Consumer<?, ?> consumer, Type type) {
KafkaMessageHeaders kafkaMessageHeaders =
new KafkaMessageHeaders(this.generateMessageId, this.generateTimestamp);

12
spring-kafka/src/test/kotlin/org/springframework/kafka/listener/EnableKafkaKotlinTests.kt
View File

@@ -203,12 +203,12 @@ class EnableKafkaKotlinTests {
ConcurrentMessageListenerContainer<String, String> {
val container = kafkaListenerContainerFactory.createContainer("kotlinTestTopic2")
container.containerProperties.groupId = "checkedEx"
container.containerProperties.messageListener = MessageListener<String, String> {
container.containerProperties.setGroupId("checkedEx")
container.containerProperties.setMessageListener(MessageListener<String, String> {
if (it.value() == "fail") {
throw Exception("checked")
}
}
})
return container;
}
@@ -218,12 +218,12 @@ class EnableKafkaKotlinTests {
ConcurrentMessageListenerContainer<String, String> {
val container = kafkaBatchListenerContainerFactory.createContainer("kotlinBatchTestTopic2")
container.containerProperties.groupId = "batchCheckedEx"
container.containerProperties.messageListener = BatchMessageListener<String, String> {
container.containerProperties.setGroupId("batchCheckedEx")
container.containerProperties.setMessageListener(BatchMessageListener<String, String> {
if (it.first().value() == "fail") {
throw Exception("checked")
}
}
})
return container;
}