/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.clients;

import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.internals.ClusterResourceListeners;
import org.apache.kafka.common.internals.Topic;
import org.apache.kafka.common.message.MetadataResponseData;
import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseBroker;
import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseBrokerCollection;
import org.apache.kafka.common.message.MetadataResponseData.MetadataResponsePartition;
import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic;
import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopicCollection;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.test.MockClusterResourceListener;
import org.apache.kafka.test.TestUtils;
import org.junit.Test;

import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

import static org.apache.kafka.test.TestUtils.assertOptional;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;

public class MetadataTest {

    private long refreshBackoffMs = 100;
    private long metadataExpireMs = 1000;
    private Metadata metadata = new Metadata(refreshBackoffMs, metadataExpireMs, new LogContext(),
            new ClusterResourceListeners());

    private static MetadataResponse emptyMetadataResponse() {
        return MetadataResponse.prepareResponse(
                Collections.emptyList(),
                null,
                -1,
                Collections.emptyList());
    }

    @Test(expected = IllegalStateException.class)
    public void testMetadataUpdateAfterClose() {
        metadata.close();
        metadata.update(emptyMetadataResponse(), 1000);
    }

    private static void checkTimeToNextUpdate(long refreshBackoffMs, long metadataExpireMs) {
        long now = 10000;

        // Metadata timeToNextUpdate is implicitly relying on the premise that the currentTimeMillis is always
        // larger than the metadataExpireMs or refreshBackoffMs.
        // It won't be a problem practically since all usages of Metadata calls first update() immediately after
        // it's construction.
        if (metadataExpireMs > now || refreshBackoffMs > now) {
            throw new IllegalArgumentException(
                    "metadataExpireMs and refreshBackoffMs must be smaller than 'now'");
        }

        long largerOfBackoffAndExpire = Math.max(refreshBackoffMs, metadataExpireMs);
        Metadata metadata = new Metadata(refreshBackoffMs, metadataExpireMs, new LogContext(),
                new ClusterResourceListeners());

        assertEquals(0, metadata.timeToNextUpdate(now));

        // lastSuccessfulRefreshMs updated to now.
        metadata.update(emptyMetadataResponse(), now);

        // The last update was successful so the remaining time to expire the current metadata should be returned.
        assertEquals(largerOfBackoffAndExpire, metadata.timeToNextUpdate(now));

        // Metadata update requested explicitly
        metadata.requestUpdate();
        // Update requested so metadataExpireMs should no longer take effect.
        assertEquals(refreshBackoffMs, metadata.timeToNextUpdate(now));

        // Reset needUpdate to false.
        metadata.update(emptyMetadataResponse(), now);
        assertEquals(largerOfBackoffAndExpire, metadata.timeToNextUpdate(now));

        // Both metadataExpireMs and refreshBackoffMs elapsed.
        now += largerOfBackoffAndExpire;
        assertEquals(0, metadata.timeToNextUpdate(now));
        assertEquals(0, metadata.timeToNextUpdate(now + 1));
    }

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

        Metadata metadata = new Metadata(refreshBackoffMs, metadataExpireMs, new LogContext(),
                new ClusterResourceListeners());
        metadata.bootstrap(Collections.singletonList(new InetSocketAddress("localhost", 9002)));

