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

Code before:
        try {
            for (Map.Entry<Long, InFlightState> offsetState : inFlightBatch.offsetState().entrySet()) {
                // For the first batch which might have offsets prior to the request base
                // offset i.e. cached batch of 10-14 offsets and request batch of 12-13.
                if (offsetState.getKey() < requestFirstOffset) {
                    continue;
                }

                if (offsetState.getKey() > requestLastOffset) {
                    // No further offsets to process.
                    break;
                }

                if (offsetState.getValue().state() != RecordState.AVAILABLE || offsetState.getValue().hasOngoingStateTransition()) {
                    log.trace("The offset {} is not available in share partition: {}-{}, skipping: {}",
                        offsetState.getKey(), groupId, topicIdPartition, inFlightBatch);
                    continue;
                }

                int recordDeliveryCount = offsetState.getValue().deliveryCount();
                // If the record is on last delivery attempt then isolate that record to be delivered alone.
                // If the respective record is corrupt then it prevents increasing delivery count of multiple
                // records in a single response batch. Condition below checks if the current record has reached
                // the delivery limit and already have some records to return in response then skip processing
                // the current record, which shall be delivered alone in next fetch.
                if (maxDeliveryCount() > 2 && recordDeliveryCount == maxDeliveryCount() - 1 && acquiredCount > 0) {
                    log.warn("The offset {} is on last delivery attempt in share partition: {}-{}, should be delivered alone in next fetch",
                        offsetState.getKey(), groupId, topicIdPartition);
                    break;
                }

                // When record delivery count reach the throttle threshold, progressively reduce batch size to isolate records.
                // The `maxFetchRecordsWhileThrottledRecords` is halved with each additional delivery attempt beyond the throttle limit.
                // Example:
                //   - maxDeliveryCount = 6, throttleRecordsDeliveryLimit = 3, batch size = 500
                //   - deliveryCount = 3: maxFetchRecords = 500 >> (3 - 3 + 1) = 250
                //   - deliveryCount = 4: maxFetchRecords = 500 >> (4 - 3 + 1) = 125
                // The `maxFetchRecordsWhileThrottledRecords` is calculated based on the first acquirable
                // record that meets the throttling criteria in the batch. Generally, when complete
                // batch is erroring out and increases delivery count then the front offset can have
                // higher delivery count then the later offsets, for example for a batch of 500 records
                // and delivery limit of 10 the division will happen as 250, 125, 62, 31, 1 which means
                // offset 124 will be at higher delivery count than offset 125. The calculation below
                // deliver 124 once alone and then proceed with 125 onwards. The code ensures that
                // the records at higher delivery count are isolated first.
                if (recordDeliveryCount >= throttleRecordsDeliveryLimit() && maxFetchRecordsWhileThrottledRecords < 0) {
                    maxFetchRecordsWhileThrottledRecords = Math.max(1, (long) inFlightBatch.offsetState().size() >> (recordDeliveryCount - throttleRecordsDeliveryLimit() + 1));
                    hasThrottledRecord = true;
                }

                InFlightState updateResult = offsetState.getValue().tryUpdateState(RecordState.ACQUIRED, DeliveryCountOps.INCREASE,

Review comment (bug):
I'd prefer you to obtain the values of `maxDeliveryCount()` and `throttleRecordsDeliveryLimit()` at the start of the method and then use those values throughout. There is a tiny chance that the configs change during the execution of the method and I think there's a minuscule chance of a peculiar bug emerging.

Code after:
        List<AcquiredRecords> offsetAcquiredRecords = new ArrayList<>();
        lock.writeLock().lock();
        try {
            for (Map.Entry<Long, InFlightState> offsetState : inFlightBatch.offsetState().entrySet()) {
                // For the first batch which might have offsets prior to the request base
                // offset i.e. cached batch of 10-14 offsets and request batch of 12-13.
                if (offsetState.getKey() < requestFirstOffset) {
                    continue;
                }

                if (offsetState.getKey() > requestLastOffset) {
                    // No further offsets to process.
                    break;
                }

                if (offsetState.getValue().state() != RecordState.AVAILABLE || offsetState.getValue().hasOngoingStateTransition()) {
                    log.trace("The offset {} is not available in share partition: {}-{}, skipping: {}",
                        offsetState.getKey(), groupId, topicIdPartition, inFlightBatch);
                    continue;
                }

                int recordDeliveryCount = offsetState.getValue().deliveryCount();
                // If the record is on last delivery attempt then isolate that record to be delivered alone.
                // If the respective record is corrupt then it prevents increasing delivery count of multiple
                // records in a single response batch. Condition below checks if the current record has reached
                // the delivery limit and already have some records to return in response then skip processing
                // the current record, which shall be delivered alone in next fetch.
                if (maxDeliveryCount > 2 && recordDeliveryCount == maxDeliveryCount - 1 && acquiredCount > 0) {
                    log.warn("The offset {} is on last delivery attempt in share partition: {}-{}, should be delivered alone in next fetch",
                        offsetState.getKey(), groupId, topicIdPartition);
                    break;
                }

                // When record delivery count reach the throttle threshold, progressively reduce batch size to isolate records.
                // The `maxFetchRecordsWhileThrottledRecords` is halved with each additional delivery attempt beyond the throttle limit.
                // Example:
                //   - maxDeliveryCount = 6, throttleRecordsDeliveryLimit = 3, batch size = 500
                //   - deliveryCount = 3: maxFetchRecords = 500 >> (3 - 3 + 1) = 250
                //   - deliveryCount = 4: maxFetchRecords = 500 >> (4 - 3 + 1) = 125
                // The `maxFetchRecordsWhileThrottledRecords` is calculated based on the first acquirable
                // record that meets the throttling criteria in the batch. Generally, when complete
                // batch is erroring out and increases delivery count then the front offset can have
                // higher delivery count then the later offsets, for example for a batch of 500 records
                // and delivery limit of 10 the division will happen as 250, 125, 62, 31, 1 which means
                // offset 124 will be at higher delivery count than offset 125. The calculation below
                // deliver 124 once alone and then proceed with 125 onwards. The code ensures that
                // the records at higher delivery count are isolated first.
                if (recordDeliveryCount >= throttleRecordsDeliveryLimit && maxFetchRecordsWhileThrottledRecords < 0) {
                    maxFetchRecordsWhileThrottledRecords = Math.max(1, (long) inFlightBatch.offsetState().size() >> (recordDeliveryCount - throttleRecordsDeliveryLimit + 1));
                    hasThrottledRecord = true;
                }