[jira] [Created] (KAFKA-21248) AsyncKafkaConsumer may auto-commit offsets for records not yet returned to the application

["sanghyeok An (Jira)" <ji...@apache.org>]

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)