|
80 | 80 | import org.springframework.util.TypeUtils;
|
81 | 81 |
|
82 | 82 | /**
|
83 |
| - * An abstract {@link org.springframework.kafka.listener.MessageListener} adapter |
| 83 | + * An abstract {@link MessageListener} adapter |
84 | 84 | * providing the necessary infrastructure to extract the payload of a
|
85 |
| - * {@link org.springframework.messaging.Message}. |
| 85 | + * {@link Message}. |
86 | 86 | *
|
87 | 87 | * @param <K> the key type.
|
88 | 88 | * @param <V> the value type.
|
@@ -205,9 +205,9 @@ public void setMessageConverter(RecordMessageConverter messageConverter) {
|
205 | 205 |
|
206 | 206 | /**
|
207 | 207 | * Return the {@link MessagingMessageConverter} for this listener,
|
208 |
| - * being able to convert {@link org.springframework.messaging.Message}. |
| 208 | + * being able to convert {@link Message}. |
209 | 209 | * @return the {@link MessagingMessageConverter} for this listener,
|
210 |
| - * being able to convert {@link org.springframework.messaging.Message}. |
| 210 | + * being able to convert {@link Message}. |
211 | 211 | */
|
212 | 212 | protected final RecordMessageConverter getMessageConverter() {
|
213 | 213 | return this.messageConverter;
|
@@ -550,7 +550,9 @@ else if (!(result instanceof CompletableFuture<?>)) {
|
550 | 550 | try {
|
551 | 551 | if (t == null) {
|
552 | 552 | asyncSuccess(r, replyTopic, source, messageReturnType);
|
553 |
| - acknowledge(acknowledgment); |
| 553 | + if (isAsyncReplies()) { |
| 554 | + acknowledge(acknowledgment); |
| 555 | + } |
554 | 556 | }
|
555 | 557 | else {
|
556 | 558 | Throwable cause = t instanceof CompletionException ? t.getCause() : t;
|
|
0 commit comments