Repository: apache/kafka
File: core/src/main/java/kafka/server/share/SharePartition.java (Java)

Code before:
        for (AcquiredRecords records : result) {
            acquiredCount += (int) (records.lastOffset() - records.firstOffset() + 1);
        }
        return new ShareAcquiredRecords(result, acquiredCount);
    }

    private List<AcquiredRecords> filterAbortedTransactionalAcquiredRecords(
        Iterable<? extends RecordBatch> batches,
        List<AcquiredRecords> acquiredRecords,
        List<FetchResponseData.AbortedTransaction> abortedTransactions
    ) {
        // The record batches that need to be archived in cachedState because they were a part of aborted transactions.
        List<RecordBatch> recordsToArchive = fetchAbortedTransactionRecordBatches(batches, abortedTransactions);
        for (RecordBatch recordBatch : recordsToArchive) {
            // Archive the offsets/batches in the cached state.
            NavigableMap<Long, InFlightBatch> subMap = fetchSubMap(recordBatch);
            archiveRecords(recordBatch.baseOffset(), recordBatch.lastOffset() + 1, subMap, RecordState.ACQUIRED);
        }
        return filterRecordBatchesFromAcquiredRecords(acquiredRecords, recordsToArchive);
    }

    private void logError(String message, Throwable e) {
        log.debug(message, e);
        if (!(e instanceof NetworkException)) {
            log.error(message, e);
        }
    }

    /**
     * This function filters out the offsets present in the acquired records list that are also a part of batches that need to be archived.
     * It follows an iterative refinement of acquired records to eliminate batches to be archived.
     * @param acquiredRecordsList The list containing acquired records. This list is sorted by the firstOffset of the acquired batch.
     * @param batchesToArchive The list containing record batches to archive. This list is sorted by the baseOffset of the record batch.
     * @return The list containing filtered acquired records offsets.
     */
    List<AcquiredRecords> filterRecordBatchesFromAcquiredRecords(
        List<AcquiredRecords> acquiredRecordsList,
        List<RecordBatch> batchesToArchive
    ) {
        Iterator<RecordBatch> batchesToArchiveIterator = batchesToArchive.iterator();
        if (!batchesToArchiveIterator.hasNext())
            return acquiredRecordsList;
        List<AcquiredRecords> result = new ArrayList<>();
        Iterator<AcquiredRecords> acquiredRecordsListIter = acquiredRecordsList.iterator();
        RecordBatch batchToArchive = batchesToArchiveIterator.next();
        AcquiredRecords unresolvedAcquiredRecords = null;

        while (unresolvedAcquiredRecords != null || acquiredRecordsListIter.hasNext()) {
            if (unresolvedAcquiredRecords == null)
                unresolvedAcquiredRecords = acquiredRecordsListIter.next();


Review comment (bug):
If not NetworkException then same error message will be twice in debug mode:
```suggestion
        if (e instanceof NetworkException) {
            log.debug(message, e);
        } else {
             log.error(message, e);
        }
```

Code after:
        for (AcquiredRecords records : result) {
            acquiredCount += (int) (records.lastOffset() - records.firstOffset() + 1);
        }
        return new ShareAcquiredRecords(result, acquiredCount);
    }

    private List<AcquiredRecords> filterAbortedTransactionalAcquiredRecords(
        Iterable<? extends RecordBatch> batches,
        List<AcquiredRecords> acquiredRecords,
        List<FetchResponseData.AbortedTransaction> abortedTransactions
    ) {
        // The record batches that need to be archived in cachedState because they were a part of aborted transactions.
        List<RecordBatch> recordsToArchive = fetchAbortedTransactionRecordBatches(batches, abortedTransactions);
        for (RecordBatch recordBatch : recordsToArchive) {
            // Archive the offsets/batches in the cached state.
            NavigableMap<Long, InFlightBatch> subMap = fetchSubMap(recordBatch);
            archiveRecords(recordBatch.baseOffset(), recordBatch.lastOffset() + 1, subMap, RecordState.ACQUIRED);
        }
        return filterRecordBatchesFromAcquiredRecords(acquiredRecords, recordsToArchive);
    }

    private void logError(String message, Throwable e) {
        if (e instanceof NetworkException) {
            log.debug(message, e);
        } else {
            log.error(message, e);
        }
    }

    /**
     * This function filters out the offsets present in the acquired records list that are also a part of batches that need to be archived.
     * It follows an iterative refinement of acquired records to eliminate batches to be archived.
     * @param acquiredRecordsList The list containing acquired records. This list is sorted by the firstOffset of the acquired batch.
     * @param batchesToArchive The list containing record batches to archive. This list is sorted by the baseOffset of the record batch.
     * @return The list containing filtered acquired records offsets.
     */
    List<AcquiredRecords> filterRecordBatchesFromAcquiredRecords(
        List<AcquiredRecords> acquiredRecordsList,
        List<RecordBatch> batchesToArchive
    ) {
        Iterator<RecordBatch> batchesToArchiveIterator = batchesToArchive.iterator();
        if (!batchesToArchiveIterator.hasNext())
            return acquiredRecordsList;
        List<AcquiredRecords> result = new ArrayList<>();
        Iterator<AcquiredRecords> acquiredRecordsListIter = acquiredRecordsList.iterator();
        RecordBatch batchToArchive = batchesToArchiveIterator.next();
        AcquiredRecords unresolvedAcquiredRecords = null;

        while (unresolvedAcquiredRecords != null || acquiredRecordsListIter.hasNext()) {
            if (unresolvedAcquiredRecords == null)
                unresolvedAcquiredRecords = acquiredRecordsListIter.next();