@ParameterizedTest
    @CsvSource(value = {"0, 0", "500, 500"})
    public void testMaybeMarkUserPartitionAsReadyWhenTopicIsEmpty(long beginOffset, long endOffset) {
        final TopicIdPartition tpId = getIdPartitions("world", 1).get(0);
        final int metadataPartition = partitioner.metadataPartition(tpId);
        consumer.updateBeginningOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), beginOffset));
        consumer.updateEndOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), endOffset));
        consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId));
        consumerTask.ingestRecords();

        assertTrue(consumerTask.isUserPartitionAssigned(tpId), "Partition " + tpId + " has not been assigned");
        assertTrue(consumerTask.isMetadataPartitionAssigned(metadataPartition));
        assertTrue(handler.isPartitionInitialized.containsKey(tpId), "Should have initialized the partition");
        assertFalse(consumerTask.readOffsetForMetadataPartition(metadataPartition).isPresent());
    }
Tests that the consumer task shuts down gracefully when there were no assignments.
storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTaskTest.java