@Test
    public void testCanReprocessSkippedRecords() {
        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 tpId3 = new TopicIdPartition(topicId, new TopicPartition("sample", 3));
        assertEquals(partitioner.metadataPartition(tpId0), partitioner.metadataPartition(tpId1));
        assertNotEquals(partitioner.metadataPartition(tpId3), partitioner.metadataPartition(tpId0));

        final int metadataPartition = partitioner.metadataPartition(tpId0);
        final int anotherMetadataPartition = partitioner.metadataPartition(tpId3);

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

        // Adding metadata records in the order opposite to the order of assignments
        addRecord(consumer, metadataPartition, tpId1, 0);
        addRecord(consumer, metadataPartition, tpId0, 1);
        consumerTask.ingestRecords();
        assertEquals(Optional.of(1L), consumerTask.readOffsetForMetadataPartition(metadataPartition));
        // Only one record is processed, tpId1 record is skipped as unassigned
        // but read offset is 1 e.g., record for tpId1 has been read by consumer
        assertEquals(1, handler.metadataCounter);

        // Adding assignment for tpId1 after related metadata records have already been read
        consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId1));
        consumerTask.ingestRecords();
        assertTrue(consumerTask.isUserPartitionAssigned(tpId1), "Partition " + tpId1 + " has not been assigned");

        // Adding assignment for tpId0 to trigger the reset to last read offset
        // and assignment for tpId3 that has different metadata partition to trigger the update of metadata snapshot
        HashSet<TopicIdPartition> partitions = new HashSet<>();
        partitions.add(tpId0);
        partitions.add(tpId3);
        consumerTask.addAssignmentsForPartitions(partitions);
        // explicitly re-adding the records since MockConsumer drops them on poll.
        addRecord(consumer, metadataPartition, tpId1, 0);
        addRecord(consumer, metadataPartition, tpId0, 1);
        consumerTask.ingestRecords();
        // Waiting for all metadata records to be re-read from the first metadata partition number
        assertEquals(Optional.of(1L), consumerTask.readOffsetForMetadataPartition(metadataPartition));
        // Verifying that all the metadata records from the first metadata partition were processed properly.
        assertEquals(2, 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