sanghyeok An created KAFKA-21248:
------------------------------------
Summary: AsyncKafkaConsumer may auto-commit offsets for records
not yet returned to the application
Key: KAFKA-21248
URL: https://issues.apache.org/jira/browse/KAFKA-21248
Project: Kafka
Issue Type: Bug
Reporter: sanghyeok An
Assignee: sanghyeok An
{{AsyncKafkaConsumer}} may include offsets for records that have not yet been
returned to application code in a periodic auto-commit. If the commit succeeds
and the application terminates before processing those records, they may be
skipped after restart, violating at-least-once processing semantics.
This can happen even when the application processes every record returned by
one {{poll()}} before calling {{poll()}} again:
{code:java}
while (running) {
ConsumerRecords<String, String> records = consumer.poll(timeout);
for (ConsumerRecord<String, String> record : records) {
process(record);
}
} {code}
{{AsyncKafkaConsumer.poll()}} submits an {{AsyncPollEvent}} to the background
thread. When buffered records are available and the required conditions are
satisfied, the application thread can collect those records before the event
has finished processing.
Collecting records advances {{SubscriptionState.position}} to the next offset
to read. This update occurs before {{poll()}} returns to the application and
does not indicate that application code has finished processing the records.
However, the background thread calls {{subscriptions.allConsumed()}} when
processing the {{AsyncPollEvent}} to determine which offsets to auto-commit. It
can therefore use a position that the application thread has already advanced
within the current {{poll()}} call.
Consider the following state for a single partition:
* {{{}enable.auto.commit=true{}}}, and the auto-commit interval has elapsed.
* The application has finished processing all records through offset 99.
* Both the committed offset stored on the broker and the current consumer
position are 100.
* Records at offsets 100 and 101 are available in the fetch buffer.
The following execution sequence is possible:
# *Application code calls {{{}consumer.poll(){}}}.*
All records from the previous {{poll()}} have already been processed. The
records at offsets 100 and 101 have not yet been delivered to application code.
# *The application thread submits an {{{}AsyncPollEvent{}}}.*
Before the background thread processes this event, the application thread can
proceed to collect buffered records.
# *The application thread collects the records at offsets 100 and 101 and
advances the position to 102.*
At this point, {{poll()}} may not have returned yet. The internal consumer
position is 102, but application code may still be waiting to receive the two
records.
# *The background thread processes the delayed {{{}AsyncPollEvent{}}}.*
Since the auto-commit interval has elapsed, it calls {{{}allConsumed(){}}}.
This method reads the current shared state and returns offset 102, which is
then included in the auto-commit request.
# *The background thread sends the commit request, and the broker stores
offset 102.*
This can complete while execution of the application thread is delayed.
Consequently, a resume position that skips these records can be stored before
they have even been returned to the application.
# *The application terminates before {{poll()}} returns.*
The records at offsets 100 and 101 have never been delivered to application
code, so {{process(record)}} has not been called for either record. However,
offset 102 has already been committed on the broker.
# *The same consumer group resumes consumption from the committed offset.*
Consumption starts at offset 102. The records at offsets 100 and 101 are
skipped without ever having been processed by application code.
In this example, the safe offset to commit when the current {{poll()}} begins
is 100. Reading the shared position when the event is processed instead
produces offset 102, which includes the records collected by the current
{{{}poll(){}}}.
The issue is a race between the position update inside {{poll()}} and the
background thread’s determination of the offsets to auto-commit. It does not
depend on how quickly application code processes returned records: the commit
can advance before those records are delivered to the application.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
