consumer.pause(Collection<TopicPartition> var1) is delayed

[messaoudi-mounir]

With default settings, the following happens: (kafka, topic with only 1 partition).
The partition is initialized with messages that not yet consumed.

• msg1
• msg2
• msg3
• ...
• msg100

Running the sample example (from docs https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_usage_examples)

Using spring-cloud-stream-binder-kafka 2.0.1.RELEASE

@SpringBootApplication
@EnableBinding(Sink.class)
public class Application {

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

	@StreamListener(Sink.INPUT)
	public void in(String in, @Header(KafkaHeaders.CONSUMER) Consumer<?, ?> consumer) {
		System.out.println(in);
		consumer.pause(Collections.singleton(new TopicPartition("myTopic", 0)));
	}
}

Expected result:
The stream listener consume the first message (msg1) then pause.
Result after execution example
The stream listener consume the first message (msg1) and keep consuming the rest of messages..
After "X" time, if i write messages on topic, the messages ar enot consumed.

I think the problem is when calling pause method on the thread that invokes the @StreamListener though, the pause will be delayed (because the container will wait for that thread to exit for "X" secs).

Any quick response will be much appreciated!

[garyrussell]

Pausing the consumer doesn't discard any messages that have already been fetched by the previous poll(). It simple prevents new messages from being fetched on subsequent polls. You would need to discard those messages and perform a seek on the consumer so the discarded records will be refetched when you resume the consumer.

You can also set max.poll.records to 1 so only one record is fetched by each poll.

[messaoudi-mounir]

Thank you @garyrussell for answer for quick reply.
worked with max.poll.records=1

closed.