[
https://issues.apache.org/jira/browse/KAFKA-15818?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Philip Nee updated KAFKA-15818:
-------------------------------
Description:
The consumer needs to be polled at a candance lower than MAX_POLL_INTERVAL_MAX
otherwise the consumer should try to leave the group. Currently, we send an
acknowledgment event to the network thread per poll. The event only triggers
update on autocommit state, we need to implement updating the poll timer so
that the consumer can leave the group when the timer expires.
The current logic looks like this:
{code:java}
if (heartbeat.pollTimeoutExpired(now)) {
// the poll timeout has expired, which means that the foreground thread has
stalled
// in between calls to poll().
log.warn("consumer poll timeout has expired. This means the time between
subsequent calls to poll() " +
"was longer than the configured max.poll.interval.ms, which typically
implies that " +
"the poll loop is spending too much time processing messages. You can
address this " +
"either by increasing max.poll.interval.ms or by reducing the maximum
size of batches " +
"returned in poll() with max.poll.records.");
maybeLeaveGroup("consumer poll timeout has expired.");
} {code}
was:
In the network thread, we need a timer configure to take MAX_POLL_INTERVAL_MAX.
The reason is if the user don't poll the consumer within the internal, the
member needs to leave the group.
Currently, we send an acknowledgement event to the network thread per poll. It
needs to do two things 1. update autocommit state 2. update max poll interval
timer
The current logic looks like this:
{code:java}
if (heartbeat.pollTimeoutExpired(now)) {
// the poll timeout has expired, which means that the foreground thread has
stalled
// in between calls to poll().
log.warn("consumer poll timeout has expired. This means the time between
subsequent calls to poll() " +
"was longer than the configured max.poll.interval.ms, which typically
implies that " +
"the poll loop is spending too much time processing messages. You can
address this " +
"either by increasing max.poll.interval.ms or by reducing the maximum
size of batches " +
"returned in poll() with max.poll.records.");
maybeLeaveGroup("consumer poll timeout has expired.");
} {code}
> Implement max poll internval
> ----------------------------
>
> Key: KAFKA-15818
> URL: https://issues.apache.org/jira/browse/KAFKA-15818
> Project: Kafka
> Issue Type: Sub-task
> Components: consumer
> Reporter: Philip Nee
> Priority: Blocker
>
> The consumer needs to be polled at a candance lower than
> MAX_POLL_INTERVAL_MAX otherwise the consumer should try to leave the group.
> Currently, we send an acknowledgment event to the network thread per poll.
> The event only triggers update on autocommit state, we need to implement
> updating the poll timer so that the consumer can leave the group when the
> timer expires.
>
> The current logic looks like this:
> {code:java}
> if (heartbeat.pollTimeoutExpired(now)) {
> // the poll timeout has expired, which means that the foreground thread
> has stalled
> // in between calls to poll().
> log.warn("consumer poll timeout has expired. This means the time between
> subsequent calls to poll() " +
> "was longer than the configured max.poll.interval.ms, which typically
> implies that " +
> "the poll loop is spending too much time processing messages. You can
> address this " +
> "either by increasing max.poll.interval.ms or by reducing the maximum
> size of batches " +
> "returned in poll() with max.poll.records.");
> maybeLeaveGroup("consumer poll timeout has expired.");
> } {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
