From 3d1e4501907a355417d94898dd1cb070e5d44e83 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 26 Feb 2025 10:40:21 -0500 Subject: [PATCH] Add nullability changes in listener/adapter package https://github.com/spring-projects/spring-kafka/issues/3762 Signed-off-by: Soby Chacko --- .../listener/GenericMessageListener.java | 4 +- .../listener/KafkaConsumerBackoffManager.java | 9 +-- ...tractDelegatingMessageListenerAdapter.java | 5 +- ...bstractRetryingMessageListenerAdapter.java | 4 +- .../BatchMessagingMessageListenerAdapter.java | 2 +- .../adapter/BatchToRecordAdapter.java | 7 +- .../adapter/ConvertingMessageListener.java | 7 +- .../adapter/DefaultBatchToRecordAdapter.java | 5 +- .../adapter/DelegatingInvocableHandler.java | 14 ++-- .../FilteringBatchMessageListenerAdapter.java | 10 +-- .../FilteringMessageListenerAdapter.java | 6 +- .../listener/adapter/HandlerAdapter.java | 13 ++-- ...fkaBackoffAwareMessageListenerAdapter.java | 8 +-- .../KafkaMessageHandlerMethodFactory.java | 7 +- ...KafkaNullAwarePayloadArgumentResolver.java | 4 +- .../KotlinAwareInvocableHandlerMethod.java | 7 +- .../MessagingMessageListenerAdapter.java | 71 ++++++++++--------- .../kafka/listener/adapter/package-info.java | 2 +- 18 files changed, 100 insertions(+), 85 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericMessageListener.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericMessageListener.java index eafb64b7..daaf560e 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericMessageListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericMessageListener.java @@ -56,7 +56,7 @@ public interface GenericMessageListener { * @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 { * @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"); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerBackoffManager.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerBackoffManager.java index 8a00e058..9b8becc8 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerBackoffManager.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaConsumerBackoffManager.java @@ -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; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractDelegatingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractDelegatingMessageListenerAdapter.java index 4be85178..e393409b 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractDelegatingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractDelegatingMessageListenerAdapter.java @@ -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 protected final ListenerType delegateType; // NOSONAR - private final ConsumerSeekAware seekAware; + private final @Nullable ConsumerSeekAware seekAware; public AbstractDelegatingMessageListenerAdapter(T delegate) { this.delegate = delegate; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.java index 84c8b6cb..e56ec588 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.java @@ -37,7 +37,7 @@ public abstract class AbstractRetryingMessageListenerAdapter private final RetryTemplate retryTemplate; - private final RecoveryCallback recoveryCallback; + private final @Nullable RecoveryCallback recoveryCallback; /** * Construct an instance with the supplied retry template. The exception will be @@ -69,7 +69,7 @@ public abstract class AbstractRetryingMessageListenerAdapter return this.retryTemplate; } - public RecoveryCallback getRecoveryCallback() { + public @Nullable RecoveryCallback getRecoveryCallback() { return this.recoveryCallback; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java index 93910236..fe705082 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java @@ -62,7 +62,7 @@ public class BatchMessagingMessageListenerAdapter extends MessagingMessage private BatchMessageConverter batchMessageConverter = new BatchMessagingMessageConverter(); - private BatchToRecordAdapter batchToRecordAdapter; + private @Nullable BatchToRecordAdapter batchToRecordAdapter; /** * Create an instance with the provided parameters. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchToRecordAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchToRecordAdapter.java index 0cd1fa7d..9bea4598 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchToRecordAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchToRecordAdapter.java @@ -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 { * @param consumer the consumer. * @param callback the callback. */ - void adapt(List> messages, List> records, Acknowledgment ack, + void adapt(List> messages, List> records, @Nullable Acknowledgment ack, Consumer consumer, Callback callback); /** @@ -66,7 +67,7 @@ public interface BatchToRecordAdapter { * @param consumer the consumer. * @param message the message. */ - void invoke(ConsumerRecord record, Acknowledgment ack, Consumer consumer, + void invoke(ConsumerRecord record, @Nullable Acknowledgment ack, Consumer consumer, Message message); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java index d2f3ae57..2e7d4cd6 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java @@ -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 implements DelegatingMessageListener implements DelegatingMessageListener implements BatchToRecordAdapter> messages, List> records, Acknowledgment ack, + public void adapt(List> messages, List> records, @Nullable Acknowledgment ack, Consumer consumer, Callback callback) { for (int i = 0; i < messages.size(); i++) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java index ed6c845b..adfc2fac 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java @@ -69,7 +69,7 @@ public class DelegatingInvocableHandler { private final ConcurrentMap payloadMethodParameters = new ConcurrentHashMap<>(); - private final InvocableHandlerMethod defaultHandler; + private final @Nullable InvocableHandlerMethod defaultHandler; private final Map 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) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java index b898c573..aef6b983 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java @@ -78,7 +78,7 @@ public class FilteringBatchMessageListenerAdapter @Override public void onMessage(List> records, @Nullable Acknowledgment acknowledgment, - Consumer consumer) { + @Nullable Consumer consumer) { final RecordFilterStrategy recordFilterStrategy = getRecordFilterStrategy(); final List> consumerRecords = recordFilterStrategy.filterBatch(records); @@ -100,8 +100,8 @@ public class FilteringBatchMessageListenerAdapter } } - private void invokeDelegate(List> consumerRecords, Acknowledgment acknowledgment, - Consumer consumer) { + private void invokeDelegate(List> consumerRecords, @Nullable Acknowledgment acknowledgment, + @Nullable Consumer consumer) { switch (this.delegateType) { case ACKNOWLEDGING_CONSUMER_AWARE: @@ -129,12 +129,12 @@ public class FilteringBatchMessageListenerAdapter } @Override - public void onMessage(List> data, Acknowledgment acknowledgment) { + public void onMessage(List> data, @Nullable Acknowledgment acknowledgment) { onMessage(data, acknowledgment, null); // NOSONAR } @Override - public void onMessage(List> data, Consumer consumer) { + public void onMessage(List> data, @Nullable Consumer consumer) { onMessage(data, null, consumer); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringMessageListenerAdapter.java index 21984b64..3f172be0 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringMessageListenerAdapter.java @@ -66,7 +66,7 @@ public class FilteringMessageListenerAdapter @Override public void onMessage(ConsumerRecord consumerRecord, @Nullable Acknowledgment acknowledgment, - Consumer consumer) { + @Nullable Consumer consumer) { if (!filter(consumerRecord)) { switch (this.delegateType) { @@ -104,12 +104,12 @@ public class FilteringMessageListenerAdapter } @Override - public void onMessage(ConsumerRecord data, Acknowledgment acknowledgment) { + public void onMessage(ConsumerRecord data, @Nullable Acknowledgment acknowledgment) { onMessage(data, acknowledgment, null); // NOSONAR } @Override - public void onMessage(ConsumerRecord data, Consumer consumer) { + public void onMessage(ConsumerRecord data, @Nullable Consumer consumer) { onMessage(data, null, consumer); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/HandlerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/HandlerAdapter.java index 1e5a9fff..b5eb5e9b 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/HandlerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/HandlerAdapter.java @@ -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(); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapter.java index 71766637..200b3dd6 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapter.java @@ -101,7 +101,7 @@ public class KafkaBackoffAwareMessageListenerAdapter } } - private void invokeDelegateOnMessage(ConsumerRecord consumerRecord, Acknowledgment acknowledgment, Consumer consumer) { + private void invokeDelegateOnMessage(ConsumerRecord 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 } private KafkaConsumerBackoffManager.Context createContext(ConsumerRecord 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 } @Override - public void onMessage(ConsumerRecord data, Acknowledgment acknowledgment) { + public void onMessage(ConsumerRecord data, @Nullable Acknowledgment acknowledgment) { onMessage(data, acknowledgment, null); // NOSONAR } @Override - public void onMessage(ConsumerRecord data, Consumer consumer) { + public void onMessage(ConsumerRecord data, @Nullable Consumer consumer) { onMessage(data, null, consumer); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaMessageHandlerMethodFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaMessageHandlerMethodFactory.java index 9e0a49dc..a1a96484 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaMessageHandlerMethodFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaMessageHandlerMethodFactory.java @@ -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) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaNullAwarePayloadArgumentResolver.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaNullAwarePayloadArgumentResolver.java index 9cec2189..3519f637 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaNullAwarePayloadArgumentResolver.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaNullAwarePayloadArgumentResolver.java @@ -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. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KotlinAwareInvocableHandlerMethod.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KotlinAwareInvocableHandlerMethod.java index 88451b0f..dbd15c1d 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KotlinAwareInvocableHandlerMethod.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KotlinAwareInvocableHandlerMethod.java @@ -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); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java index 68c6d612..0b038b56 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java @@ -115,11 +115,11 @@ public abstract class MessagingMessageListenerAdapter 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 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 implements ConsumerS private boolean messageReturnType; - private ReplyHeadersConfigurer replyHeadersConfigurer; + private @Nullable ReplyHeadersConfigurer replyHeadersConfigurer; private boolean splitIterables = true; @@ -159,7 +159,7 @@ public abstract class MessagingMessageListenerAdapter implements ConsumerS private ObservationRegistry observationRegistry = ObservationRegistry.NOOP; - private BiConsumer, RuntimeException> asyncRetryCallback; + private @Nullable BiConsumer, RuntimeException> asyncRetryCallback; /** * Create an instance with the provided bean and method. @@ -345,7 +345,7 @@ public abstract class MessagingMessageListenerAdapter 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 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 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 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 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 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 builder = MessageBuilder.withPayload(result) .setHeader(KafkaHeaders.TOPIC, topic); if (this.replyHeadersConfigurer != null) { @@ -701,10 +701,10 @@ public abstract class MessagingMessageListenerAdapter 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 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 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 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 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 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) { } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/package-info.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/package-info.java index 596cc1c3..ce88a3e5 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/package-info.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/package-info.java @@ -1,5 +1,5 @@ /** * Provides classes for adapting listeners. */ -@org.springframework.lang.NonNullApi +@org.jspecify.annotations.NullMarked package org.springframework.kafka.listener.adapter;