How to deal with long-running message processors?

[smares]

Since we cannot set a value for max.poll.records, how many records are polled from Kafka with a single request? 500? I am asking because max.poll.interval.ms states:

The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member. For consumers using a non-null group.instance.id which reach this timeout, partitions will not be immediately reassigned. Instead, the consumer will stop sending heartbeats and partitions will be reassigned after expiration of session.timeout.ms. This mirrors the behavior of a static consumer which has shutdown.

Given I don't use auto-commit, but commit every 10 messages. If I have a consumer that takes 1 minute for processing a single message and by default 500 messages are polled, does that mean that with a max.poll.interval.ms set to the default of 5 minutes, Kafka will trigger a rebalance after 5 processed messages (since the next poll that pulls 500 messages happens only after the queued 500 messages are all processed) thinking the consumer is dead?

[Steveb-p]

You'd have to try this, but this does not mean afaik that the PHP process needs to explicitly call poll().

Kafka communication is done in a separate "thread" (if not a straight up thread) and client should keep sending heartbeat to Kafka. Only after 5 minutes passed and the client was not active at all, it will be considered "dead" and a rebalance will happen.

But you should test it out to be sure.
See https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md?plain=1, as this is the library php-rdkafka uses under the hood, and might be more accurate.

[smares]

Thanks! I think the heartbeats are not enough since I see rebalancing happening from time to time + my attempt to commit a message fails with Specified group generation id is not valid. According to the logs, this is always the case when processing one message takes longer than max.poll.interval.ms. Now I was unsure if increasing max.poll.interval.ms to a value as high as the longest possible duration of ONE message processing time is enough, or if theoretically it should be the longst possible duration of one message times the number of messages polled (500). Latter would be ridiculous since ~5 minutes x 500 would be 41h. 😆 That's why I thought I must also set max.poll.records to 1 and since this option cannot be set, I could set fetch.max.bytes to 0 and make use of the "if the first message batch in the first non-empty partition of the Fetch request is larger than this value, then the message batch will still be returned to ensure the consumer can make progress" logic.

[smares]

I am not even sure if a commit resets max.poll.interval.ms because if yes, I could simply manually commit after each processed message (+ still increase max.poll.interval.ms to account for the longest possible duration of processing a single message, taking into account time for multiple API calls and their timeouts).

Edit: It does not.