@Test
    public void testCanProcessRecord() {
        final Uuid topicId = Uuid.fromString("Bp9TDduJRGa9Q5rlvCJOxg");
        final TopicIdPartition tpId0 = new TopicIdPartition(topicId, new TopicPartition("sample", 0));
        final TopicIdPartition tpId1 = new TopicIdPartition(topicId, new TopicPartition("sample", 1));
        final TopicIdPartition tpId2 = new TopicIdPartition(topicId, new TopicPartition("sample", 2));
        assertEquals(partitioner.metadataPartition(tpId0), partitioner.metadataPartition(tpId1));
        assertEquals(partitioner.metadataPartition(tpId0), partitioner.metadataPartition(tpId2));

        final int metadataPartition = partitioner.metadataPartition(tpId0);
        consumer.updateEndOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), 0L));
        final Set<TopicIdPartition> assignments = Collections.singleton(tpId0);
        consumerTask.addAssignmentsForPartitions(assignments);
        consumerTask.ingestRecords();
        assertTrue(consumerTask.isUserPartitionAssigned(tpId0), "Partition " + tpId0 + " has not been assigned");

        addRecord(consumer, metadataPartition, tpId0, 0);
        addRecord(consumer, metadataPartition, tpId0, 1);
        consumerTask.ingestRecords();
        assertEquals(Optional.of(1L), consumerTask.readOffsetForMetadataPartition(metadataPartition), "Partition offset did not reach expected value of 1");
        assertEquals(2, handler.metadataCounter);

        // should only read the tpId1 records
        consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId1));
        consumerTask.ingestRecords();
        assertTrue(consumerTask.isUserPartitionAssigned(tpId1), "Partition " + tpId1 + " has not been assigned");

        addRecord(consumer, metadataPartition, tpId1, 2);
        consumerTask.ingestRecords();
        assertEquals(Optional.of(2L), consumerTask.readOffsetForMetadataPartition(metadataPartition), "Partition offset did not reach expected value of 2");
        assertEquals(3, handler.metadataCounter);

        // shouldn't read tpId2 records because it's not assigned
        addRecord(consumer, metadataPartition, tpId2, 3);
        consumerTask.ingestRecords();
        assertEquals(Optional.of(3L), consumerTask.readOffsetForMetadataPartition(metadataPartition), "Partition offset did not reach expected value of 3");
        assertEquals(3, handler.metadataCounter);
    }
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