Add Missing Pause/Resume Events

Resolves #2387

- `FallbackBatchErrorHandler`
- `Acknowledgment.nack()`

**2.9.x only - I will back port/forward port**

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerPausedEvent.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2022 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.
@@ -33,6 +33,8 @@ public class ConsumerPausedEvent extends KafkaEvent {
 
 	private final Collection<TopicPartition> partitions;
 
+	private final String reason;
+
 	/**
 	 * Construct an instance with the provided source and partitions.
 	 * @param source the container instance that generated the event.
@@ -43,6 +45,23 @@ public class ConsumerPausedEvent extends KafkaEvent {
 	public ConsumerPausedEvent(Object source, Object container, Collection<TopicPartition> partitions) {
 		super(source, container);
 		this.partitions = partitions;
+		this.reason = null;
+	}
+
+	/**
+	 * Construct an instance with the provided source and partitions.
+	 * @param source the container instance that generated the event.
+	 * @param container the container or the parent container if the container is a child.
+	 * @param partitions the partitions.
+	 * @param reason the reason for the pause.
+	 * @since 2.8.9
+	 */
+	public ConsumerPausedEvent(Object source, Object container, Collection<TopicPartition> partitions,
+			String reason) {
+
+		super(source, container);
+		this.partitions = partitions;
+		this.reason = reason;
 	}
 
 	/**
@@ -55,7 +74,7 @@ public Collection<TopicPartition> getPartitions() {
 
 	@Override
 	public String toString() {
-		return "ConsumerPausedEvent [partitions=" + this.partitions + "]";
+		return "ConsumerPausedEvent [reason=" + this.reason + ", partitions=" + this.partitions + "]";
 	}
 
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/CommonErrorHandler.java

@@ -157,8 +157,24 @@ default void setAckAfterHandle(boolean ack) {
 	 * @param consumer the consumer.
 	 * @param partitions the newly assigned partitions.
 	 * @since 2.8.8
+	 * @deprecated in favor of {@link #onPartitionsAssigned(Consumer, Collection, Runnable)}.
 	 */
+	@Deprecated
 	default void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
 	}
 
+	/**
+	 * Called when partitions are assigned.
+	 * @param consumer the consumer.
+	 * @param partitions the newly assigned partitions.
+	 * @param publishPause called to publish a consumer paused event.
+	 * @since 2.8.9
+	 */
+	@SuppressWarnings("deprecation")
+	default void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions,
+			Runnable publishPause) {
+
+		onPartitionsAssigned(consumer, partitions);
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java

@@ -160,8 +160,10 @@ public void handleOtherException(Exception thrownException, Consumer<?, ?> consu
 	}
 
 	@Override
-	public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
-		getFallbackBatchHandler().onPartitionsAssigned(consumer, partitions);
+	public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions,
+			Runnable publishPause) {
+
+		getFallbackBatchHandler().onPartitionsAssigned(consumer, partitions, publishPause);
 	}
 
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlerAdapter.java

