boolean canAcquireRecords() {
        if (nextFetchOffset() != endOffset() + 1) {
            return true;
        }

        lock.readLock().lock();
        long numRecords;
        try {
            if (cachedState.isEmpty()) {
                numRecords = 0;
            } else {
                numRecords = this.endOffset - this.startOffset + 1;
            }
        } finally {
            lock.readLock().unlock();
        }
        return numRecords < maxInFlightMessages;
    }
@Test
    public void testCanAcquireRecordsWithEmptyCache() {
        SharePartition sharePartition = SharePartitionBuilder.builder().withMaxInflightMessages(1).build();
        assertTrue(sharePartition.canAcquireRecords());
    }