private void releaseAcquisitionLockOnTimeoutForCompleteBatch(InFlightBatch inFlightBatch,
                                                                 List<PersisterStateBatch> stateBatches,
                                                                 String memberId) {
        if (inFlightBatch.batchState() == RecordState.ACQUIRED) {
            InFlightState updateResult = inFlightBatch.tryUpdateBatchState(
                    inFlightBatch.lastOffset() < startOffset ? RecordState.ARCHIVED : RecordState.AVAILABLE,
                    false,
                    maxDeliveryCount,
                    EMPTY_MEMBER_ID);
            if (updateResult == null) {
                log.error("Unable to release acquisition lock on timeout for the batch: {}"
                        + " for the share partition: {}-{} memberId: {}", inFlightBatch, groupId, topicIdPartition, memberId);
                return;
            }
            stateBatches.add(new PersisterStateBatch(inFlightBatch.firstOffset(), inFlightBatch.lastOffset(),
                    updateResult.state.id, (short) updateResult.deliveryCount));

            // Update acquisition lock timeout task for the batch to null since it is completed now.
            updateResult.updateAcquisitionLockTimeoutTask(null);
            if (updateResult.state != RecordState.ARCHIVED) {
                findNextFetchOffset.set(true);
            }
            return;
        }
        log.debug("The batch is not in acquired state while release of acquisition lock on timeout, skipping, batch: {}"
                + " for the share group: {}-{}-{}", inFlightBatch, groupId, memberId, topicIdPartition);
    }
Apply acquisition lock to acquired records.

@param memberId The member id of the client that is putting the acquisition lock.
@param firstOffset The first offset of the acquired records.
@param lastOffset The last offset of the acquired records.
@param delayMs The delay in milliseconds after which the acquisition lock will be released.
core/src/main/java/kafka/server/share/SharePartition.java