@@ -161,9 +161,12 @@ public void handleBatch(Exception thrownException, ConsumerRecords<?, ?> data, C
 
 	@SuppressWarnings("deprecation")
 	@Override
-	public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
+	public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions,
+			Runnable publishPause) {
+
 		if (this.batchErrorHandler instanceof RetryingBatchErrorHandler) {
-			((RetryingBatchErrorHandler) this.batchErrorHandler).onPartitionsAssigned(consumer, partitions);
+			((RetryingBatchErrorHandler) this.batchErrorHandler).onPartitionsAssigned(consumer, partitions,
+					publishPause);
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java

@@ -17,10 +17,12 @@
 package org.springframework.kafka.listener;
 
 import java.time.Duration;
+import java.util.Set;
 import java.util.function.BiConsumer;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
@@ -62,7 +64,11 @@ public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> r
 		BackOffExecution execution = backOff.start();
 		long nextBackOff = execution.nextBackOff();
 		String failed = null;
-		consumer.pause(consumer.assignment());
+		Set<TopicPartition> assignment = consumer.assignment();
+		consumer.pause(assignment);
+		if (container instanceof KafkaMessageListenerContainer) {
+			((KafkaMessageListenerContainer<?, ?>) container).publishConsumerPausedEvent(assignment, "For batch retry");
+		}
 		try {
 			while (nextBackOff != BackOffExecution.STOP) {
 				consumer.poll(Duration.ZERO);
@@ -99,7 +105,11 @@ public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> r
 			}
 		}
 		finally {
-			consumer.resume(consumer.assignment());
+			Set<TopicPartition> assignment2 = consumer.assignment();
+			consumer.resume(assignment2);
+			if (container instanceof KafkaMessageListenerContainer) {
+				((KafkaMessageListenerContainer<?, ?>) container).publishConsumerResumedEvent(assignment2);
+			}
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -442,15 +442,15 @@ private void publishNonResponsiveConsumerEvent(long timeSinceLastPoll, Consumer<
 		}
 	}
 
-	private void publishConsumerPausedEvent(Collection<TopicPartition> partitions) {
+	void publishConsumerPausedEvent(Collection<TopicPartition> partitions, String reason) {
 		ApplicationEventPublisher publisher = getApplicationEventPublisher();
 		if (publisher != null) {
 			publisher.publishEvent(new ConsumerPausedEvent(this, this.thisOrParentContainer,
-					Collections.unmodifiableCollection(partitions)));
+					Collections.unmodifiableCollection(partitions), reason));
 		}
 	}
 
-	private void publishConsumerResumedEvent(Collection<TopicPartition> partitions) {
+	void publishConsumerResumedEvent(Collection<TopicPartition> partitions) {
 		ApplicationEventPublisher publisher = getApplicationEventPublisher();
 		if (publisher != null) {
 			publisher.publishEvent(new ConsumerResumedEvent(this, this.thisOrParentContainer,
@@ -1633,7 +1633,9 @@ private void doPauseConsumerIfNecessary() {
 					this.consumer.pause(assigned);
 					this.consumerPaused = true;
 					this.logger.debug(() -> "Paused consumption from: " + this.consumer.paused());
-					publishConsumerPausedEvent(assigned);
+					publishConsumerPausedEvent(assigned, this.pausedForAsyncAcks
+							? "Incomplete out of order acks"
+							: "User requested");
 				}
 			}
 		}
@@ -1644,6 +1646,7 @@ private void resumeConsumerIfNeccessary() {
 					this.nackWakeTimeMillis = 0;
 					this.consumer.resume(this.pausedForNack);
 					this.logger.debug(() -> "Resumed after nack sleep: " + this.pausedForNack);
+					publishConsumerResumedEvent(this.pausedForNack);
 					this.pausedForNack.clear();
 				}
 			}
@@ -2539,6 +2542,7 @@ private void pauseForNackSleep() {
 				this.logger.debug(() -> "Pausing for nack sleep: " + ListenerConsumer.this.pausedForNack);
 				try {
 					this.consumer.pause(this.pausedForNack);
+					publishConsumerPausedEvent(this.pausedForNack, "Nack with sleep time received");
 				}
 				catch (IllegalStateException ex) {
 					// this should never happen; defensive, just in case...
@@ -3375,7 +3379,8 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 				}
 				if (ListenerConsumer.this.commonErrorHandler != null) {
 					ListenerConsumer.this.commonErrorHandler.onPartitionsAssigned(ListenerConsumer.this.consumer,
-							partitions);
+							partitions, () -> publishConsumerPausedEvent(partitions,
+									"Paused by error handler after rebalance"));
 				}
 			}
 
@@ -3386,7 +3391,7 @@ private void repauseIfNeeded(Collection<TopicPartition> partitions) {
 					ListenerConsumer.this.logger.warn("Paused consumer resumed by Kafka due to rebalance; "
 							+ "consumer paused again, so the initial poll() will never return any records");
 					ListenerConsumer.this.logger.debug(() -> "Paused consumption from: " + partitions);
-					publishConsumerPausedEvent(partitions);
+					publishConsumerPausedEvent(partitions, "Re-paused after rebalance");
 				}
 				Collection<TopicPartition> toRepause = new LinkedList<>();
 				partitions.forEach(tp -> {
@@ -3397,7 +3402,7 @@ private void repauseIfNeeded(Collection<TopicPartition> partitions) {
 				if (!ListenerConsumer.this.consumerPaused && toRepause.size() > 0) {
 					ListenerConsumer.this.consumer.pause(toRepause);
 					ListenerConsumer.this.logger.debug(() -> "Paused consumption from: " + toRepause);
-					publishConsumerPausedEvent(toRepause);
+					publishConsumerPausedEvent(toRepause, "Re-paused after rebalance");
 				}
 				this.revoked.removeAll(toRepause);
 				ListenerConsumer.this.pausedPartitions.removeAll(this.revoked);

spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java

@@ -114,10 +114,20 @@ public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> re
 		}
 	}
 
+	@Deprecated
 	public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
 		if (this.retrying.get()) {
 			consumer.pause(consumer.assignment());
 		}
 	}
 
+	public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions,
+			Runnable publishPause) {
+
+		if (this.retrying.get()) {
+			consumer.pause(consumer.assignment());
+			publishPause.run();
+		}
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/FallbackBatchErrorHandlerTests.java

@@ -189,25 +189,31 @@ void rePauseOnRebalance() {
 				Collections.singletonList(new ConsumerRecord<>("foo", 1, 0L, "foo", "bar")));
 		ConsumerRecords<?, ?> records = new ConsumerRecords<>(map);
 		Consumer<?, ?> consumer = mock(Consumer.class);
+		given(consumer.assignment()).willReturn(map.keySet());
+		AtomicBoolean pubPauseCalled = new AtomicBoolean();
 		willAnswer(inv -> {
-			eh.onPartitionsAssigned(consumer, List.of(new TopicPartition("foo", 0), new TopicPartition("foo", 1)));
+			eh.onPartitionsAssigned(consumer, List.of(new TopicPartition("foo", 0), new TopicPartition("foo", 1)),
+					() -> pubPauseCalled.set(true));
 			return records;
 		}).given(consumer).poll(any());
-		MessageListenerContainer container = mock(MessageListenerContainer.class);
+		KafkaMessageListenerContainer<?, ?> container = mock(KafkaMessageListenerContainer.class);
 		given(container.isRunning()).willReturn(true);
 		eh.handle(new RuntimeException(), records, consumer, container, () -> {
 			this.invoked++;
 			throw new RuntimeException();
 		});
 		assertThat(this.invoked).isEqualTo(1);
 		assertThat(recovered).hasSize(2);
-		InOrder inOrder = inOrder(consumer);
+		InOrder inOrder = inOrder(consumer, container);
 		inOrder.verify(consumer).pause(any());
+		inOrder.verify(container).publishConsumerPausedEvent(map.keySet(), "For batch retry");
 		inOrder.verify(consumer).poll(any());
 		inOrder.verify(consumer).pause(any());
 		inOrder.verify(consumer).resume(any());
+		inOrder.verify(container).publishConsumerResumedEvent(map.keySet());
 		verify(consumer, times(3)).assignment();
 		verifyNoMoreInteractions(consumer);
+		assertThat(pubPauseCalled.get()).isTrue();
 	}
 
 	@Test

spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackPauseResumeTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-2022 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.
@@ -51,11 +51,14 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.event.EventListener;
 import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.event.ConsumerPausedEvent;
+import org.springframework.kafka.event.ConsumerResumedEvent;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -93,6 +96,8 @@ public void dontResumeAlreadyPaused() throws Exception {
 		assertThat(this.config.resumedForNack).hasSize(1);
 		assertThat(this.config.pausedForNack).contains(new TopicPartition("foo", 1));
 		assertThat(this.config.resumedForNack).contains(new TopicPartition("foo", 1));
+		assertThat(this.config.pauseEvents).hasSize(1);
+		assertThat(this.config.resumeEvents).hasSize(1);
 	}
 
 	@Configuration
@@ -113,6 +118,10 @@ public static class Config {
 
 		final Set<TopicPartition> resumedForNack = new HashSet<>();
 
+		final List<ConsumerPausedEvent> pauseEvents = new ArrayList<>();
+
+		final List<ConsumerResumedEvent> resumeEvents = new ArrayList<>();
+
 		volatile int count;
 
 		volatile long replayTime;
@@ -232,6 +241,16 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			return factory;
 		}
 
+		@EventListener
+		public void paused(ConsumerPausedEvent event) {
+			this.pauseEvents.add(event);
+		}
+
+		@EventListener
+		public void resumed(ConsumerResumedEvent event) {
+			this.resumeEvents.add(event);
+		}
+
 	}
 
 }