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

Code before:
    private ListIterator<OffsetAndDeliveryCount> acquiredRecordIterator;
    private OffsetAndDeliveryCount nextAcquired;
    private final ShareFetchMetricsAggregator metricAggregator;

    ShareCompletedFetch(final LogContext logContext,
                        final BufferSupplier decompressionBufferSupplier,
                        final int nodeId,
                        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.
        int initialListSize = !partitionAcquiredRecords.isEmpty() ? (int) (partitionAcquiredRecords.get(0).lastOffset() -
                partitionAcquiredRecords.get(0).firstOffset() + 1) : 0;
        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;
    }

Review comment (suggestion):
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.

Code after:
    private ListIterator<OffsetAndDeliveryCount> acquiredRecordIterator;
    private OffsetAndDeliveryCount nextAcquired;
    private final ShareFetchMetricsAggregator metricAggregator;

    ShareCompletedFetch(final LogContext logContext,
                        final BufferSupplier decompressionBufferSupplier,
                        final int nodeId,
                        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() {