From adb89706dec00305ee24e620c2ca57126661a059 Mon Sep 17 00:00:00 2001 From: Kurt Hong Date: Fri, 3 Nov 2023 00:01:14 +0900 Subject: [PATCH] Remove `RetryListener` from KafkaInbounds * KafkaMessageDrivenChannelAdapter's ATTRIBUTES_HOLDER should be isolated. In order to achieve custom retry in batch mode, we may to use a RetryTemplate in listener itself. But if the RetryTemplate is shared with another KafkaMessageDrivenChannelAdapter, batch mode's ATTRIBUTES_HOLDER might be over-written by another KafkaMessageDrivenChannelAdapter's IntegrationRecordMessageListener. The situation is like shown below. - There is only one RetryTemple bean in the application. - There are two KafkaMessageDrivenChannelAdapters(A,B) in the application. - A KafkaMessageDrivenChannelAdapter is batch mode and utilizing the retryTemplate in the listener. - B KafkaMessageDrivenChannelAdapter is record mode and using the retryTemplate itself. - (B KafkaMessageDrivenChannelAdapter's recordListener is registered in the retryTemplate.) - When A retry is attempted in the listener, it will trigger B KafkaMessageDrivenChannelAdapter's recordListener. - B KafkaMessageDrivenChannelAdapter's recordListener will overwrite A KafkaMessageDrivenChannelAdapter's ATTRIBUTES_HOLDER. * should not mutate an externally provided RetryTemplate * should not expose KafkaInboundEndpoint's methods outside the class and fix a checkstyle error. * removing unused code and polishing * restore retry around batch and bring back KafkaInboundEndpoint to endpoints. * remove retry logic in batch mode and move ATTRIBUTES_HOLDER into KafkaInboundEndpoint * remove generic type parameters of KafkaInboundEndpoint * fix style error --------- Co-authored-by: kurt --- .../kafka/inbound/KafkaInboundEndpoint.java | 20 +++-- .../kafka/inbound/KafkaInboundGateway.java | 38 ++-------- .../KafkaMessageDrivenChannelAdapter.java | 76 ++----------------- 3 files changed, 25 insertions(+), 109 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundEndpoint.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundEndpoint.java index 1b5f74f6fc..3f2090322e 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundEndpoint.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-2023 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. @@ -17,7 +17,9 @@ package org.springframework.integration.kafka.inbound; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.springframework.core.AttributeAccessor; import org.springframework.kafka.KafkaException; import org.springframework.kafka.support.Acknowledgment; import org.springframework.retry.RecoveryCallback; @@ -50,6 +52,8 @@ public interface KafkaInboundEndpoint { */ String CONTEXT_RECORD = "record"; + ThreadLocal ATTRIBUTES_HOLDER = new ThreadLocal<>(); + /** * Execute the runnable with the retry template and recovery callback. * @param template the template. @@ -59,14 +63,17 @@ public interface KafkaInboundEndpoint { * @param consumer the consumer. * @param runnable the runnable. */ - default void doWithRetry(RetryTemplate template, RecoveryCallback callback, Object data, + default void doWithRetry(RetryTemplate template, RecoveryCallback callback, ConsumerRecord record, Acknowledgment acknowledgment, Consumer consumer, Runnable runnable) { try { template.execute(context -> { - context.setAttribute(CONTEXT_RECORD, data); - context.setAttribute(CONTEXT_ACKNOWLEDGMENT, acknowledgment); - context.setAttribute(CONTEXT_CONSUMER, consumer); + if (context.getRetryCount() == 0) { + context.setAttribute(CONTEXT_RECORD, record); + context.setAttribute(CONTEXT_ACKNOWLEDGMENT, acknowledgment); + context.setAttribute(CONTEXT_CONSUMER, consumer); + ATTRIBUTES_HOLDER.set(context); + } runnable.run(); return null; }, callback); @@ -74,6 +81,9 @@ public interface KafkaInboundEndpoint { catch (Exception ex) { throw new KafkaException("Failed to execute runnable", ex); } + finally { + ATTRIBUTES_HOLDER.remove(); + } } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java index 0f37ccf8f1..7eebada480 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2022 the original author or authors. + * Copyright 2018-2023 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. @@ -55,9 +55,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.retry.RecoveryCallback; -import org.springframework.retry.RetryCallback; import org.springframework.retry.RetryContext; -import org.springframework.retry.RetryListener; import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; @@ -78,8 +76,6 @@ import org.springframework.util.Assert; public class KafkaInboundGateway extends MessagingGatewaySupport implements KafkaInboundEndpoint, Pausable, OrderlyShutdownCapable { - private static final ThreadLocal ATTRIBUTES_HOLDER = new ThreadLocal<>(); - private final IntegrationRecordMessageListener listener = new IntegrationRecordMessageListener(); private final AbstractMessageListenerContainer messageListenerContainer; @@ -195,7 +191,6 @@ public class KafkaInboundGateway extends MessagingGatewaySupport protected void onInit() { super.onInit(); if (this.retryTemplate != null) { - this.retryTemplate.registerListener(this.listener); MessageChannel errorChannel = getErrorChannel(); if (this.recoveryCallback != null && errorChannel != null) { this.recoveryCallback = new ErrorMessageSendingRecoverer(errorChannel, getErrorMessageStrategy()); @@ -261,7 +256,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport private void setAttributesIfNecessary(Object record, @Nullable Message message, boolean conversionError) { boolean needHolder = ATTRIBUTES_HOLDER.get() == null && (getErrorChannel() != null && (this.retryTemplate == null || conversionError)); - boolean needAttributes = needHolder | this.retryTemplate != null; + boolean needAttributes = needHolder || this.retryTemplate != null; if (needHolder) { ATTRIBUTES_HOLDER.set(ErrorMessageUtils.getAttributeAccessor(null, null)); } @@ -285,8 +280,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport } } - private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter - implements RetryListener { + private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter { IntegrationRecordMessageListener() { super(null, null); // NOSONAR - out of use @@ -332,9 +326,8 @@ public class KafkaInboundGateway extends MessagingGatewaySupport RetryTemplate template = KafkaInboundGateway.this.retryTemplate; if (template != null) { doWithRetry(template, KafkaInboundGateway.this.recoveryCallback, record, acknowledgment, consumer, - () -> { - doSendAndReceive(enhanceHeadersAndSaveAttributes(message, record)); - }); + () -> doSendAndReceive(enhanceHeadersAndSaveAttributes(message, record)) + ); } else { doSendAndReceive(enhanceHeadersAndSaveAttributes(message, record)); @@ -427,27 +420,6 @@ public class KafkaInboundGateway extends MessagingGatewaySupport return reply; } - @Override - public boolean open(RetryContext context, RetryCallback callback) { - if (KafkaInboundGateway.this.retryTemplate != null) { - ATTRIBUTES_HOLDER.set(context); - } - return true; - } - - @Override - public void close(RetryContext context, RetryCallback callback, - Throwable throwable) { - - ATTRIBUTES_HOLDER.remove(); - } - - @Override - public void onError(RetryContext context, RetryCallback callback, - Throwable throwable) { - // Empty - } - } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index dd6b351ccd..0324b4fab2 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -63,9 +63,7 @@ import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.retry.RecoveryCallback; -import org.springframework.retry.RetryCallback; import org.springframework.retry.RetryContext; -import org.springframework.retry.RetryListener; import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; @@ -85,8 +83,6 @@ import org.springframework.util.Assert; public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport implements KafkaInboundEndpoint, OrderlyShutdownCapable, Pausable { - private static final ThreadLocal ATTRIBUTES_HOLDER = new ThreadLocal<>(); - private final AbstractMessageListenerContainer messageListenerContainer; private final IntegrationRecordMessageListener recordListener = new IntegrationRecordMessageListener(); @@ -298,7 +294,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo if (this.recoveryCallback != null && errorChannel != null) { this.recoveryCallback = new ErrorMessageSendingRecoverer(errorChannel, getErrorMessageStrategy()); } - this.retryTemplate.registerListener(this.recordListener); } if (!doFilterInRetry && this.recordFilterStrategy != null) { listener = new FilteringMessageListenerAdapter<>(listener, this.recordFilterStrategy, @@ -366,7 +361,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo private void setAttributesIfNecessary(Object record, @Nullable Message message, boolean conversionError) { boolean needHolder = ATTRIBUTES_HOLDER.get() == null && (getErrorChannel() != null && (this.retryTemplate == null || conversionError)); - boolean needAttributes = needHolder | this.retryTemplate != null; + boolean needAttributes = needHolder || this.retryTemplate != null; if (needHolder) { ATTRIBUTES_HOLDER.set(ErrorMessageUtils.getAttributeAccessor(null, null)); } @@ -424,8 +419,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo batch } - private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter - implements RetryListener { + private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter { IntegrationRecordMessageListener() { super(null, null); // NOSONAR - out of use @@ -511,34 +505,12 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo return messageToReturn; } - @Override - public boolean open(RetryContext context, RetryCallback callback) { - if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) { - ATTRIBUTES_HOLDER.set(context); - } - return true; - } - - @Override - public void close(RetryContext context, RetryCallback callback, - Throwable throwable) { - - ATTRIBUTES_HOLDER.remove(); - } - - @Override - public void onError(RetryContext context, RetryCallback callback, - Throwable throwable) { - // Empty - } - } - private class IntegrationBatchMessageListener extends BatchMessagingMessageListenerAdapter - implements RetryListener { + private class IntegrationBatchMessageListener extends BatchMessagingMessageListenerAdapter { IntegrationBatchMessageListener() { - super(null, null); // NOSONAR - out if use + super(null, null); // NOSONAR - out of use } @Override @@ -557,33 +529,10 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo message = toMessage(records, acknowledgment, consumer); } if (message != null) { - RetryTemplate template = KafkaMessageDrivenChannelAdapter.this.retryTemplate; - if (template != null) { - doWIthRetry(records, acknowledgment, consumer, message, template); - } - else { - sendMessageIfAny(message, records); - } + sendMessageIfAny(message, records); } } - private void doWIthRetry(List> records, Acknowledgment acknowledgment, - Consumer consumer, Message message, RetryTemplate template) { - - doWithRetry(template, KafkaMessageDrivenChannelAdapter.this.recoveryCallback, records, acknowledgment, - consumer, () -> { - if (KafkaMessageDrivenChannelAdapter.this.filterInRetry) { - List> filtered = - KafkaMessageDrivenChannelAdapter.this.recordFilterStrategy.filterBatch(records); - Message toSend = message; - if (filtered.size() != records.size()) { - toSend = toMessage(filtered, acknowledgment, consumer); - } - sendMessageIfAny(toSend, filtered); - } - }); - } - @Nullable private Message toMessage(List> records, Acknowledgment acknowledgment, Consumer consumer) { @@ -607,21 +556,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo return message; } - @Override - public boolean open(RetryContext context, RetryCallback callback) { - if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) { - ATTRIBUTES_HOLDER.set(context); - } - return true; - } - - @Override - public void close(RetryContext context, RetryCallback callback, - Throwable throwable) { - - ATTRIBUTES_HOLDER.remove(); - } - } }