@Test
    public void testMetadataTopicErrors() {
        Time time = new MockTime();

        Map<String, Errors> topicErrors = new HashMap<>(3);
        topicErrors.put("invalidTopic", Errors.INVALID_TOPIC_EXCEPTION);
        topicErrors.put("sensitiveTopic1", Errors.TOPIC_AUTHORIZATION_FAILED);
        topicErrors.put("sensitiveTopic2", Errors.TOPIC_AUTHORIZATION_FAILED);
        MetadataResponse metadataResponse = RequestTestUtils.metadataUpdateWith("clusterId", 1, topicErrors, Collections.emptyMap());

        metadata.updateWithCurrentRequestVersion(metadataResponse, false, time.milliseconds());
        TopicAuthorizationException e1 = assertThrows(TopicAuthorizationException.class,
            () -> metadata.maybeThrowExceptionForTopic("sensitiveTopic1"));
        assertEquals(Collections.singleton("sensitiveTopic1"), e1.unauthorizedTopics());
        // We clear the exception once it has been raised to the user
        metadata.maybeThrowAnyException();

        metadata.updateWithCurrentRequestVersion(metadataResponse, false, time.milliseconds());
        TopicAuthorizationException e2 = assertThrows(TopicAuthorizationException.class,
            () -> metadata.maybeThrowExceptionForTopic("sensitiveTopic2"));
        assertEquals(Collections.singleton("sensitiveTopic2"), e2.unauthorizedTopics());
        metadata.maybeThrowAnyException();

        metadata.updateWithCurrentRequestVersion(metadataResponse, false, time.milliseconds());
        InvalidTopicException e3 = assertThrows(InvalidTopicException.class,
            () -> metadata.maybeThrowExceptionForTopic("invalidTopic"));
        assertEquals(Collections.singleton("invalidTopic"), e3.invalidTopics());
        metadata.maybeThrowAnyException();

        // Other topics should not throw exception, but they should clear existing exception
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, time.milliseconds());
        metadata.maybeThrowExceptionForTopic("anotherTopic");
        metadata.maybeThrowAnyException();
    }
Prior to Kafka version 2.4 (which coincides with Metadata version 9), the broker does not propagate leader epoch
information accurately while a reassignment is in progress, so we cannot rely on it. This is explained in more
detail in MetadataResponse's constructor.
clients/src/test/java/org/apache/kafka/clients/MetadataTest.java