[
https://issues.apache.org/jira/browse/KAFKA-21249?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
sanghyeok An updated KAFKA-21249:
---------------------------------
Description:
When using the Java {{AsyncKafkaConsumer}} with {{group.protocol=consumer}} and
{{{}enable.auto.commit=true{}}}, a pre-rebalance auto-commit retry may commit
an offset beyond records that the application has not yet processed.
When CommitRequestManager.autoCommitSyncBeforeRebalanceWithRetries() receives a
retriable error, it calls {{subscriptions.allConsumed()}} again and replaces
the existing request’s offset map. These offsets reflect consumer positions
advanced while collecting records for {{{}poll(){}}}, rather than the
application’s processing progress.
During a rebalance, record collection is blocked for partitions pending
revocation, but records can still be returned from retained partitions.
Although the revocation and assignment sequence waits for the auto-commit to
complete, the application thread does not wait for the entire commit retry
sequence. Consequently, the following sequence is possible:
# After the application finishes processing records through offset 99 in
partition P, a pre-rebalance auto-commit starts with offset 100.
# While the initial commit request is pending, {{poll()}} returns records
100–149 from retained partition P. The internal position advances to 150, and
the application starts processing the batch.
# The initial commit request fails with a retriable error.
# The retry logic reads the current position again, changing the offset to
commit from 100 to 150.
# If offset 150 is committed and the process exits before processing the
entire batch, the restarted consumer may resume from 150 and skip the remaining
unprocessed records.
This can occur even when a single application thread follows the sequence
{{{}poll → process all returned records → next poll{}}}. It does not require
the periodic auto-commit interval to expire.
An auto-commit retry should not commit beyond the application’s processing
progress. However, this retry path can include positions advanced by records
returned after the initial request was created. This may cause records to
remain unprocessed by the consumer group after a failure.
was:
When using the Java {{AsyncKafkaConsumer}} with {{group.protocol=consumer}} and
{{{}enable.auto.commit=true{}}}, a pre-rebalance auto-commit retry may commit
an offset beyond records that the application has not yet processed.
When{{ CommitRequestManager.autoCommitSyncBeforeRebalanceWithRetries()}}
receives a retriable error, it calls {{subscriptions.allConsumed()}} again and
replaces the existing request’s offset map. These offsets reflect consumer
positions advanced while collecting records for {{{}poll(){}}}, rather than the
application’s processing progress.
During a rebalance, record collection is blocked for partitions pending
revocation, but records can still be returned from retained partitions.
Although the revocation and assignment sequence waits for the auto-commit to
complete, the application thread does not wait for the entire commit retry
sequence. Consequently, the following sequence is possible:
# After the application finishes processing records through offset 99 in
partition P, a pre-rebalance auto-commit starts with offset 100.
# While the initial commit request is pending, {{poll()}} returns records
100–149 from retained partition P. The internal position advances to 150, and
the application starts processing the batch.
# The initial commit request fails with a retriable error.
# The retry logic reads the current position again, changing the offset to
commit from 100 to 150.
# If offset 150 is committed and the process exits before processing the
entire batch, the restarted consumer may resume from 150 and skip the remaining
unprocessed records.
This can occur even when a single application thread follows the sequence
{{{}poll → process all returned records → next poll{}}}. It does not require
the periodic auto-commit interval to expire.
An auto-commit retry should not commit beyond the application’s processing
progress. However, this retry path can include positions advanced by records
returned after the initial request was created. This may cause records to
remain unprocessed by the consumer group after a failure.
> Pre-rebalance auto-commit retries may commit offsets beyond unprocessed
> records
> -------------------------------------------------------------------------------
>
> Key: KAFKA-21249
> URL: https://issues.apache.org/jira/browse/KAFKA-21249
> Project: Kafka
> Issue Type: Bug
> Reporter: sanghyeok An
> Assignee: sanghyeok An
> Priority: Major
> Labels: clients, consumer
>
> When using the Java {{AsyncKafkaConsumer}} with {{group.protocol=consumer}}
> and {{{}enable.auto.commit=true{}}}, a pre-rebalance auto-commit retry may
> commit an offset beyond records that the application has not yet processed.
> When CommitRequestManager.autoCommitSyncBeforeRebalanceWithRetries() receives
> a retriable error, it calls {{subscriptions.allConsumed()}} again and
> replaces the existing request’s offset map. These offsets reflect consumer
> positions advanced while collecting records for {{{}poll(){}}}, rather than
> the application’s processing progress.
> During a rebalance, record collection is blocked for partitions pending
> revocation, but records can still be returned from retained partitions.
> Although the revocation and assignment sequence waits for the auto-commit to
> complete, the application thread does not wait for the entire commit retry
> sequence. Consequently, the following sequence is possible:
> # After the application finishes processing records through offset 99 in
> partition P, a pre-rebalance auto-commit starts with offset 100.
> # While the initial commit request is pending, {{poll()}} returns records
> 100–149 from retained partition P. The internal position advances to 150, and
> the application starts processing the batch.
> # The initial commit request fails with a retriable error.
> # The retry logic reads the current position again, changing the offset to
> commit from 100 to 150.
> # If offset 150 is committed and the process exits before processing the
> entire batch, the restarted consumer may resume from 150 and skip the
> remaining unprocessed records.
> This can occur even when a single application thread follows the sequence
> {{{}poll → process all returned records → next poll{}}}. It does not require
> the periodic auto-commit interval to expire.
> An auto-commit retry should not commit beyond the application’s processing
> progress. However, this retry path can include positions advanced by records
> returned after the initial request was created. This may cause records to
> remain unprocessed by the consumer group after a failure.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
