Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java (Java)

Code before:
                        final TopicIdPartition partition,
                        final ShareFetchResponseData.PartitionData partitionData,
                        final ShareFetchMetricsAggregator metricAggregator,
                        final short requestVersion) {
        this.log = logContext.logger(org.apache.kafka.clients.consumer.internals.ShareCompletedFetch.class);
        this.decompressionBufferSupplier = decompressionBufferSupplier;
        this.nodeId = nodeId;
        this.partition = partition;
        this.partitionData = partitionData;
        this.metricAggregator = metricAggregator;
        this.requestVersion = requestVersion;
        this.batches = ShareFetchResponse.recordsOrFail(partitionData).batches().iterator();
        this.acquiredRecordList = buildAcquiredRecordList(partitionData.acquiredRecords());
        this.nextAcquired = null;
    }

    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++) {
                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());
                } else {
                    acquiredRecordList.add(new OffsetAndDeliveryCount(offset, acquiredRecords.deliveryCount()));
                }
            }
        });
        return acquiredRecordList;
    }

    boolean isInitialized() {
        return initialized;
    }

    void setInitialized() {
        this.initialized = true;
    }

    public boolean isConsumed() {
        return isConsumed;
    }

    /**
     * Draining a {@link ShareCompletedFetch} will signal that the data has been consumed and the underlying resources
     * are closed. This is somewhat analogous to {@link Closeable#close() closing}, though no error will result if a
     * caller invokes {@link #fetchRecords(Deserializers, int, boolean)}; an empty {@link List list} will be
     * returned instead.
     */

Review comment (refactor):
I wonder if you could change the `partitionAcquiredRecords` into a `LinkedHashMap` or similar to combine the duplicate checking with the ordered iteration.

Code after:
                        final TopicIdPartition partition,
                        final ShareFetchResponseData.PartitionData partitionData,
                        final ShareFetchMetricsAggregator metricAggregator,
                        final short requestVersion) {
        this.log = logContext.logger(org.apache.kafka.clients.consumer.internals.ShareCompletedFetch.class);
        this.decompressionBufferSupplier = decompressionBufferSupplier;
        this.nodeId = nodeId;
        this.partition = partition;
        this.partitionData = partitionData;
        this.metricAggregator = metricAggregator;
        this.requestVersion = requestVersion;
        this.batches = ShareFetchResponse.recordsOrFail(partitionData).batches().iterator();
        this.acquiredRecordList = buildAcquiredRecordList(partitionData.acquiredRecords());
        this.nextAcquired = null;
    }

    private List<OffsetAndDeliveryCount> buildAcquiredRecordList(List<ShareFetchResponseData.AcquiredRecords> partitionAcquiredRecords) {
        // 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.
        if (partitionAcquiredRecords.isEmpty()) {
            return List.of();
        }
        int initialListSize = (int) (partitionAcquiredRecords.get(0).lastOffset() - partitionAcquiredRecords.get(0).firstOffset() + 1);
        List<OffsetAndDeliveryCount> acquiredRecordList = new ArrayList<>(initialListSize);

        // 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++) {
                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());
                } else {
                    acquiredRecordList.add(new OffsetAndDeliveryCount(offset, acquiredRecords.deliveryCount()));
                }
            }
        });
        return acquiredRecordList;
    }

    boolean isInitialized() {
        return initialized;
    }

    void setInitialized() {
        this.initialized = true;
    }

    public boolean isConsumed() {
        return isConsumed;
    }