Since version 2.7, paused consumer auto-resume without asking just after pausing.

[theogoria]

Affects Version(s): 2.7.1

Hello,

I think I have found a bug with the pause method of Listener Containers.
When I call registry.getListenerContainers().forEach(MessageListenerContainer::pause); I have something like this in the log :

2021-07-06T18:20:16,650- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Paused consumption from: [partition-0]
2021-07-06T18:20:16,678- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Received: 0 records
2021-07-06T18:20:16,678- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Paused consumption from: [partition-1]
2021-07-06T18:20:16,718- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Received: 0 records
2021-07-06T18:20:16,718- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Paused consumption from: [partition-2]
2021-07-06T18:20:16,767- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Received: 0 records
2021-07-06T18:20:16,767- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Paused consumption from: [partition-3]
2021-07-06T18:20:21,651- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Resumed consumption from [partition-0]
2021-07-06T18:20:21,652- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Received: 0 records
2021-07-06T18:20:21,679- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Resumed consumption from [partition-1]
2021-07-06T18:20:21,679- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Received: 0 records
2021-07-06T18:20:21,719- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Resumed consumption from [partition-2]
2021-07-06T18:20:21,719- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Received: 0 records
2021-07-06T18:20:21,767- [DEBUG] [T-1] k.l.KafkaMessageListenerContainer$ListenerConsumer - [{}] - Resumed consumption from [partition-3]

We can see that the consumtion has resumed without the call of the resume method.

I think it's because the addition of the possibility to pause partitions in version 2.7.

When the pause of the consumer is requested all the partitions are paused but they are not added to the collection pauseRequestedPartitions of the AbstractMessageListenerContainer so they are not returned by isPartitionPauseRequested and they are restarted by the method KafkaMessageListenerContainer::resumePartitionsIfNecessary at the end of the poll.

[tomazfernandes]

Hi @theogoria, thanks for your input.

There are some things I don't quite understand here.

First thing is that pausing the consumer doesn't prevent it from polling - it will continue polling as usual and retrieve 0 records each time. So your logs shows the expected behaviour. If polling a paused partition did retrieve any records then we'd have a problem.

Second thing is that for resuming partitions we have this filter:
.filter(tp -> !isPartitionPauseRequested(tp) && pausedConsumerPartitions.contains(tp))
So if the partition was not paused by the pausePartition method, pausedConsumerPartitions will not contain the partition, and that partition will be filtered out and not resumed.

So I don't think there's any auto-resuming going on there. Am I missing something?

That being said, I see there's room for the pauseConsumer and pausePartition to have conflicts. For example if the user pauses a partition and then resumes the consumer, the container pause state will be left inconsistent and prevent the user from pausing the partition again.

The easiest solution I think would be adding / removing the paused / resumed partitions from the pausedConsumerPartitions collection, or simply delegate all pausing to the pause partition method:

Set<TopicPartition> paused = this.consumer.paused(); paused .forEach(KafkaMessageListenerContainer.this::resumePartition);

Since partition resuming and pausing is done after the consumer one we'd get it right next. One side effect would be firing PartitionPausedEvents when we pause the consumer, not sure if that's a behavior we want and whether it's a breaking change.

Which goes along the lines proposed in the PR, it just has an unnecessary manipulation of the pausedPartitions list after.

Either way it's not an as urgent matter as it sounded first I guess.

@garyrussell, what do you think of this?

Thanks!

[garyrussell]

I don't mind combining them; but there is a possibility that a partition is paused for retry; then the user pauses the entire container and resumes it (before the paused partition would normally be resumed). But, I suppose if that was the case, the delivery would be rejected again because it is too early, and the partition re-paused for the remainder of the pause time.

If there truly is a bug (that's why I would like to see a test case that exposes it without any changes to the container), then I would prefer to get it into next Monday's release.

[garyrussell]

It seems there is a bug...

@SpringBootApplication
public class Kgh1866Application {

	public static void main(String[] args) {
		SpringApplication.run(Kgh1866Application.class, args);
	}


	@KafkaListener(id = "kgh1866", topics = "kgh1866")
	public void listen(String in) {
		System.out.println(in);
	}

	@Bean
	public NewTopic topic() {
		return TopicBuilder.name("kgh1866").partitions(4).replicas(1).build();
	}

	@EventListener
	public void listener(ListenerContainerIdleEvent event) {
		event.getContainer(ConcurrentMessageListenerContainer.class).pause();
	}

}

logging.level.org.springframework.kafka=debug

spring.kafka.listener.idle-event-interval=5000

2021-07-13 17:44:25.747 DEBUG 79452 --- [  kgh1866-0-C-1] essageListenerContainer$ListenerConsumer : Received: 0 records
2021-07-13 17:44:25.748 DEBUG 79452 --- [  kgh1866-0-C-1] essageListenerContainer$ListenerConsumer : Commit list: {}
2021-07-13 17:44:25.749 DEBUG 79452 --- [  kgh1866-0-C-1] essageListenerContainer$ListenerConsumer : Paused consumption from: [kgh1866-0, kgh1866-2, kgh1866-1, kgh1866-3]
2021-07-13 17:44:30.754 DEBUG 79452 --- [  kgh1866-0-C-1] essageListenerContainer$ListenerConsumer : Resumed consumption from [kgh1866-0, kgh1866-2, kgh1866-1, kgh1866-3]
2021-07-13 17:44:30.755 DEBUG 79452 --- [  kgh1866-0-C-1] essageListenerContainer$ListenerConsumer : Received: 0 records

[tomazfernandes]

Hmm, you're right, I see it now.

The only thing I don't understand from the open PR is the need for adding the partitions to the list manually after pausing / resuming them, since they are already added by the methods.

For the BackOff behavior when a consumer is resumed I think what you described should be good enough, with the user resuming the consumer and the backed off partitions pausing again if necessary.

The last point would be about the PartitionPaused / Resumed Event, that would be fired for each partition when the consumer is paused / resumed. It seems it's a small change of contract, but I don't know if that's acceptable.

[garyrussell]

This filter

.filter(tp -> !isPartitionPauseRequested(tp)
	&& pausedConsumerPartitions.contains(tp))

will always return true for partitions that have not been individually paused.

I think the simplest solution is only call it if the container is not paused

			if (!this.consumerPaused) {
				resumePartitionsIfNecessary();
			}

[tomazfernandes]

What about the opposite? If a partition is paused and we resume the consumer, the partition will be resumed by resumeConsumerIfNecessary and then paused again in the next loop by pausePartitionIfNecessary, right?

[theogoria]

Hello @tomazfernandes and @garyrussell and thanks for your replies.

In my PR I considered that pausing a consumer is basically pausing all the partitions so effectively that introduce a kind of conflict between the functionalities of pausing a consumer and pausing a partition but that allow things such as pausing the consumer and then only resume one partition.

And I have added the partitions manually to pausedPartitions list to mark them has paused, this is not the same list, this is to keep the result of isPartitionPaused updated.

[garyrussell]

Yes, we could track actually paused partitions and not resume them when the container is resumed. Although, as I said, that will be self-correcting anyway.

[theogoria]

Yes, call resumePartitionsIfNecessary only if the container is not paused is a good solution to keep the two functionalities separated.

Do you wan't me to update my PR or to close it ?

[garyrussell]

Feel free to update the PR, or I will take it tomorrow.