Feature request: pause/resume support

[Ycallaer]

Hi all,
I've got a question about the difference in behaviour of consumer.poll(0).
If you check the java doc for the consumer ( https://kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html) it states the following for the poll(0) function:

If 0, returns immediately with any records that are available currently in the buffer, else returns empty

I tried to do the same in python but when executing the code it already gives me the next record.
Sample of the code I am using:
`

                    future_response= executor.submit(do_processing,msg.value(),self.logger, kafka_key.generated_uuid,False,False)

                    is_running = True
                    while is_running == True:
                        is_running=future_response.running()
                        self.logger.info("Still waiting for the future to finish. Going into sleep")
                        msgx=consumer.poll(0)
                        time.sleep(0.5)

`
If I do a print statement of msgx.value() in the while loop it gives me the next record.
Why is there a difference in implementation ?
Is there a way to do manual heartbeating in python?

Since the future might take a while to get the result I want to avoid to set session.timeout.ms to a very high value.

[edenhill]

I believe they work the same.

If 0, returns immediately with any records that are available currently in the buffer, else returns empty

I'm interpreting that as it will return 0 or more messages already buffered/queued/pre-fetched, and that's exactly what the Python client will do as well.

Re Heartbeats: they're done in a background thread and the application does not need to worry about them.
Do note though that an application is still bound by the session timeout. If processing a single message takes longer than that, and auto commits are disalbed (e.g., to commit after processing), then an application risks not being able to commit offsets if the group rebalanced during processing, which might result in another consumer picking up the same message.

[Ycallaer]

Ok thanks for the feedback. I was trying to recreate the following java scenario in python:

consumer.pause(consumer.assignment().toArray(new TopicPartition[0]));
Future<Boolean> future = executor.submit(new ConsumeRecords(records,partitionToUncommittedOffsetMap));
futures.add(future);
			Boolean isCompleted = false;
			while(!isCompleted && !closed.get()) {
				try	{
					isCompleted = future.get(3, TimeUnit.SECONDS); // wait up-to heart-beat interval
				} catch (TimeoutException e) {
					logger.debug("C : {}, heartbeats the coordinator", clientId);
					consumer.poll(0); // does heart-beat
					commitOffsets(partitionToUncommittedOffsetMap); 
				} catch (CancellationException e) {
					logger.debug("C : {}, ConsumeRecords Job got cancelled", clientId);
					break;
				} catch (ExecutionException | InterruptedException e) {
					logger.error("C : {}, Error while consuming records", clientId, e);
					break;
				}
			}
			futures.remove(future);
			consumer.resume(consumer.assignment().toArray(new TopicPartition[0]));

Because the consumer is paused the consumer.poll(0) will probably not fetch anything from the buffer right ?
Is there a way in python to pause the consumer ?

[edenhill]

While the underlying C library has pause/resume support this is not currently exposed in the Python bindings.

But since you don't need to call poll() to keep heartbeats alive in the Python client do you really need to pause?
Do note though that if you don't call poll() until your processing is done you will not only delay the next message but also any callbacks that might be triggered, such as rebalance (on_assign/revoke, stats, on_commit, ...etc).

[Ycallaer]

The reason I want to do the polling myself is that the processing call might take be between 1 and 10 minutes depending on the incoming data.
The large messages are rare but they do occur. If I want to take these into account I need to set "session.timeout.ms" to a rather large value (which has a drawback if there is a real failure).
What we currently do is package the application in Docker and spin up several containers depending on the load (we also have multiple partitions). This allows us to control the load on the application. Doing the call of the process async might result in problems with the application (the share load would be too high).

Are there any plans to make pause / resume available in python ?

[edenhill]

How are you handling offset commits?
Do you leave enable.auto.commit=true enabled, which causes offsets to be (stored for future) commit prior to passing the message to the application?
Or do you disable it auto commits and do manual explicit commits after processing a message?

re pause/resume: We're adding APIs on demand if there's a use for it.

[Ycallaer]

I do manually explicit commits after processing is done in order to make sure that each message gets properly processed.

Can I already officially request adding pause resume to the API? Maybe more people might need it in the future.

[edenhill]

Feature request noted!

Are you calling poll() while you are processing a message?

[duke-bartholomew]

@Ycallaer +1 on Pause/Resume support.
This would surely help when implementing your own 'chooser' logic.
Currently this is not possible without breaking the consumer group assignments.

[jmgpeeters]

"The reason I want to do the polling myself is that the processing call might take be between 1 and 10 minutes depending on the incoming data."

We have exactly the same issue. It's currently very hard to deal with it robustly, in any python kafka client (not sure about other languages), i.e. haven't actually managed, because of reliably handling potential rebalances (which unpause everything, I think) whilst you are paused etc. Adding rebalance listeners to immediately re-pause feels a bit brittle.

[edenhill]

Adding support for KIP-62 is probably the way to go here.

Decoupling the processing timeout: We propose to introduce a separate locally enforced timeout for record processing and a background thread to keep the session active until this timeout expires. We call this new timeout as the "process timeout" and expose it in the consumer's configuration as max.poll.interval.ms. This config sets the maximum delay between client calls to poll(). When the timeout expires, the consumer will stop sending heartbeats and send an explicit LeaveGroup request. As soon as the consumer resumes processing with another call to poll(), the consumer will rejoin the group. This is equivalent to the current processing model except that it allows the user to set a higher timeout when processing while also using a lower session timeout for faster crash detection.

[dangra]

hi @edenhill, is KIP-62 something to implement in librdkafka or python bindings?