        assertEquals(0, metadata.timeToAllowUpdate(time.milliseconds()));
        assertEquals(0, metadata.timeToNextUpdate(time.milliseconds()));
    }

    @Test
    public void testTimeToNextUpdate() {
        checkTimeToNextUpdate(100, 1000);
        checkTimeToNextUpdate(1000, 100);
        checkTimeToNextUpdate(0, 0);
        checkTimeToNextUpdate(0, 100);
        checkTimeToNextUpdate(100, 0);
    }

    @Test
    public void testTimeToNextUpdate_RetryBackoff() {
        long now = 10000;

        // lastRefreshMs updated to now.
        metadata.failedUpdate(now);

        // Backing off. Remaining time until next try should be returned.
        assertEquals(refreshBackoffMs, metadata.timeToNextUpdate(now));

        // Even though metadata update requested explicitly, still respects backoff.
        metadata.requestUpdate();
        assertEquals(refreshBackoffMs, metadata.timeToNextUpdate(now));

        // refreshBackoffMs elapsed.
        now += refreshBackoffMs;
        // It should return 0 to let next try.
        assertEquals(0, metadata.timeToNextUpdate(now));
        assertEquals(0, metadata.timeToNextUpdate(now + 1));
    }

    /**
     * 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.
     */
    @Test
    public void testIgnoreLeaderEpochInOlderMetadataResponse() {
        TopicPartition tp = new TopicPartition("topic", 0);

        MetadataResponsePartition partitionMetadata = new MetadataResponsePartition()
                .setPartitionIndex(tp.partition())
                .setLeaderId(5)
                .setLeaderEpoch(10)
                .setReplicaNodes(Arrays.asList(1, 2, 3))
                .setIsrNodes(Arrays.asList(1, 2, 3))
                .setOfflineReplicas(Collections.emptyList())
                .setErrorCode(Errors.NONE.code());

        MetadataResponseTopic topicMetadata = new MetadataResponseTopic()
                .setName(tp.topic())
                .setErrorCode(Errors.NONE.code())
                .setPartitions(Collections.singletonList(partitionMetadata))
                .setIsInternal(false);

        MetadataResponseTopicCollection topics = new MetadataResponseTopicCollection();
        topics.add(topicMetadata);

        MetadataResponseData data = new MetadataResponseData()
                .setClusterId("clusterId")
                .setControllerId(0)
                .setTopics(topics)
                .setBrokers(new MetadataResponseBrokerCollection());

        for (short version = ApiKeys.METADATA.oldestVersion(); version < 9; version++) {
            Struct struct = data.toStruct(version);
            MetadataResponse response = new MetadataResponse(struct, version);
            assertFalse(response.hasReliableLeaderEpochs());
            metadata.update(response, 100);
            assertTrue(metadata.partitionInfoIfCurrent(tp).isPresent());
            MetadataCache.PartitionInfoAndEpoch info = metadata.partitionInfoIfCurrent(tp).get();
            assertEquals(-1, info.epoch());
        }

        for (short version = 9; version <= ApiKeys.METADATA.latestVersion(); version++) {
            Struct struct = data.toStruct(version);
            MetadataResponse response = new MetadataResponse(struct, version);
            assertTrue(response.hasReliableLeaderEpochs());
            metadata.update(response, 100);
            assertTrue(metadata.partitionInfoIfCurrent(tp).isPresent());
            MetadataCache.PartitionInfoAndEpoch info = metadata.partitionInfoIfCurrent(tp).get();
            assertEquals(10, info.epoch());
        }
    }

    @Test
    public void testStaleMetadata() {
        TopicPartition tp = new TopicPartition("topic", 0);

        MetadataResponsePartition partitionMetadata = new MetadataResponsePartition()
                .setPartitionIndex(tp.partition())
                .setLeaderId(1)
                .setLeaderEpoch(10)
                .setReplicaNodes(Arrays.asList(1, 2, 3))
                .setIsrNodes(Arrays.asList(1, 2, 3))
                .setOfflineReplicas(Collections.emptyList())
                .setErrorCode(Errors.NONE.code());

        MetadataResponseTopic topicMetadata = new MetadataResponseTopic()
                .setName(tp.topic())
                .setErrorCode(Errors.NONE.code())
                .setPartitions(Collections.singletonList(partitionMetadata))
                .setIsInternal(false);

        MetadataResponseTopicCollection topics = new MetadataResponseTopicCollection();
        topics.add(topicMetadata);

        MetadataResponseData data = new MetadataResponseData()
                .setClusterId("clusterId")
                .setControllerId(0)
                .setTopics(topics)
                .setBrokers(new MetadataResponseBrokerCollection());

        metadata.update(new MetadataResponse(data), 100);

        // Older epoch with changed ISR should be ignored
        partitionMetadata
                .setPartitionIndex(tp.partition())
                .setLeaderId(1)
                .setLeaderEpoch(9)
                .setReplicaNodes(Arrays.asList(1, 2, 3))
                .setIsrNodes(Arrays.asList(1, 2))
                .setOfflineReplicas(Collections.emptyList())
                .setErrorCode(Errors.NONE.code());

        metadata.update(new MetadataResponse(data), 101);
        assertEquals(Optional.of(10), metadata.lastSeenLeaderEpoch(tp));

        assertTrue(metadata.partitionInfoIfCurrent(tp).isPresent());
        MetadataCache.PartitionInfoAndEpoch info = metadata.partitionInfoIfCurrent(tp).get();

        List<Integer> cachedIsr = Arrays.stream(info.partitionInfo().inSyncReplicas())
                .map(Node::id).collect(Collectors.toList());
        assertEquals(Arrays.asList(1, 2, 3), cachedIsr);
        assertEquals(10, info.epoch());
    }

    @Test
    public void testFailedUpdate() {
        long time = 100;
        metadata.update(emptyMetadataResponse(), time);

        assertEquals(100, metadata.timeToNextUpdate(1000));
        metadata.failedUpdate(1100);

        assertEquals(100, metadata.timeToNextUpdate(1100));
        assertEquals(100, metadata.lastSuccessfulUpdate());

        metadata.update(emptyMetadataResponse(), time);
        assertEquals(100, metadata.timeToNextUpdate(1000));
    }

    @Test
    public void testClusterListenerGetsNotifiedOfUpdate() {
        MockClusterResourceListener mockClusterListener = new MockClusterResourceListener();
        ClusterResourceListeners listeners = new ClusterResourceListeners();
        listeners.maybeAdd(mockClusterListener);
        metadata = new Metadata(refreshBackoffMs, metadataExpireMs, new LogContext(), listeners);

        String hostName = "www.example.com";
        metadata.bootstrap(Collections.singletonList(new InetSocketAddress(hostName, 9002)));
        assertFalse("ClusterResourceListener should not called when metadata is updated with bootstrap Cluster",
                MockClusterResourceListener.IS_ON_UPDATE_CALLED.get());

        Map<String, Integer> partitionCounts = new HashMap<>();
        partitionCounts.put("topic", 1);
        partitionCounts.put("topic1", 1);
        MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, partitionCounts);
        metadata.update(metadataResponse, 100);

        assertEquals("MockClusterResourceListener did not get cluster metadata correctly",
                "dummy", mockClusterListener.clusterResource().clusterId());
        assertTrue("MockClusterResourceListener should be called when metadata is updated with non-bootstrap Cluster",
                MockClusterResourceListener.IS_ON_UPDATE_CALLED.get());
    }


    @Test
    public void testRequestUpdate() {
        assertFalse(metadata.updateRequested());

        int[] epochs =           {42,   42,    41,    41,    42,    43,   43,    42,    41,    44};
        boolean[] updateResult = {true, false, false, false, false, true, false, false, false, true};
        TopicPartition tp = new TopicPartition("topic", 0);

        for (int i = 0; i < epochs.length; i++) {
            metadata.updateLastSeenEpochIfNewer(tp, epochs[i]);
            if (updateResult[i]) {
                assertTrue("Expected metadata update to be requested [" + i + "]", metadata.updateRequested());
            } else {
                assertFalse("Did not expect metadata update to be requested [" + i + "]", metadata.updateRequested());
            }
            metadata.update(emptyMetadataResponse(), 0L);
            assertFalse(metadata.updateRequested());
        }
    }

    @Test
    public void testRejectOldMetadata() {
        Map<String, Integer> partitionCounts = new HashMap<>();
        partitionCounts.put("topic-1", 1);
        TopicPartition tp = new TopicPartition("topic-1", 0);

        metadata.update(emptyMetadataResponse(), 0L);

        // First epoch seen, accept it
        {
            MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 100);
            metadata.update(metadataResponse, 10L);
            assertNotNull(metadata.fetch().partition(tp));
            assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
        }

        // Fake an empty ISR, but with an older epoch, should reject it
        {
            MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 99,
                (error, partition, leader, leaderEpoch, replicas, isr, offlineReplicas) ->
                        new MetadataResponse.PartitionMetadata(error, partition, leader, leaderEpoch, replicas, Collections.emptyList(), offlineReplicas));
            metadata.update(metadataResponse, 20L);
            assertEquals(metadata.fetch().partition(tp).inSyncReplicas().length, 1);
            assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
        }

        // Fake an empty ISR, with same epoch, accept it
        {
            MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 100,
                (error, partition, leader, leaderEpoch, replicas, isr, offlineReplicas) ->
                        new MetadataResponse.PartitionMetadata(error, partition, leader, leaderEpoch, replicas, Collections.emptyList(), offlineReplicas));
            metadata.update(metadataResponse, 20L);
            assertEquals(metadata.fetch().partition(tp).inSyncReplicas().length, 0);
            assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
        }

        // Empty metadata response, should not keep old partition but should keep the last-seen epoch
        {
            MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.emptyMap());
            metadata.update(metadataResponse, 20L);
            assertNull(metadata.fetch().partition(tp));
            assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
        }

        // Back in the metadata, with old epoch, should not get added
        {
            MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 99);
            metadata.update(metadataResponse, 10L);
            assertNull(metadata.fetch().partition(tp));
            assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
        }
    }

    @Test
    public void testMaybeRequestUpdate() {
        TopicPartition tp = new TopicPartition("topic-1", 0);
        metadata.update(emptyMetadataResponse(), 0L);
        assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 1));
        assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 1);

        metadata.update(emptyMetadataResponse(), 1L);
        assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 1));
        assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 1);

        metadata.update(emptyMetadataResponse(), 2L);
        assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 0));
        assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 1);

        metadata.update(emptyMetadataResponse(), 3L);
        assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 2));
        assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 2);
    }

    @Test
    public void testOutOfBandEpochUpdate() {
        Map<String, Integer> partitionCounts = new HashMap<>();
        partitionCounts.put("topic-1", 5);
        TopicPartition tp = new TopicPartition("topic-1", 0);

        metadata.update(emptyMetadataResponse(), 0L);

        assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 99));

        // Update epoch to 100
        MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 100);
        metadata.update(metadataResponse, 10L);
        assertNotNull(metadata.fetch().partition(tp));
        assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);

        // Simulate a leader epoch from another response, like a fetch response (not yet implemented)
        assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 101));

        // Cache of partition stays, but current partition info is not available since it's stale
        assertNotNull(metadata.fetch().partition(tp));
        assertEquals(metadata.fetch().partitionCountForTopic("topic-1").longValue(), 5);
        assertFalse(metadata.partitionInfoIfCurrent(tp).isPresent());
        assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 101);

        // Metadata with older epoch is rejected, metadata state is unchanged
        metadata.update(metadataResponse, 20L);
        assertNotNull(metadata.fetch().partition(tp));
        assertEquals(metadata.fetch().partitionCountForTopic("topic-1").longValue(), 5);
        assertFalse(metadata.partitionInfoIfCurrent(tp).isPresent());
        assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 101);

        // Metadata with equal or newer epoch is accepted
        metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 101);
        metadata.update(metadataResponse, 30L);
        assertNotNull(metadata.fetch().partition(tp));
        assertEquals(metadata.fetch().partitionCountForTopic("topic-1").longValue(), 5);
        assertTrue(metadata.partitionInfoIfCurrent(tp).isPresent());
        assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 101);
    }

    @Test
    public void testNoEpoch() {
        metadata.update(emptyMetadataResponse(), 0L);
        MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1));
        metadata.update(metadataResponse, 10L);

        TopicPartition tp = new TopicPartition("topic-1", 0);

        // no epoch
        assertFalse(metadata.lastSeenLeaderEpoch(tp).isPresent());

        // still works
        assertTrue(metadata.partitionInfoIfCurrent(tp).isPresent());
        assertEquals(metadata.partitionInfoIfCurrent(tp).get().partitionInfo().partition(), 0);
        assertEquals(metadata.partitionInfoIfCurrent(tp).get().partitionInfo().leader().id(), 0);
    }

    @Test
    public void testClusterCopy() {
        Map<String, Integer> counts = new HashMap<>();
        Map<String, Errors> errors = new HashMap<>();
        counts.put("topic1", 2);
        counts.put("topic2", 3);
        counts.put(Topic.GROUP_METADATA_TOPIC_NAME, 3);
        errors.put("topic3", Errors.INVALID_TOPIC_EXCEPTION);
        errors.put("topic4", Errors.TOPIC_AUTHORIZATION_FAILED);

        MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 4, errors, counts);
        metadata.update(metadataResponse, 0L);

        Cluster cluster = metadata.fetch();
        assertEquals(cluster.clusterResource().clusterId(), "dummy");
        assertEquals(cluster.nodes().size(), 4);

        // topic counts
        assertEquals(cluster.invalidTopics(), Collections.singleton("topic3"));
        assertEquals(cluster.unauthorizedTopics(), Collections.singleton("topic4"));
        assertEquals(cluster.topics().size(), 3);
        assertEquals(cluster.internalTopics(), Collections.singleton(Topic.GROUP_METADATA_TOPIC_NAME));

        // partition counts
        assertEquals(cluster.partitionsForTopic("topic1").size(), 2);
        assertEquals(cluster.partitionsForTopic("topic2").size(), 3);

        // Sentinel instances
        InetSocketAddress address = InetSocketAddress.createUnresolved("localhost", 0);
        Cluster fromMetadata = MetadataCache.bootstrap(Collections.singletonList(address)).cluster();
        Cluster fromCluster = Cluster.bootstrap(Collections.singletonList(address));
        assertEquals(fromMetadata, fromCluster);

        Cluster fromMetadataEmpty = MetadataCache.empty().cluster();
        Cluster fromClusterEmpty = Cluster.empty();
        assertEquals(fromMetadataEmpty, fromClusterEmpty);
    }

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

        metadata.requestUpdate();
        Metadata.MetadataRequestAndVersion versionAndBuilder = metadata.newMetadataRequestAndVersion();
        metadata.update(versionAndBuilder.requestVersion,
                TestUtils.metadataUpdateWith(1, Collections.singletonMap("topic", 1)), time.milliseconds());
        assertFalse(metadata.updateRequested());

        // bump the request version for new topics added to the metadata
        metadata.requestUpdateForNewTopics();

        // simulating a bump while a metadata request is in flight
        versionAndBuilder = metadata.newMetadataRequestAndVersion();
        metadata.requestUpdateForNewTopics();
        metadata.update(versionAndBuilder.requestVersion,
                TestUtils.metadataUpdateWith(1, Collections.singletonMap("topic", 1)), time.milliseconds());

        // metadata update is still needed
        assertTrue(metadata.updateRequested());

        // the next update will resolve it
        versionAndBuilder = metadata.newMetadataRequestAndVersion();
        metadata.update(versionAndBuilder.requestVersion,
                TestUtils.metadataUpdateWith(1, Collections.singletonMap("topic", 1)), time.milliseconds());
        assertFalse(metadata.updateRequested());
    }

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

        String invalidTopic = "topic dfsa";
        MetadataResponse invalidTopicResponse = TestUtils.metadataUpdateWith("clusterId", 1,
                Collections.singletonMap(invalidTopic, Errors.INVALID_TOPIC_EXCEPTION), Collections.emptyMap());
        metadata.update(invalidTopicResponse, time.milliseconds());

        InvalidTopicException e = assertThrows(InvalidTopicException.class, () -> metadata.maybeThrowAnyException());

        assertEquals(Collections.singleton(invalidTopic), e.invalidTopics());
        // We clear the exception once it has been raised to the user
        metadata.maybeThrowAnyException();

        // Reset the invalid topic error
        metadata.update(invalidTopicResponse, time.milliseconds());

        // If we get a good update, the error should clear even if we haven't had a chance to raise it to the user
        metadata.update(emptyMetadataResponse(), time.milliseconds());
        metadata.maybeThrowAnyException();
    }

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

        String invalidTopic = "foo";
        MetadataResponse unauthorizedTopicResponse = TestUtils.metadataUpdateWith("clusterId", 1,
                Collections.singletonMap(invalidTopic, Errors.TOPIC_AUTHORIZATION_FAILED), Collections.emptyMap());
        metadata.update(unauthorizedTopicResponse, time.milliseconds());

        TopicAuthorizationException e = assertThrows(TopicAuthorizationException.class, () -> metadata.maybeThrowAnyException());
        assertEquals(Collections.singleton(invalidTopic), e.unauthorizedTopics());
        // We clear the exception once it has been raised to the user
        metadata.maybeThrowAnyException();

        // Reset the unauthorized topic error
        metadata.update(unauthorizedTopicResponse, time.milliseconds());

        // If we get a good update, the error should clear even if we haven't had a chance to raise it to the user
        metadata.update(emptyMetadataResponse(), time.milliseconds());
        metadata.maybeThrowAnyException();
    }

    @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 = TestUtils.metadataUpdateWith("clusterId", 1, topicErrors, Collections.emptyMap());

        metadata.update(metadataResponse, 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.update(metadataResponse, time.milliseconds());
        TopicAuthorizationException e2 = assertThrows(TopicAuthorizationException.class,
            () -> metadata.maybeThrowExceptionForTopic("sensitiveTopic2"));
        assertEquals(Collections.singleton("sensitiveTopic2"), e2.unauthorizedTopics());
        metadata.maybeThrowAnyException();

        metadata.update(metadataResponse, 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.update(metadataResponse, time.milliseconds());
        metadata.maybeThrowExceptionForTopic("anotherTopic");
        metadata.maybeThrowAnyException();
    }

    @Test
    public void testNodeIfOffline() {
        Map<String, Integer> partitionCounts = new HashMap<>();
        partitionCounts.put("topic-1", 1);
        Node node0 = new Node(0, "localhost", 9092);
        Node node1 = new Node(1, "localhost", 9093);

        MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 2, Collections.emptyMap(), partitionCounts, _tp -> 99,
            (error, partition, leader, leaderEpoch, replicas, isr, offlineReplicas) ->
                new MetadataResponse.PartitionMetadata(error, partition, node0, leaderEpoch,
                    Collections.singletonList(node0), Collections.emptyList(), Collections.singletonList(node1)));
        metadata.update(emptyMetadataResponse(), 0L);
        metadata.update(metadataResponse, 10L);

        TopicPartition tp = new TopicPartition("topic-1", 0);

        assertOptional(metadata.fetch().nodeIfOnline(tp, 0), node -> assertEquals(node.id(), 0));
        assertFalse(metadata.fetch().nodeIfOnline(tp, 1).isPresent());
        assertEquals(metadata.fetch().nodeById(0).id(), 0);
        assertEquals(metadata.fetch().nodeById(1).id(), 1);
    }

    @Test
    public void testLeaderMetadataInconsistentWithBrokerMetadata() {
        // Tests a reordering scenario which can lead to inconsistent leader state.
        // A partition initially has one broker offline. That broker comes online and
        // is elected leader. The client sees these two events in the opposite order.

        TopicPartition tp = new TopicPartition("topic", 0);

        Node node0 = new Node(0, "localhost", 9092);
        Node node1 = new Node(1, "localhost", 9093);
        Node node2 = new Node(2, "localhost", 9094);

        // The first metadata received by broker (epoch=10)
        MetadataResponsePartition firstPartitionMetadata = new MetadataResponsePartition()
                .setPartitionIndex(tp.partition())
                .setErrorCode(Errors.NONE.code())
                .setLeaderEpoch(10)
                .setLeaderId(0)
                .setReplicaNodes(Arrays.asList(0, 1, 2))
                .setIsrNodes(Arrays.asList(0, 1, 2))
                .setOfflineReplicas(Collections.emptyList());

        // The second metadata received has stale metadata (epoch=8)
        MetadataResponsePartition secondPartitionMetadata = new MetadataResponsePartition()
                .setPartitionIndex(tp.partition())
                .setErrorCode(Errors.NONE.code())
                .setLeaderEpoch(8)
                .setLeaderId(1)
                .setReplicaNodes(Arrays.asList(0, 1, 2))
                .setIsrNodes(Arrays.asList(1, 2))
                .setOfflineReplicas(Collections.singletonList(0));

        metadata.update(new MetadataResponse(new MetadataResponseData()
                        .setTopics(buildTopicCollection(tp.topic(), firstPartitionMetadata))
                        .setBrokers(buildBrokerCollection(Arrays.asList(node0, node1, node2)))),
                10L);

        metadata.update(new MetadataResponse(new MetadataResponseData()
                        .setTopics(buildTopicCollection(tp.topic(), secondPartitionMetadata))
                        .setBrokers(buildBrokerCollection(Arrays.asList(node1, node2)))),
                20L);

        assertNull(metadata.fetch().leaderFor(tp));
        assertEquals(Optional.of(10), metadata.lastSeenLeaderEpoch(tp));
        assertTrue(metadata.leaderAndEpoch(tp).leader.isEmpty());
    }

    private MetadataResponseTopicCollection buildTopicCollection(String topic, MetadataResponsePartition partitionMetadata) {
        MetadataResponseTopic topicMetadata = new MetadataResponseTopic()
                .setErrorCode(Errors.NONE.code())
                .setName(topic)
                .setIsInternal(false);

        topicMetadata.setPartitions(Collections.singletonList(partitionMetadata));

        MetadataResponseTopicCollection topics = new MetadataResponseTopicCollection();
        topics.add(topicMetadata);
        return topics;
    }

    private MetadataResponseBrokerCollection buildBrokerCollection(List<Node> nodes) {
        MetadataResponseBrokerCollection brokers = new MetadataResponseBrokerCollection();
        for (Node node : nodes) {
            MetadataResponseBroker broker = new MetadataResponseBroker()
                    .setNodeId(node.id())
                    .setHost(node.host())
                    .setPort(node.port())
                    .setRack(node.rack());
            brokers.add(broker);
        }
        return brokers;
    }

}
