Starting paused `ConcurrentMessageListenerContainer` results in messages being consumed

[bgK]

Affects Version(s): 2.2.6 (probably 2.3.x as well)

Use case:
I need to manually control the start offset for topic partitions while still using topic subscriptions; with multiple instances of the application sharing the same groupId.

The solution I plan to use is as follows:

• Start all the instances of the application but without starting the CMLCs.
• Have an external system transmit the desired start offsets to all the instances. Pause all the CMLCs. Start them all. In a ConsumerSeekAware message listener, seek the topic partitions in the onPartitionsAssigned callback.
• Once the consumer group is stabilized, resume all the CMLCs to start message consumption.

This works mostly fine but I have encountered two issues:

The paused state is not propagated to KafkaMessageListenerContainer on start:

ConcurrentMessageListenerContainer cmlc = ...;
cmlc.pause();
cmlc.start();

After executing this code, ConcurrentMessageListenerContainer is paused, but the created KafkaMessageListenerContainer are not.

It's seemingly possible to fix this issue by applying the following modification in ConcurrentMessageListenerContainer:

protected void doStart() {
...
	KafkaMessageListenerContainer<K, V> container = new KafkaMessageListenerContainer<>();
...
	if (isPaused()) {
		container.pause();
	}
...
	container.start();
	this.containers.add(container);
}

The first call to KafkaConsumer#poll still returns messages:
KafkaMessageListenerContainer.ListenerConsumer.ListenerConsumerRebalanceListener#onPartitionsAssigned is indirectly called by KafkaMessageListenerContainer.ListenerConsumer#pollAndInvoke, via consumer.poll.

In my case, the KMLC is paused when the rebalance occurs and thus this code is executed:

	ListenerConsumer.this.consumerPaused = false;
	ListenerConsumer.this.logger.warn("Paused consumer resumed by Kafka due to rebalance; "
		+ "the container will pause again before polling, unless the container's "
		+ "'paused' property is reset by a custom rebalance listener");

However, the call to consumer.poll that caused the rebalance listener to be called still returns messages.

As a quick test, I changed the code above to:

	ListenerConsumer.this.consumer.pause(partitions);

This yields the desired behavior. The call to consumer.poll does not return messages.
There may be other implications to this change I don't understand ...

(I wonder if there is a similar issue if a seek is performed from ConsumerSeekAware#onPartitionsAssigned while the KMLC is not paused -- messages from before the seek being returned by the call to consumer.poll. I did not test that case though.)

[garyrussell]

However, the call to consumer.poll that caused the rebalance listener to be called still returns messages.

...

This yields the desired behavior. The call to consumer.poll does not return messages.

I wonder if that is just happenstance? The rebalance listener is called from the poll() so I would not expect calling pause() here would affect the results of that poll().

[garyrussell]

Hmmm - with my testing, the poll() that causes the assignment never returns any records.

[bgK]

I wonder if that is just happenstance? The rebalance listener is called from the poll() so I would not expect calling pause() here would affect the results of that poll().

I'm not too familiar with the kafka codebase, but it seems to me the case is handled in KafkaConsumer:
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java#L1135

[garyrussell]

Yes; looks like it; thanks.

[garyrussell]

And I see there is a race - I was able to get the first poll() to return records using a debugger.

So, I think your fix is correct.

Thanks for the diagnosis.

[garyrussell]

I wonder if there is a similar issue if a seek is performed ...

It appears that this case is protected by the seek changing the offset in the TP state...

https://github.com/apache/kafka/blob/ce008e72de6ce67a8e9bb3d5d78fb0bb3849a38a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java#L1140-L1147