private AcquiredRecords acquireNewBatchRecords(
        String memberId,
        long firstOffset,
        long lastOffset
    ) {
        lock.writeLock().lock();
        try {
            // Schedule acquisition lock timeout for the batch.
            AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstOffset, lastOffset);
            // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
            cachedState.put(firstOffset, new InFlightBatch(
                memberId,
                firstOffset,
                lastOffset,
                RecordState.ACQUIRED,
                1,
                timerTask));
            // if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
            if (cachedState.firstKey() == firstOffset)  {
                startOffset = firstOffset;
            }
            endOffset = lastOffset;
            return new AcquiredRecords()
                .setFirstOffset(firstOffset)
                .setLastOffset(lastOffset)
                .setDeliveryCount((short) 1);
        } finally {
            lock.writeLock().unlock();
        }
    }
Release the fetch lock once the records are fetched from the leader.
core/src/main/java/kafka/server/share/SharePartition.java