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

Add nullability changes in listener/adapter package

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-02-26 10:40:21 -05:00
parent efd348465a
commit 3d1e450190
18 changed files with 100 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -56,7 +56,7 @@ public interface GenericMessageListener<T> {
* @param consumer the consumer.
* @since 2.0
*/
default void onMessage(T data, Consumer<?, ?> consumer) {
default void onMessage(T data, @Nullable Consumer<?, ?> consumer) {
throw new UnsupportedOperationException("Container should never call this");
}
@@ -68,7 +68,7 @@ public interface GenericMessageListener<T> {
* @param consumer the consumer.
* @since 2.0
*/
default void onMessage(T data, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
default void onMessage(T data, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer) {
throw new UnsupportedOperationException("Container should never call this");
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2021 the original author or authors.
* Copyright 2018-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package org.springframework.kafka.listener;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.jspecify.annotations.Nullable;
/**
* Interface for backing off a {@link MessageListenerContainer}
@@ -32,7 +33,7 @@ public interface KafkaConsumerBackoffManager {
void backOffIfNecessary(Context context);
default Context createContext(long dueTimestamp, String listenerId, TopicPartition topicPartition,
Consumer<?, ?> messageConsumer) {
@Nullable Consumer<?, ?> messageConsumer) {
return new Context(dueTimestamp, topicPartition, listenerId, messageConsumer);
}
@@ -64,7 +65,7 @@ public interface KafkaConsumerBackoffManager {
private final Consumer<?, ?> consumerForTimingAdjustment;
Context(long dueTimestamp, TopicPartition topicPartition, String listenerId,
Consumer<?, ?> consumerForTimingAdjustment) {
@Nullable Consumer<?, ?> consumerForTimingAdjustment) {
this.dueTimestamp = dueTimestamp;
this.listenerId = listenerId;
@@ -84,7 +85,7 @@ public interface KafkaConsumerBackoffManager {
return this.topicPartition;
}
public Consumer<?, ?> getConsumerForTimingAdjustment() {
public @Nullable Consumer<?, ?> getConsumerForTimingAdjustment() {
return this.consumerForTimingAdjustment;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@ import java.util.Map;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.common.TopicPartition;
import org.jspecify.annotations.Nullable;
import org.springframework.core.log.LogAccessor;
import org.springframework.kafka.listener.ConsumerSeekAware;
@@ -46,7 +47,7 @@ public abstract class AbstractDelegatingMessageListenerAdapter<T>
protected final ListenerType delegateType; // NOSONAR
private final ConsumerSeekAware seekAware;
private final @Nullable ConsumerSeekAware seekAware;
public AbstractDelegatingMessageListenerAdapter(T delegate) {
this.delegate = delegate;

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

@@ -37,7 +37,7 @@ public abstract class AbstractRetryingMessageListenerAdapter<K, V, T>
private final RetryTemplate retryTemplate;
private final RecoveryCallback<? extends Object> recoveryCallback;
private final @Nullable RecoveryCallback<? extends Object> recoveryCallback;
/**
* Construct an instance with the supplied retry template. The exception will be
@@ -69,7 +69,7 @@ public abstract class AbstractRetryingMessageListenerAdapter<K, V, T>
return this.retryTemplate;
}
public RecoveryCallback<? extends Object> getRecoveryCallback() {
public @Nullable RecoveryCallback<? extends Object> getRecoveryCallback() {
return this.recoveryCallback;
}

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

@@ -62,7 +62,7 @@ public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessage
private BatchMessageConverter batchMessageConverter = new BatchMessagingMessageConverter();
private BatchToRecordAdapter<K, V> batchToRecordAdapter;
private @Nullable BatchToRecordAdapter<K, V> batchToRecordAdapter;
/**
* Create an instance with the provided parameters.

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

@@ -1,5 +1,5 @@
/*
* Copyright 2020 the original author or authors.
* Copyright 2020-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -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;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.Message;
@@ -47,7 +48,7 @@ public interface BatchToRecordAdapter<K, V> {
* @param consumer the consumer.
* @param callback the callback.
*/
void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records, Acknowledgment ack,
void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment ack,
Consumer<?, ?> consumer, Callback<K, V> callback);
/**
@@ -66,7 +67,7 @@ public interface BatchToRecordAdapter<K, V> {
* @param consumer the consumer.
* @param message the message.
*/
void invoke(ConsumerRecord<K, V> record, Acknowledgment ack, Consumer<?, ?> consumer,
void invoke(ConsumerRecord<K, V> record, @Nullable Acknowledgment ack, Consumer<?, ?> consumer,
Message<?> message);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@ import java.util.Map;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jspecify.annotations.Nullable;
import org.springframework.kafka.listener.AcknowledgingConsumerAwareMessageListener;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
@@ -60,7 +61,7 @@ public class ConvertingMessageListener<V> implements DelegatingMessageListener<M
private MessageConverter messageConverter;
private KafkaHeaderMapper headerMapper;
private @Nullable KafkaHeaderMapper headerMapper;
/**
* Construct an instance with the provided {@link MessageListener} and {@link Class}
@@ -106,7 +107,7 @@ public class ConvertingMessageListener<V> implements DelegatingMessageListener<M
@Override
@SuppressWarnings("unchecked")
public void onMessage(ConsumerRecord receivedRecord, Acknowledgment acknowledgment, Consumer consumer) {
public void onMessage(ConsumerRecord receivedRecord, @Nullable Acknowledgment acknowledgment, Consumer consumer) {
ConsumerRecord convertedConsumerRecord = convertConsumerRecord(receivedRecord);
if (this.delegate instanceof AcknowledgingConsumerAwareMessageListener) {
this.delegate.onMessage(convertedConsumerRecord, acknowledgment, consumer);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2020 the original author or authors.
* Copyright 2020-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -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;
import org.springframework.core.log.LogAccessor;
import org.springframework.kafka.listener.ConsumerRecordRecoverer;
@@ -62,7 +63,7 @@ public class DefaultBatchToRecordAdapter<K, V> implements BatchToRecordAdapter<K
}
@Override
public void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records, Acknowledgment ack,
public void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment ack,
Consumer<?, ?> consumer, Callback<K, V> callback) {
for (int i = 0; i < messages.size(); i++) {

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

@@ -69,7 +69,7 @@ public class DelegatingInvocableHandler {
private final ConcurrentMap<InvocableHandlerMethod, MethodParameter> payloadMethodParameters =
new ConcurrentHashMap<>();
private final InvocableHandlerMethod defaultHandler;
private final @Nullable InvocableHandlerMethod defaultHandler;
private final Map<InvocableHandlerMethod, Expression> handlerSendTo = new ConcurrentHashMap<>();
@@ -79,13 +79,13 @@ public class DelegatingInvocableHandler {
private final Object bean;
private final BeanExpressionResolver resolver;
private final @Nullable BeanExpressionResolver resolver;
private final BeanExpressionContext beanExpressionContext;
private final @Nullable BeanExpressionContext beanExpressionContext;
private final ConfigurableListableBeanFactory beanFactory;
private final @Nullable ConfigurableListableBeanFactory beanFactory;
private final PayloadValidator validator;
private final @Nullable PayloadValidator validator;
private final boolean asyncReplies;
@@ -168,7 +168,8 @@ public class DelegatingInvocableHandler {
* @throws Exception raised if no suitable argument resolver can be found,
* or the method raised an exception.
*/
public Object invoke(Message<?> message, Object... providedArgs) throws Exception { //NOSONAR
@SuppressWarnings("NullAway") // Dataflow analysis limitation
public Object invoke(Message<?> message, @Nullable Object... providedArgs) throws Exception { //NOSONAR
Class<?> payloadClass = message.getPayload().getClass();
InvocableHandlerMethod handler = getHandlerForPayload(payloadClass);
if (this.validator != null && this.defaultHandler != null) {
@@ -345,6 +346,7 @@ public class DelegatingInvocableHandler {
* @since 3.2
*/
@Nullable
@SuppressWarnings("NullAway") // Dataflow analysis limitation
public InvocationResult getInvocationResultFor(Object result, Object inboundPayload) {
InvocableHandlerMethod handler = findHandlerForPayload(inboundPayload.getClass());
if (handler != null) {

10
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java
View File

@@ -78,7 +78,7 @@ public class FilteringBatchMessageListenerAdapter<K, V>
@Override
public void onMessage(List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer) {
@Nullable Consumer<?, ?> consumer) {
final RecordFilterStrategy<K, V> recordFilterStrategy = getRecordFilterStrategy();
final List<ConsumerRecord<K, V>> consumerRecords = recordFilterStrategy.filterBatch(records);
@@ -100,8 +100,8 @@ public class FilteringBatchMessageListenerAdapter<K, V>
}
}
private void invokeDelegate(List<ConsumerRecord<K, V>> consumerRecords, Acknowledgment acknowledgment,
Consumer<?, ?> consumer) {
private void invokeDelegate(List<ConsumerRecord<K, V>> consumerRecords, @Nullable Acknowledgment acknowledgment,
@Nullable Consumer<?, ?> consumer) {
switch (this.delegateType) {
case ACKNOWLEDGING_CONSUMER_AWARE:
@@ -129,12 +129,12 @@ public class FilteringBatchMessageListenerAdapter<K, V>
}
@Override
public void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment) {
public void onMessage(List<ConsumerRecord<K, V>> data, @Nullable Acknowledgment acknowledgment) {
onMessage(data, acknowledgment, null); // NOSONAR
}
@Override
public void onMessage(List<ConsumerRecord<K, V>> data, Consumer<?, ?> consumer) {
public void onMessage(List<ConsumerRecord<K, V>> data, @Nullable Consumer<?, ?> consumer) {
onMessage(data, null, consumer);
}

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

@@ -66,7 +66,7 @@ public class FilteringMessageListenerAdapter<K, V>
@Override
public void onMessage(ConsumerRecord<K, V> consumerRecord, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer) {
@Nullable Consumer<?, ?> consumer) {
if (!filter(consumerRecord)) {
switch (this.delegateType) {
@@ -104,12 +104,12 @@ public class FilteringMessageListenerAdapter<K, V>
}
@Override
public void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment) {
public void onMessage(ConsumerRecord<K, V> data, @Nullable Acknowledgment acknowledgment) {
onMessage(data, acknowledgment, null); // NOSONAR
}
@Override
public void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer) {
public void onMessage(ConsumerRecord<K, V> data, @Nullable Consumer<?, ?> consumer) {
onMessage(data, null, consumer);
}

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

@@ -17,6 +17,7 @@
package org.springframework.kafka.listener.adapter;
import java.lang.reflect.Method;
import java.util.Objects;
import org.jspecify.annotations.Nullable;
@@ -35,9 +36,9 @@ import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
*/
public class HandlerAdapter {
private final InvocableHandlerMethod invokerHandlerMethod;
private final @Nullable InvocableHandlerMethod invokerHandlerMethod;
private final DelegatingInvocableHandler delegatingHandler;
private final @Nullable DelegatingInvocableHandler delegatingHandler;
private final boolean asyncReplies;
@@ -74,11 +75,11 @@ public class HandlerAdapter {
}
@Nullable
public Object invoke(Message<?> message, Object... providedArgs) throws Exception { //NOSONAR
public Object invoke(Message<?> message, @Nullable Object... providedArgs) throws Exception { //NOSONAR
if (this.invokerHandlerMethod != null) {
return this.invokerHandlerMethod.invoke(message, providedArgs); // NOSONAR
}
else if (this.delegatingHandler.hasDefaultHandler()) {
else if (Objects.requireNonNull(this.delegatingHandler).hasDefaultHandler()) {
// Needed to avoid returning raw Message which matches Object
Object[] args = new Object[providedArgs.length + 1];
args[0] = message.getPayload();
@@ -95,7 +96,7 @@ public class HandlerAdapter {
return this.invokerHandlerMethod.getMethod().toGenericString();
}
else {
return this.delegatingHandler.getMethodNameFor(payload);
return Objects.requireNonNull(this.delegatingHandler).getMethodNameFor(payload);
}
}
@@ -104,7 +105,7 @@ public class HandlerAdapter {
return this.invokerHandlerMethod.getBean();
}
else {
return this.delegatingHandler.getBean();
return Objects.requireNonNull(this.delegatingHandler).getBean();
}
}

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

@@ -101,7 +101,7 @@ public class KafkaBackoffAwareMessageListenerAdapter<K, V>
}
}
private void invokeDelegateOnMessage(ConsumerRecord<K, V> consumerRecord, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
private void invokeDelegateOnMessage(ConsumerRecord<K, V> consumerRecord, @Nullable Acknowledgment acknowledgment, @Nullable Consumer<?, ?> consumer) {
switch (this.delegateType) {
case ACKNOWLEDGING_CONSUMER_AWARE -> this.delegate.onMessage(consumerRecord, acknowledgment, consumer);
case ACKNOWLEDGING -> this.delegate.onMessage(consumerRecord, acknowledgment);
@@ -111,7 +111,7 @@ public class KafkaBackoffAwareMessageListenerAdapter<K, V>
}
private KafkaConsumerBackoffManager.Context createContext(ConsumerRecord<K, V> data, long nextExecutionTimestamp,
Consumer<?, ?> consumer) {
@Nullable Consumer<?, ?> consumer) {
return this.kafkaConsumerBackoffManager.createContext(nextExecutionTimestamp, this.listenerId,
new TopicPartition(data.topic(), data.partition()), consumer);
@@ -135,12 +135,12 @@ public class KafkaBackoffAwareMessageListenerAdapter<K, V>
}
@Override
public void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment) {
public void onMessage(ConsumerRecord<K, V> data, @Nullable Acknowledgment acknowledgment) {
onMessage(data, acknowledgment, null); // NOSONAR
}
@Override
public void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer) {
public void onMessage(ConsumerRecord<K, V> data, @Nullable Consumer<?, ?> consumer) {
onMessage(data, null, consumer);
}
}

7
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaMessageHandlerMethodFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2023-2024 the original author or authors.
* Copyright 2023-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,8 @@ package org.springframework.kafka.listener.adapter;
import java.lang.reflect.Method;
import java.util.List;
import org.jspecify.annotations.Nullable;
import org.springframework.core.KotlinDetector;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
@@ -39,9 +41,10 @@ public class KafkaMessageHandlerMethodFactory extends DefaultMessageHandlerMetho
private final HandlerMethodArgumentResolverComposite argumentResolvers =
new HandlerMethodArgumentResolverComposite();
@SuppressWarnings("NullAway.Init")
private MessageConverter messageConverter;
private Validator validator;
private @Nullable Validator validator;
@Override
public void setMessageConverter(MessageConverter messageConverter) {

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

@@ -40,12 +40,12 @@ import org.springframework.validation.Validator;
*/
public class KafkaNullAwarePayloadArgumentResolver extends PayloadMethodArgumentResolver {
KafkaNullAwarePayloadArgumentResolver(MessageConverter messageConverter, Validator validator) {
KafkaNullAwarePayloadArgumentResolver(MessageConverter messageConverter, @Nullable Validator validator) {
super(messageConverter, validator);
}
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception { // NOSONAR
public @Nullable Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception { // NOSONAR
Object resolved = super.resolveArgument(parameter, message);
/*
* Replace KafkaNull list elements with null.

7
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KotlinAwareInvocableHandlerMethod.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2023-2024 the original author or authors.
* Copyright 2023-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,6 +18,8 @@ package org.springframework.kafka.listener.adapter;
import java.lang.reflect.Method;
import org.jspecify.annotations.Nullable;
import org.springframework.core.CoroutinesUtils;
import org.springframework.core.KotlinDetector;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
@@ -36,7 +38,8 @@ public class KotlinAwareInvocableHandlerMethod extends InvocableHandlerMethod {
}
@Override
protected Object doInvoke(Object... args) throws Exception {
@Nullable
protected Object doInvoke(@Nullable Object... args) throws Exception {
Method method = getBridgedMethod();
if (KotlinDetector.isSuspendingFunction(method)) {
return CoroutinesUtils.invokeSuspendingFunction(method, getBean(), args);

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

@@ -115,11 +115,11 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); //NOSONAR
private final Type inferredType;
private final @Nullable Type inferredType;
private final StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
private final KafkaListenerErrorHandler errorHandler;
private final @Nullable KafkaListenerErrorHandler errorHandler;
@Nullable
private HandlerAdapter handlerMethod;
@@ -138,10 +138,10 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
private Type fallbackType = Object.class;
private Expression replyTopicExpression;
private @Nullable Expression replyTopicExpression;
@SuppressWarnings("rawtypes")
private KafkaTemplate replyTemplate;
private @Nullable KafkaTemplate replyTemplate;
private boolean hasAckParameter;
@@ -151,7 +151,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
private boolean messageReturnType;
private ReplyHeadersConfigurer replyHeadersConfigurer;
private @Nullable ReplyHeadersConfigurer replyHeadersConfigurer;
private boolean splitIterables = true;
@@ -159,7 +159,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
private BiConsumer<ConsumerRecord<K, V>, RuntimeException> asyncRetryCallback;
private @Nullable BiConsumer<ConsumerRecord<K, V>, RuntimeException> asyncRetryCallback;
/**
* Create an instance with the provided bean and method.
@@ -345,7 +345,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
* @since 2.2
* @see #setReplyHeadersConfigurer(ReplyHeadersConfigurer)
*/
protected ReplyHeadersConfigurer getReplyHeadersConfigurer() {
protected @Nullable ReplyHeadersConfigurer getReplyHeadersConfigurer() {
return this.replyHeadersConfigurer;
}
@@ -567,7 +567,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
@Nullable
private String evaluateReplyTopic(Object request, Object source, Object result) {
private String evaluateReplyTopic(Object request, @Nullable Object source, Object result) {
String replyTo = null;
if (result instanceof InvocationResult invResult) {
replyTo = evaluateTopic(request, source, result, invResult.sendTo());
@@ -579,7 +579,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
@Nullable
private String evaluateTopic(Object request, Object source, Object result, @Nullable Expression sendTo) {
private String evaluateTopic(Object request, @Nullable Object source, Object result, @Nullable Expression sendTo) {
if (sendTo instanceof LiteralExpression) {
return sendTo.getValue(String.class);
}
@@ -617,21 +617,21 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
else if (result instanceof Message<?> mResult) {
Message<?> reply = checkHeaders(mResult, topic, source);
this.replyTemplate.send(reply);
Objects.requireNonNull(this.replyTemplate).send(reply);
}
else if (result instanceof Iterable<?> iterable && (iterableOfMessages(iterable) || this.splitIterables)) {
iterable.forEach(v -> {
if (v instanceof Message<?> mv) {
Message<?> aReply = checkHeaders(mv, topic, source);
this.replyTemplate.send(aReply);
Objects.requireNonNull(this.replyTemplate).send(aReply);
}
else {
this.replyTemplate.send(topic, v);
Objects.requireNonNull(this.replyTemplate).send(Objects.requireNonNull(topic), v);
}
});
}
else {
sendSingleResult(result, topic, source);
sendSingleResult(result, Objects.requireNonNull(topic), source);
}
}
@@ -645,9 +645,9 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
boolean needsTopic = topic != null && headers.get(KafkaHeaders.TOPIC) == null;
boolean sourceIsMessage = source instanceof Message;
boolean needsCorrelation = headers.get(this.correlationHeaderName) == null && sourceIsMessage
&& getCorrelation((Message<?>) source) != null;
&& getCorrelation(Objects.requireNonNull((Message<?>) source)) != null;
boolean needsPartition = headers.get(KafkaHeaders.PARTITION) == null && sourceIsMessage
&& getReplyPartition((Message<?>) source) != null;
&& getReplyPartition(Objects.requireNonNull((Message<?>) source)) != null;
if (needsTopic || needsCorrelation || needsPartition) {
MessageBuilder<?> builder = MessageBuilder.fromMessage(reply);
if (needsTopic) {
@@ -670,12 +670,12 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
sendReplyForMessageSource(result, topic, message, getCorrelation(message));
}
else {
this.replyTemplate.send(topic, result);
Objects.requireNonNull(this.replyTemplate).send(topic, result);
}
}
@SuppressWarnings("unchecked")
private void sendReplyForMessageSource(Object result, String topic, Message<?> source, @Nullable byte[] correlationId) {
private void sendReplyForMessageSource(Object result, String topic, Message<?> source, byte[] correlationId) {
MessageBuilder<Object> builder = MessageBuilder.withPayload(result)
.setHeader(KafkaHeaders.TOPIC, topic);
if (this.replyHeadersConfigurer != null) {
@@ -701,10 +701,10 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
setPartition(builder, source);
setKey(builder, source);
this.replyTemplate.send(builder.build());
Objects.requireNonNull(this.replyTemplate).send(builder.build());
}
protected void asyncSuccess(@Nullable Object result, String replyTopic, Message<?> source,
protected void asyncSuccess(@Nullable Object result, @Nullable String replyTopic, @Nullable Message<?> source,
boolean returnTypeMessage) {
if (result == null) {
@@ -723,14 +723,15 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
}
@SuppressWarnings("NullAway") // Dataflow analysis limitation
protected void asyncFailure(Object request, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer,
Throwable t, Message<?> source) {
@Nullable Throwable t, @Nullable Message<?> source) {
try {
Throwable cause = t instanceof CompletionException ? t.getCause() : t;
handleException(request, acknowledgment, consumer, source,
new ListenerExecutionFailedException(createMessagingErrorMessage(
"Async Fail", source.getPayload()), cause));
"Async Fail", Objects.requireNonNull(source).getPayload()), cause));
}
catch (Throwable ex) {
this.logger.error(t, () -> "Future, Mono, or suspend function was completed with an exception for " + source);
@@ -749,15 +750,15 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
protected void handleException(Object records, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer,
Message<?> message, ListenerExecutionFailedException e) {
@Nullable Message<?> message, ListenerExecutionFailedException e) {
if (this.errorHandler != null) {
try {
if (NULL_MESSAGE.equals(message)) {
message = new GenericMessage<>(records);
}
Object errorResult = this.errorHandler.handleError(message, e, consumer, acknowledgment);
if (errorResult != null && !(errorResult instanceof InvocationResult)) {
Object errorResult = this.errorHandler.handleError(Objects.requireNonNull(message), e, consumer, acknowledgment);
if (errorResult != null && !(errorResult instanceof InvocationResult) && this.handlerMethod != null) {
Object result = this.handlerMethod.getInvocationResultFor(errorResult, message.getPayload());
handleResult(Objects.requireNonNullElse(result, errorResult),
records, acknowledgment, consumer, message);
@@ -766,7 +767,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
catch (Exception ex) {
throw new ListenerExecutionFailedException(createMessagingErrorMessage(// NOSONAR stack trace loss
"Listener error handler threw an exception for the incoming message",
message.getPayload()), ex);
Objects.requireNonNull(message).getPayload()), ex);
}
}
else {
@@ -774,42 +775,42 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
}
private void setCorrelation(MessageBuilder<?> builder, Message<?> source) {
private void setCorrelation(MessageBuilder<?> builder, @Nullable Message<?> source) {
byte[] correlationBytes = getCorrelation(source);
if (correlationBytes != null) {
builder.setHeader(this.correlationHeaderName, correlationBytes);
}
}
@Nullable
private byte[] getCorrelation(Message<?> source) {
@SuppressWarnings("NullAway") // Dataflow analysis limitation
private byte[] getCorrelation(@Nullable Message<?> source) {
return source.getHeaders().get(this.correlationHeaderName, byte[].class);
}
private void setPartition(MessageBuilder<?> builder, Message<?> source) {
private void setPartition(MessageBuilder<?> builder, @Nullable Message<?> source) {
byte[] partitionBytes = getReplyPartition(source);
if (partitionBytes != null) {
builder.setHeader(KafkaHeaders.PARTITION, ByteBuffer.wrap(partitionBytes).getInt());
}
}
private void setKey(MessageBuilder<?> builder, Message<?> source) {
Object key = source.getHeaders().get(KafkaHeaders.RECEIVED_KEY);
private void setKey(MessageBuilder<?> builder, @Nullable Message<?> source) {
Object key = Objects.requireNonNull(source).getHeaders().get(KafkaHeaders.RECEIVED_KEY);
// Set the reply record key only for non-batch requests
if (key != null && !(key instanceof List)) {
builder.setHeader(KafkaHeaders.KEY, key);
}
}
@Nullable
private byte[] getReplyPartition(Message<?> source) {
@SuppressWarnings("NullAway") // Dataflow analysis limitation
private byte[] getReplyPartition(@Nullable Message<?> source) {
return source.getHeaders().get(KafkaHeaders.REPLY_PARTITION, byte[].class);
}
protected final String createMessagingErrorMessage(String description, Object payload) {
return description + "\n"
+ "Endpoint handler details:\n"
+ "Method [" + this.handlerMethod.getMethodAsString(payload) + "]\n"
+ "Method [" + Objects.requireNonNull(this.handlerMethod).getMethodAsString(payload) + "]\n"
+ "Bean [" + this.handlerMethod.getBean() + "]";
}
@@ -941,7 +942,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
* @param result the result.
* @since 2.0
*/
public record ReplyExpressionRoot(Object request, Object source, Object result) {
public record ReplyExpressionRoot(Object request, @Nullable Object source, Object result) {
}

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

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