@Test
    public void testConsumerShouldCloseOnNonRetriableError() {
        final TopicIdPartition tpId = getIdPartitions("world", 1).get(0);
        final int metadataPartition = partitioner.metadataPartition(tpId);
        consumer.updateEndOffsets(Collections.singletonMap(toRemoteLogPartition(metadataPartition), 1L));
        consumerTask.addAssignmentsForPartitions(Collections.singleton(tpId));
        consumerTask.ingestRecords();

        assertTrue(consumerTask.isUserPartitionAssigned(tpId), "Partition " + tpId + " has not been assigned");
        assertTrue(consumerTask.isMetadataPartitionAssigned(metadataPartition));

        consumer.setPollException(new AuthorizationException("Unauthorized to read the topic!"));
        // Due to the exception set up earlier, calling run() will trigger an exception and close the Consumer, instead of resulting in an infinite loop
        // The purpose of calling run() is to validate the capability of ConsumerTask to shut down automatically
        consumerTask.run();
        assertTrue(consumer.closed(), "Should close the consume on non-retriable error");
    }
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