private void releaseAcquisitionLockOnTimeout(String memberId, long firstOffset, long lastOffset) {
        lock.writeLock().lock();
        try {
            Map.Entry<Long, InFlightBatch> floorOffset = cachedState.floorEntry(firstOffset);
            if (floorOffset == null) {
                log.error("Base offset {} not found for share partition: {}-{}", firstOffset, groupId, topicIdPartition);
                return;
            }
            List<PersisterStateBatch> stateBatches = new ArrayList<>();
            NavigableMap<Long, InFlightBatch> subMap = cachedState.subMap(floorOffset.getKey(), true, lastOffset, true);
            for (Map.Entry<Long, InFlightBatch> entry : subMap.entrySet()) {
                InFlightBatch inFlightBatch = entry.getValue();

                if (inFlightBatch.offsetState() == null
                        && inFlightBatch.batchState() == RecordState.ACQUIRED
                        && checkForStartOffsetWithinBatch(inFlightBatch.firstOffset(), inFlightBatch.lastOffset())) {

                    // For the case when batch.firstOffset < start offset <= batch.lastOffset, we will be having some
                    // acquired records that need to move to archived state despite their delivery count.
                    inFlightBatch.maybeInitializeOffsetStateUpdate();
                }

                // Case when the state of complete batch is valid
                if (inFlightBatch.offsetState() == null) {
                    releaseAcquisitionLockOnTimeoutForCompleteBatch(inFlightBatch, stateBatches, memberId);
                } else { // Case when batch has a valid offset state map.
                    releaseAcquisitionLockOnTimeoutForPerOffsetBatch(inFlightBatch, stateBatches, memberId, firstOffset, lastOffset);
                }
            }

            if (!stateBatches.isEmpty() && !isWriteShareGroupStateSuccessful(stateBatches)) {

                // Even if write share group state RPC call fails, we will still go ahead with the state transition.
                log.error("Failed to write the share group state on acquisition lock timeout for share partition: {}-{} memberId {}. " +
                                "Proceeding with state transition.", groupId, topicIdPartition, memberId);
            }

            // Update the cached state and start and end offsets after releasing the acquisition lock on timeout.
            maybeUpdateCachedStateAndOffsets();
        } finally {
            lock.writeLock().unlock();
        }
    }
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