private void ensureBlockOnRecord(long blockedOffset) {
        for (int i = 0; i < 2; i++) {
            // the fetchRecords() should always throw exception due to the invalid message at the starting offset.
            assertThrows(KafkaException.class, this::fetchRecords);
            assertEquals(blockedOffset, subscriptions.position(tp0).offset);
        }
    }
This validates that no records are fetched for partitions that are waiting for the
onPartitionsAssigned callback to complete. Note that this markPendingOnAssignedCallback
functionality is only used from the {@link AsyncKafkaConsumer}.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java