ShivsundarR commented on code in PR #20752:
URL: https://github.com/apache/kafka/pull/20752#discussion_r2454044648
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java:
##########
@@ -100,9 +100,16 @@ public class ShareCompletedFetch {
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<>();
partitionAcquiredRecords.forEach(acquiredRecords -> {
for (long offset = acquiredRecords.firstOffset(); offset <=
acquiredRecords.lastOffset(); offset++) {
- 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());
Review Comment:
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`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
