Revert KafkaBackkOffAwareMessageListenerAdapter.

spring-projects · Oct 11, 2024 · f2719ad · f2719ad
1 parent 645d7ce
commit f2719ad
Showing 1 changed file with 8 additions and 9 deletions.
diff --git a/...a/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapter.java b/...a/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapter.java
@@ -71,10 +71,10 @@ public class KafkaBackoffAwareMessageListenerAdapter<K, V>
 	 * @since 2.7
 	 */
 	public KafkaBackoffAwareMessageListenerAdapter(MessageListener<K, V> delegate,
-												KafkaConsumerBackoffManager kafkaConsumerBackoffManager,
-												String listenerId,
-												String backoffTimestampHeader,
-												Clock clock) {
+												   KafkaConsumerBackoffManager kafkaConsumerBackoffManager,
+												   String listenerId,
+												   String backoffTimestampHeader,
+												   Clock clock) {
 		super(delegate);
 		this.listenerId = listenerId;
 		this.kafkaConsumerBackoffManager = kafkaConsumerBackoffManager;
@@ -83,13 +83,13 @@ public KafkaBackoffAwareMessageListenerAdapter(MessageListener<K, V> delegate,
 	}
 
 	public KafkaBackoffAwareMessageListenerAdapter(MessageListener<K, V> adapter,
-			KafkaConsumerBackoffManager kafkaConsumerBackoffManager, String listenerId, Clock clock) throws KafkaBackoffException {
+												   KafkaConsumerBackoffManager kafkaConsumerBackoffManager, String listenerId, Clock clock) throws KafkaBackoffException {
 		this(adapter, kafkaConsumerBackoffManager, listenerId, RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP, clock);
 	}
 
 	@Override
 	public void onMessage(ConsumerRecord<K, V> consumerRecord, @Nullable Acknowledgment acknowledgment,
-								@Nullable Consumer<?, ?> consumer) {
+						  @Nullable Consumer<?, ?> consumer) {
 		maybeGetBackoffTimestamp(consumerRecord)
 				.ifPresent(nextExecutionTimestamp -> this.kafkaConsumerBackoffManager
 						.backOffIfNecessary(createContext(consumerRecord, nextExecutionTimestamp, consumer)));
@@ -111,10 +111,10 @@ private void invokeDelegateOnMessage(ConsumerRecord<K, V> consumerRecord, Acknow
 	}
 
 	private KafkaConsumerBackoffManager.Context createContext(ConsumerRecord<K, V> data, long nextExecutionTimestamp,
-			Consumer<?, ?> consumer) {
+															  Consumer<?, ?> consumer) {
 
 		return this.kafkaConsumerBackoffManager.createContext(nextExecutionTimestamp, this.listenerId,
-				new TopicPartition(data.topic(), data.partition()), consumer);
+															  new TopicPartition(data.topic(), data.partition()), consumer);
 	}
 
 	private Optional<Long> maybeGetBackoffTimestamp(ConsumerRecord<K, V> data) {
@@ -143,5 +143,4 @@ public void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment)
 	public void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer) {
 		onMessage(data, null, consumer);
 	}
-
 }