private static Stream<Arguments> listOffsetsMetadataNonRetriableErrors() {
        return Stream.of(
                Arguments.of(
                        Errors.TOPIC_AUTHORIZATION_FAILED,
                        Errors.TOPIC_AUTHORIZATION_FAILED,
                        TopicAuthorizationException.class
                ),
                Arguments.of(
                        // We fail fast when the entire topic is unknown...
                        Errors.UNKNOWN_TOPIC_OR_PARTITION,
                        Errors.NONE,
                        UnknownTopicOrPartitionException.class
                ),
                Arguments.of(
                        // ... even if a partition in the topic is also somehow reported as unknown...
                        Errors.UNKNOWN_TOPIC_OR_PARTITION,
                        Errors.UNKNOWN_TOPIC_OR_PARTITION,
                        UnknownTopicOrPartitionException.class
                ),
                Arguments.of(
                        // ... or a partition in the topic has a different, otherwise-retriable error
                        Errors.UNKNOWN_TOPIC_OR_PARTITION,
                        Errors.LEADER_NOT_AVAILABLE,
                        UnknownTopicOrPartitionException.class
                )
        );
    }
Test that we propagate exceptions encountered when fetching metadata.
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java