Skip to content

Commit 85aa161

Browse files
sobychackosobychacko
committed
More jspecify nullability changes in support package
#3762 Signed-off-by: Soby Chacko <[email protected]>
1 parent e5907af commit 85aa161

22 files changed

+72
-53
lines changed

‎spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -350,7 +350,7 @@ public void setThreadNameSupplier(Function<MessageListenerContainer, String> thr
350350
this.threadNameSupplier = threadNameSupplier;
351351
}
352352

353-
@SuppressWarnings("unchecked")
353+
@SuppressWarnings({"unchecked", "NullAway"})
354354
@Override
355355
public C createListenerContainer(KafkaListenerEndpoint endpoint) {
356356
C instance = createContainerInstance(endpoint);
@@ -372,6 +372,7 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
372372
return instance;
373373
}
374374

375+
@SuppressWarnings("NullAway") // Dataflow analysis limitation
375376
private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> aklEndpoint) {
376377
if (aklEndpoint.getRecordFilterStrategy() == null) {
377378
JavaUtils.INSTANCE
@@ -403,7 +404,7 @@ private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> aklEndpoint)
403404
* @param instance the container instance to configure.
404405
* @param endpoint the endpoint.
405406
*/
406-
@SuppressWarnings("deprecation")
407+
@SuppressWarnings({"deprecation", "NullAway"})
407408
protected void initializeContainer(C instance, KafkaListenerEndpoint endpoint) {
408409
ContainerProperties properties = instance.getContainerProperties();
409410
BeanUtils.copyProperties(this.containerProperties, properties, "topics", "topicPartitions", "topicPattern",

‎spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -525,7 +525,7 @@ public void setupListenerContainer(MessageListenerContainer listenerContainer,
525525
protected abstract MessagingMessageListenerAdapter<K, V> createMessageListener(MessageListenerContainer container,
526526
@Nullable MessageConverter messageConverter);
527527

528-
@SuppressWarnings("unchecked")
528+
@SuppressWarnings({"unchecked", "NullAway"})
529529
private void setupMessageListener(MessageListenerContainer container,
530530
@Nullable MessageConverter messageConverter) {
531531

‎spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -318,9 +318,9 @@ public MessageListenerContainer unregisterListenerContainer(String id) {
318318
* @param factory the {@link KafkaListenerContainerFactory} to use.
319319
* @return the {@link MessageListenerContainer}.
320320
*/
321+
@SuppressWarnings("NullAway") // Dataflow analysis limitation
321322
protected MessageListenerContainer createListenerContainer(KafkaListenerEndpoint endpoint,
322323
KafkaListenerContainerFactory<?> factory) {
323-
324324
if (endpoint instanceof MultiMethodKafkaListenerEndpoint<?, ?> mmkle) {
325325
Object bean = mmkle.getBean();
326326
if (bean instanceof EndpointHandlerMultiMethod ehmm) {

‎spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -131,8 +131,7 @@ public void setMessagingConverter(SmartMessageConverter messagingConverter) {
131131
this.messagingConverter = messagingConverter;
132132
}
133133

134-
@Nullable
135-
private String getReplyTopic() {
134+
private @Nullable String getReplyTopic() {
136135
Method replyingMethod = getMethod();
137136
if (replyingMethod != null) {
138137
SendTo ann = AnnotatedElementUtils.findMergedAnnotation(replyingMethod, SendTo.class);
@@ -171,6 +170,7 @@ private String getReplyTopic() {
171170
}
172171

173172
@Override
173+
@SuppressWarnings("NullAway") // Dataflow analysis limitation
174174
protected MessagingMessageListenerAdapter<K, V> createMessageListener(MessageListenerContainer container,
175175
@Nullable MessageConverter messageConverter) {
176176

‎spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -251,6 +251,7 @@ protected Object headerValueToAddOut(String key, Object value) {
251251
return valueToAdd;
252252
}
253253

254+
@SuppressWarnings("NullAway") // Dataflow analysis limitation
254255
@Nullable
255256
private byte[] mapRawOut(String header, Object value) {
256257
if (this.mapAllStringsOut || this.rawMappedHeaders.containsKey(header)) {
@@ -269,7 +270,7 @@ else if (value instanceof String) {
269270
* @param header the header.
270271
* @return the value to add.
271272
*/
272-
protected Object headerValueToAddIn(Header header) {
273+
protected @Nullable Object headerValueToAddIn(@Nullable Header header) {
273274
if (header == null || header.value() == null) {
274275
return null;
275276
}

‎spring-kafka/src/main/java/org/springframework/kafka/support/CompositeProducerListener.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2018-2020 the original author or authors.
2+
* Copyright 2018-2025 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -22,6 +22,7 @@
2222

2323
import org.apache.kafka.clients.producer.ProducerRecord;
2424
import org.apache.kafka.clients.producer.RecordMetadata;
25+
import org.jspecify.annotations.Nullable;
2526

2627
import org.springframework.util.Assert;
2728

@@ -74,7 +75,7 @@ public void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata record
7475
}
7576

7677
@Override
77-
public void onError(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata, Exception exception) {
78+
public void onError(ProducerRecord<K, V> producerRecord, @Nullable RecordMetadata recordMetadata, Exception exception) {
7879
this.delegates.forEach(d -> d.onError(producerRecord, recordMetadata, exception));
7980
}
8081

‎spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2017-2024 the original author or authors.
2+
* Copyright 2017-2025 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -430,6 +430,7 @@ public static class NonTrustedHeaderType {
430430

431431
private String untrustedType;
432432

433+
@SuppressWarnings("NullAway.Init")
433434
public NonTrustedHeaderType() {
434435
}
435436

‎spring-kafka/src/main/java/org/springframework/kafka/support/EndpointHandlerMethod.java

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2021-2024 the original author or authors.
2+
* Copyright 2021-2025 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -19,6 +19,8 @@
1919
import java.lang.reflect.Method;
2020
import java.util.Arrays;
2121

22+
import org.jspecify.annotations.Nullable;
23+
2224
import org.springframework.beans.factory.BeanCurrentlyInCreationException;
2325
import org.springframework.beans.factory.BeanFactory;
2426
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
@@ -41,11 +43,11 @@ public class EndpointHandlerMethod {
4143

4244
private final Object beanOrClass;
4345

44-
private String methodName;
46+
private @Nullable String methodName;
4547

46-
private Object bean;
48+
private @Nullable Object bean;
4749

48-
private Method method;
50+
private @Nullable Method method;
4951

5052
public EndpointHandlerMethod(Object beanOrClass, String methodName) {
5153
Assert.notNull(beanOrClass, () -> "No destination bean or class provided!");

‎spring-kafka/src/main/java/org/springframework/kafka/support/JacksonPresent.java

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2017-2019 the original author or authors.
2+
* Copyright 2017-2025 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
1616

1717
package org.springframework.kafka.support;
1818

19+
import org.jspecify.annotations.Nullable;
20+
1921
import org.springframework.util.ClassUtils;
2022

2123
/**
@@ -28,7 +30,7 @@
2830
*/
2931
public final class JacksonPresent {
3032

31-
private static final ClassLoader classLoader = ClassUtils.getDefaultClassLoader(); // NOSONAR
33+
private static final @Nullable ClassLoader classLoader = ClassUtils.getDefaultClassLoader(); // NOSONAR
3234

3335
private static final boolean jackson2Present = // NOSONAR
3436
ClassUtils.isPresent("com.fasterxml.jackson.databind.ObjectMapper", classLoader) &&

‎spring-kafka/src/main/java/org/springframework/kafka/support/JavaUtils.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ public <T> JavaUtils acceptIfCondition(boolean condition, T value, Consumer<T> c
6868
* @param <T> the value type.
6969
* @return this.
7070
*/
71-
public <T> JavaUtils acceptIfNotNull(@Nullable T value, Consumer<T> consumer) {
71+
public <T> JavaUtils acceptIfNotNull(@Nullable T value, Consumer<@Nullable T> consumer) {
7272
if (value != null) {
7373
consumer.accept(value);
7474
}
@@ -162,7 +162,7 @@ public <T1, T2> JavaUtils acceptIfCondition(boolean condition, T1 t1, T2 t2, BiC
162162
* @param <T2> the second argument type.
163163
* @return this.
164164
*/
165-
public <T1, T2> JavaUtils acceptIfNotNull(T1 t1, T2 t2, BiConsumer<T1, T2> consumer) {
165+
public <T1, T2> JavaUtils acceptIfNotNull(T1 t1, @Nullable T2 t2, BiConsumer<T1, @Nullable T2> consumer) {
166166
if (t2 != null) {
167167
consumer.accept(t1, t2);
168168
}

‎spring-kafka/src/main/java/org/springframework/kafka/support/KafkaMessageHeaderAccessor.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,8 @@ public int getBlockingRetryDeliveryAttempt() {
5353
Assert.state(getHeader(KafkaHeaders.DELIVERY_ATTEMPT) != null,
5454
"Blocking delivery attempt header not present, "
5555
+ "see ContainerProperties.setDeliveryAttemptHeader() to enable");
56-
return getHeader(KafkaHeaders.DELIVERY_ATTEMPT, Integer.class);
56+
Integer deliveryAttempts = getHeader(KafkaHeaders.DELIVERY_ATTEMPT, Integer.class);
57+
return deliveryAttempts == null ? 0 : deliveryAttempts;
5758
}
5859

5960
/**

‎spring-kafka/src/main/java/org/springframework/kafka/support/KafkaStreamBrancher.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019-2021 the original author or authors.
2+
* Copyright 2019-2025 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@
2323

2424
import org.apache.kafka.streams.kstream.KStream;
2525
import org.apache.kafka.streams.kstream.Predicate;
26+
import org.jspecify.annotations.Nullable;
2627

2728
/**
2829
* Provides a method-chaining way to build {@link org.apache.kafka.streams.kstream.KStream#branch branches} in
@@ -54,7 +55,7 @@ public final class KafkaStreamBrancher<K, V> {
5455

5556
private final List<Consumer<? super KStream<K, V>>> consumerList = new ArrayList<>();
5657

57-
private Consumer<? super KStream<K, V>> defaultConsumer;
58+
private @Nullable Consumer<? super KStream<K, V>> defaultConsumer;
5859

5960
/**
6061
* Defines a new branch.

‎spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2018-2024 the original author or authors.
2+
* Copyright 2018-2025 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -28,6 +28,7 @@
2828
import org.apache.kafka.clients.consumer.ConsumerRecord;
2929
import org.apache.kafka.clients.producer.ProducerConfig;
3030
import org.apache.kafka.clients.producer.ProducerRecord;
31+
import org.jspecify.annotations.Nullable;
3132

3233
import org.springframework.messaging.Message;
3334
import org.springframework.util.Assert;
@@ -101,7 +102,7 @@ public static boolean returnTypeMessageOrCollectionOf(Method method) {
101102
* @param groupId the group id.
102103
* @since 2.3
103104
*/
104-
public static void setConsumerGroupId(String groupId) {
105+
public static void setConsumerGroupId(@Nullable String groupId) {
105106
if (groupId != null) {
106107
KafkaUtils.GROUP_IDS.put(Thread.currentThread(), groupId);
107108
}
@@ -112,7 +113,7 @@ public static void setConsumerGroupId(String groupId) {
112113
* @return the group id.
113114
* @since 2.3
114115
*/
115-
public static String getConsumerGroupId() {
116+
public static @Nullable String getConsumerGroupId() {
116117
return KafkaUtils.GROUP_IDS.get(Thread.currentThread());
117118
}
118119

@@ -150,9 +151,10 @@ else if (dt instanceof String str) {
150151
catch (@SuppressWarnings("unused") NumberFormatException ex) {
151152
}
152153
}
154+
Integer deliveryTimeoutInMs = (Integer) ProducerConfig.configDef().defaultValues()
155+
.get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG);
153156
return Duration.ofMillis(Math.max(
154-
((Integer) ProducerConfig.configDef().defaultValues()
155-
.get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG)).longValue() + buffer,
157+
deliveryTimeoutInMs == null ? 0 : deliveryTimeoutInMs.longValue() + buffer,
156158
min));
157159
}
158160

‎spring-kafka/src/main/java/org/springframework/kafka/support/LogIfLevelEnabled.java

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2018-2023 the original author or authors.
2+
* Copyright 2018-2025 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -18,6 +18,8 @@
1818

1919
import java.util.function.Supplier;
2020

21+
import org.jspecify.annotations.Nullable;
22+
2123
import org.springframework.core.log.LogAccessor;
2224
import org.springframework.util.Assert;
2325

@@ -101,7 +103,7 @@ public void log(Supplier<CharSequence> messageSupplier, Throwable thrown) {
101103
}
102104
}
103105

104-
private void fatal(Supplier<CharSequence> messageSupplier, Throwable thrown) {
106+
private void fatal(Supplier<CharSequence> messageSupplier, @Nullable Throwable thrown) {
105107
if (thrown != null) {
106108
this.logger.fatal(thrown, messageSupplier);
107109
}
@@ -110,7 +112,7 @@ private void fatal(Supplier<CharSequence> messageSupplier, Throwable thrown) {
110112
}
111113
}
112114

113-
private void error(Supplier<CharSequence> messageSupplier, Throwable thrown) {
115+
private void error(Supplier<CharSequence> messageSupplier, @Nullable Throwable thrown) {
114116
if (thrown != null) {
115117
this.logger.error(thrown, messageSupplier);
116118
}
@@ -119,7 +121,7 @@ private void error(Supplier<CharSequence> messageSupplier, Throwable thrown) {
119121
}
120122
}
121123

122-
private void warn(Supplier<CharSequence> messageSupplier, Throwable thrown) {
124+
private void warn(Supplier<CharSequence> messageSupplier, @Nullable Throwable thrown) {
123125
if (thrown != null) {
124126
this.logger.warn(thrown, messageSupplier);
125127
}
@@ -128,7 +130,7 @@ private void warn(Supplier<CharSequence> messageSupplier, Throwable thrown) {
128130
}
129131
}
130132

131-
private void info(Supplier<CharSequence> messageSupplier, Throwable thrown) {
133+
private void info(Supplier<CharSequence> messageSupplier, @Nullable Throwable thrown) {
132134
if (thrown != null) {
133135
this.logger.info(thrown, messageSupplier);
134136
}
@@ -137,7 +139,7 @@ private void info(Supplier<CharSequence> messageSupplier, Throwable thrown) {
137139
}
138140
}
139141

140-
private void debug(Supplier<CharSequence> messageSupplier, Throwable thrown) {
142+
private void debug(Supplier<CharSequence> messageSupplier, @Nullable Throwable thrown) {
141143
if (thrown != null) {
142144
this.logger.debug(thrown, messageSupplier);
143145
}
@@ -146,7 +148,7 @@ private void debug(Supplier<CharSequence> messageSupplier, Throwable thrown) {
146148
}
147149
}
148150

149-
private void trace(Supplier<CharSequence> messageSupplier, Throwable thrown) {
151+
private void trace(Supplier<CharSequence> messageSupplier, @Nullable Throwable thrown) {
150152
if (thrown != null) {
151153
this.logger.trace(thrown, messageSupplier);
152154
}

0 commit comments

Comments
 (0)