@Test
    public void testRemoveAssignmentsForPartitions() {
        final List<TopicIdPartition> allPartitions = getIdPartitions("sample", 3);
        final Map<TopicPartition, Long> endOffsets = allPartitions.stream()
            .map(idp -> toRemoteLogPartition(partitioner.metadataPartition(idp)))
            .collect(Collectors.toMap(Function.identity(), e -> 0L, (a, b) -> b));
        consumer.updateEndOffsets(endOffsets);
        consumerTask.addAssignmentsForPartitions(new HashSet<>(allPartitions));
        consumerTask.ingestRecords();

        final TopicIdPartition tpId = allPartitions.get(0);
        assertTrue(consumerTask.isUserPartitionAssigned(tpId), "Partition " + tpId + " has not been assigned");
        addRecord(consumer, partitioner.metadataPartition(tpId), tpId, 0);
        consumerTask.ingestRecords();
        assertTrue(consumerTask.readOffsetForMetadataPartition(partitioner.metadataPartition(tpId)).isPresent());

        final Set<TopicIdPartition> removePartitions = Collections.singleton(tpId);
        consumerTask.removeAssignmentsForPartitions(removePartitions);
        consumerTask.ingestRecords();
        for (final TopicIdPartition idPartition : allPartitions) {
            assertEquals(!removePartitions.contains(idPartition), consumerTask.isUserPartitionAssigned(idPartition),
                    "Partition " + idPartition + " has not been removed");
        }
        for (TopicIdPartition removePartition : removePartitions) {
            assertTrue(handler.isPartitionCleared.containsKey(removePartition),
                    "Partition " + removePartition + " has not been cleared");
        }
    }
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