KAFKA-19789: Log an error when we get duplicate acquired offsets in ShareFetchResponse. #20752

ShivsundarR · 2025-10-22T17:26:15Z

What
https://issues.apache.org/jira/browse/KAFKA-19789

There were some scenarios where ShareFetchResponse contained
duplicate acquired records, this was a broker side bug.
Although ideally this should not happen, the client was not expecting
this case and acknowledged with GAP type for any duplicate occurrence.
This case should be logged as an error in the client, and we must not
acknowledge the duplicate offsets as the broker is already in a bad
state.
PR adds an error log for this case and a unit test for the same.

chia7712

@ShivsundarR thanks for this patch!

chia7712 · 2025-10-22T18:49:35Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java

-                acquiredRecordList.add(new OffsetAndDeliveryCount(offset, acquiredRecords.deliveryCount()));
+                if (!offsets.add(offset)) {
+                    log.error("Duplicate acquired record offset {} found in share fetch response for partition {}. " +
+                            "This indicates a broker processing issue.", offset, partition.topicPartition());


Just curious, are there any known issues that lead to duplicate offsets?

Yes, there was a broker side issue when SharePartition was at capacity - https://issues.apache.org/jira/browse/KAFKA-19808. Due to this, we were getting duplicate offsets (with different delivery counts) in the ShareFetchResponse.

There are no current known issues, but there was previously an issue in the broker and adding logging would have made it quicker to get to the bottom of it.

chia7712 · 2025-10-22T19:01:11Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetchTest.java

+
+        // Verify all offsets are unique
+        Set<Long> offsetSet = new HashSet<>();
+        for (ConsumerRecord<String, String> record : records) {


I'm not sure if this covers the new behavior, since inFlightRecords already handles offset deduplication.

Yes the logic around inFlightRecords ensures we do not send duplicate offsets to the application side, but the client does respond with a GAP acknowledgement to the broker for any duplicate offset.

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java

Line 219 in 2c44448

inFlightBatch.addGap(nextAcquired.offset);

Without deduplication, when the offset is encountered second time,lastRecord.offset > nextAcquired.offset, (as nextAcquired will be an older offset) will be true, so the client acknowledges these offsets as GAPs which is kind of hiding the main issue.
As the broker is already in a bad state(duplication should never happen), we thought of logging an error and ignoring any duplicates on the client.

AndrewJSchofield

Thanks for the PR. Just one initial comment from a first look.

AndrewJSchofield · 2025-10-23T10:00:16Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java

    private List<OffsetAndDeliveryCount> buildAcquiredRecordList(List<ShareFetchResponseData.AcquiredRecords> partitionAcquiredRecords) {
        List<OffsetAndDeliveryCount> acquiredRecordList = new LinkedList<>();
+        // Set to find duplicates in case of overlapping acquired records
+        Set<Long> offsets = new HashSet<>();


I wonder if you could change the partitionAcquiredRecords into a LinkedHashMap or similar to combine the duplicate checking with the ordered iteration.

I had a look into making the acquiredRecordsList(LinkedList<OffsetAndDeliveryCount>) into a LinkedHashMap This change would actually have a bit of a code change around listIterator, we might have to use map.entrySet().iterator() for rewinding to the start of the list.
And as we are doing sequential operations and not key based, probably better to keep it as a list?
I have changed it to ArrayList instead of a LinkedList though as it would give better iteration performance for build once and iterate use cases.

AndrewJSchofield · 2025-10-23T16:38:20Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java


    private List<OffsetAndDeliveryCount> buildAcquiredRecordList(List<ShareFetchResponseData.AcquiredRecords> partitionAcquiredRecords) {
-        List<OffsetAndDeliveryCount> acquiredRecordList = new LinkedList<>();
+        List<OffsetAndDeliveryCount> acquiredRecordList = new ArrayList<>();


By default, a new list will have space for 10 elements. Resizing is expensive. Maybe one optimisation would be to see how many offsets are in the first element in the partitionAcquiredRecords, and using that number as the initial size of the list. In the case of only one batch of offsets, the list will be the correct size alrready. wdyt?

That makes sense, if most of the times the response is gonna contain only 1 batch, we can avoid resizing. I have made the change. Thanks.

AndrewJSchofield

Thanks for the update. Just one more comment.

AndrewJSchofield · 2025-10-24T09:59:39Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java

    private List<OffsetAndDeliveryCount> buildAcquiredRecordList(List<ShareFetchResponseData.AcquiredRecords> partitionAcquiredRecords) {
-        List<OffsetAndDeliveryCount> acquiredRecordList = new LinkedList<>();
+        // Setting the size of the array to the size of the first batch of acquired records. In case there is only 1 batch acquired, resizing would not happen.
+        int initialListSize = !partitionAcquiredRecords.isEmpty() ? (int) (partitionAcquiredRecords.get(0).lastOffset() -


In the case where the partitionAcquiredRecords is empty, we can just make an empty list and return directly. We don't need to make the HashSet only to discard it unused because the loop will not have any iterations.

Yes makes sense, I have updated the code now.

Log an error when we get duplicate acquired offsets

5e4ebde

github-actions bot added triage PRs from the community consumer clients small Small PRs labels Oct 22, 2025

ShivsundarR added KIP-932 Queues for Kafka ci-approved and removed triage PRs from the community labels Oct 22, 2025

chia7712 reviewed Oct 22, 2025

View reviewed changes

AndrewJSchofield self-requested a review October 23, 2025 09:34

AndrewJSchofield reviewed Oct 23, 2025

View reviewed changes

Use ArrayList

07f9502

AndrewJSchofield reviewed Oct 23, 2025

View reviewed changes

Set initial size for array

a729d1f

ShivsundarR requested a review from AndrewJSchofield October 24, 2025 09:52

AndrewJSchofield requested changes Oct 24, 2025

View reviewed changes

ShivsundarR added 2 commits October 24, 2025 15:57

Address comment

8db08e5

Merge remote-tracking branch 'upstream/trunk' into KAFKA-19789

3c4a21b

ShivsundarR requested a review from AndrewJSchofield October 24, 2025 18:08

KAFKA-19789: Log an error when we get duplicate acquired offsets in ShareFetchResponse. #20752

Are you sure you want to change the base?

KAFKA-19789: Log an error when we get duplicate acquired offsets in ShareFetchResponse. #20752

Conversation

ShivsundarR commented Oct 22, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ShivsundarR commented Oct 22, 2025 •

edited by github-actions bot

Loading