Skip to content

Commit e6affce

Browse files
committed
Fix NPE in MessagingMessageListenerAdapter
Related to #1189 * Some other code clean up including deprecation warning
1 parent c4e9d32 commit e6affce

File tree

4 files changed

+17
-14
lines changed

4 files changed

+17
-14
lines changed

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

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
115115

116116
private final KafkaListenerErrorHandler errorHandler;
117117

118+
@Nullable
118119
private HandlerAdapter handlerMethod;
119120

120121
private boolean isConsumerRecordList;
@@ -216,7 +217,7 @@ public void setMessagingConverter(SmartMessageConverter messageConverter) {
216217
}
217218

218219
/**
219-
* Returns the inferred type for conversion or, if null, the
220+
* Return the inferred type for conversion or, if null, the
220221
* {@link #setFallbackType(Class) fallbackType}.
221222
* @return the type.
222223
*/
@@ -245,7 +246,7 @@ public void setHandlerMethod(HandlerAdapter handlerMethod) {
245246
}
246247

247248
public boolean isAsyncReplies() {
248-
return this.handlerMethod.isAsyncReplies();
249+
return this.handlerMethod != null && this.handlerMethod.isAsyncReplies();
249250
}
250251

251252
protected boolean isConsumerRecordList() {
@@ -262,7 +263,7 @@ public boolean isConversionNeeded() {
262263

263264
/**
264265
* Set the topic to which to send any result from the method invocation.
265-
* May be a SpEL expression {@code !{...}} evaluated at runtime.
266+
* Maybe a SpEL expression {@code !{...}} evaluated at runtime.
266267
* @param replyTopicParam the topic or expression.
267268
* @since 2.0
268269
*/
@@ -334,7 +335,7 @@ protected boolean isSplitIterables() {
334335
}
335336

336337
/**
337-
* Set to false to disable splitting {@link Iterable} reply values into separate
338+
* Set to {@code false} to disable splitting {@link Iterable} reply values into separate
338339
* records.
339340
* @param splitIterables false to disable; default true.
340341
* @since 2.3.5
@@ -406,6 +407,7 @@ protected final Object invokeHandler(Object data, @Nullable Acknowledgment ackno
406407
if (ack == null && this.noOpAck) {
407408
ack = NO_OP_ACK;
408409
}
410+
Assert.notNull(this.handlerMethod, "the 'handlerMethod' must not be null");
409411
try {
410412
if (data instanceof List && !this.isConsumerRecordList) {
411413
return this.handlerMethod.invoke(message, ack, consumer);
@@ -545,7 +547,9 @@ private String evaluateTopic(Object request, Object source, Object result, @Null
545547
* @since 2.1.3
546548
*/
547549
@SuppressWarnings("unchecked")
548-
protected void sendResponse(Object result, String topic, @Nullable Object source, boolean returnTypeMessage) {
550+
protected void sendResponse(Object result, @Nullable String topic, @Nullable Object source,
551+
boolean returnTypeMessage) {
552+
549553
if (!returnTypeMessage && topic == null) {
550554
this.logger.debug(() -> "No replyTopic to handle the reply: " + result);
551555
}
@@ -622,7 +626,7 @@ private void sendReplyForMessageSource(Object result, String topic, Message<?> s
622626
})
623627
.filter(e -> this.replyHeadersConfigurer.shouldCopy(e.getKey(), e.getValue()))
624628
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
625-
if (headersToCopy.size() > 0) {
629+
if (!headersToCopy.isEmpty()) {
626630
builder.copyHeaders(headersToCopy);
627631
}
628632
headersToCopy = this.replyHeadersConfigurer.additionalHeaders();
@@ -637,7 +641,9 @@ private void sendReplyForMessageSource(Object result, String topic, Message<?> s
637641
this.replyTemplate.send(builder.build());
638642
}
639643

640-
protected void asyncSuccess(@Nullable Object result, String replyTopic, Message<?> source, boolean returnTypeMessage) {
644+
protected void asyncSuccess(@Nullable Object result, String replyTopic, Message<?> source,
645+
boolean returnTypeMessage) {
646+
641647
if (result == null) {
642648
if (this.logger.isDebugEnabled()) {
643649
this.logger.debug("Async result is null, ignoring");

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicResolver.java

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -153,11 +153,6 @@ public DestinationTopic getDestinationTopicByName(String mainListenerId, String
153153
() -> "No DestinationTopic found for " + mainListenerId + ":" + topic).getSourceDestination();
154154
}
155155

156-
@Override
157-
public DestinationTopic getDltFor(String mainListenerId, String topicName) {
158-
return getDltFor(mainListenerId, topicName, null);
159-
}
160-
161156
@Nullable
162157
@Override
163158
public DestinationTopic getDltFor(String mainListenerId, String topicName, Exception e) {

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicContainer.java

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -71,7 +71,9 @@ public interface DestinationTopicContainer {
7171
*/
7272
@Nullable
7373
@Deprecated(since = "3.2", forRemoval = true)
74-
DestinationTopic getDltFor(String mainListenerId, String topicName);
74+
default DestinationTopic getDltFor(String mainListenerId, String topicName) {
75+
return getDltFor(mainListenerId, topicName, null);
76+
}
7577

7678
/**
7779
* Returns the {@link DestinationTopic} instance registered as

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicResolverTests.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -222,7 +222,7 @@ void shouldGetNextDestinationTopic() {
222222
@Test
223223
void shouldGetGeneralPurposeDltWhenExceptionIsNotKnown() {
224224
assertThat(defaultDestinationTopicContainer
225-
.getDltFor("id", mainDestinationTopic.getDestinationName()))
225+
.getDltFor("id", mainDestinationTopic.getDestinationName(), null))
226226
.isEqualTo(dltDestinationTopic);
227227
}
228228

0 commit comments

Comments
 (0)