/**
 * 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.pulsar.broker.service;

import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.fail;

import java.util.LinkedHashSet;
import java.util.concurrent.TimeUnit;

import lombok.Cleanup;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundles;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import org.testng.collections.Lists;

import com.google.common.collect.Sets;

@Test(groups = "broker")
public class PeerReplicatorTest extends ReplicatorTestBase {

    @Override
    @BeforeClass(timeOut = 300000)
    public void setup() throws Exception {
        super.setup();
    }

    @Override
    @AfterClass(alwaysRun = true, timeOut = 300000)
    public void cleanup() throws Exception {
        super.cleanup();
    }

    @DataProvider(name = "lookupType")
    public Object[][] codecProvider() {
        return new Object[][] { { "http" }, { "binary" } };
    }

    /**
     * It verifies that lookup/admin requests for global-namespace would be redirected to peer-cluster if local cluster
     * doesn't own it and peer-cluster owns it, else request will be failed.
     * <pre>
     * 1. Create global-namespace ns1 for replication cluster-r1
     * 2. Try to create producer using broker in cluster r3
     * 3. Reject lookup: "r3" receives request and doesn't find namespace in local/peer cluster
     * 4. Add "r1" as a peer-cluster into "r3"
     * 5. Try to create producer using broker in cluster r3
     * 6. Success : "r3" finds "r1" in peer cluster which owns n1 and redirects to "r1"
     * 7. call admin-api to "r3" which redirects request to "r1"
     *
     * </pre>
     *
     * @param protocol
     * @throws Exception
     */
    @Test(dataProvider = "lookupType", timeOut = 10000)
    public void testPeerClusterTopicLookup(String protocol) throws Exception {

     // clean up peer-clusters
        admin1.clusters().updatePeerClusterNames("r1", null);
        admin1.clusters().updatePeerClusterNames("r2", null);
        admin1.clusters().updatePeerClusterNames("r3", null);

        final String serviceUrl = protocol.equalsIgnoreCase("http") ? pulsar3.getWebServiceAddress()
                : pulsar3.getBrokerServiceUrl();
        final String namespace1 = "pulsar/global/peer1-" + protocol;
        final String namespace2 = "pulsar/global/peer2-" + protocol;
        admin1.namespaces().createNamespace(namespace1);
        admin1.namespaces().createNamespace(namespace2);
        // add replication cluster
        admin1.namespaces().setNamespaceReplicationClusters(namespace1, Sets.newHashSet("r1"));
        admin1.namespaces().setNamespaceReplicationClusters(namespace2, Sets.newHashSet("r2"));
        admin1.clusters().updatePeerClusterNames("r3", null);
        // disable tls as redirection url is prepared according tls configuration
        pulsar1.getConfiguration().setTlsEnabled(false);
        pulsar2.getConfiguration().setTlsEnabled(false);
        pulsar3.getConfiguration().setTlsEnabled(false);

        final String topic1 = "persistent://" + namespace1 + "/topic1";
        final String topic2 = "persistent://" + namespace2 + "/topic2";

        @Cleanup
        PulsarClient client3 = PulsarClient.builder().serviceUrl(serviceUrl).statsInterval(0, TimeUnit.SECONDS)
            .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
        try {
            // try to create producer for topic1 (part of cluster: r1) by calling cluster: r3
            client3.newProducer().topic(topic1).create();
            fail("should have failed as cluster:r3 doesn't own namespace");
        } catch (PulsarClientException e) {
            // Ok
        }

        try {
            // try to create producer for topic2 (part of cluster: r2) by calling cluster: r3
            client3.newProducer().topic(topic2).create();
            fail("should have failed as cluster:r3 doesn't own namespace");
        } catch (PulsarClientException e) {
            // Ok
        }

        // set peer-clusters : r3->r1
        admin1.clusters().updatePeerClusterNames("r3", Sets.newLinkedHashSet(Lists.newArrayList("r1")));
        Producer<byte[]> producer = client3.newProducer().topic(topic1).create();
        PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getOrCreateTopic(topic1).get();
        assertNotNull(topic);
        pulsar1.getBrokerService().updateRates();
        // get stats for topic1 using cluster-r3's admin3
        TopicStats stats = admin1.topics().getStats(topic1);
        assertNotNull(stats);
        assertEquals(stats.getPublishers().size(), 1);
        stats = admin3.topics().getStats(topic1);
        assertNotNull(stats);
        assertEquals(stats.getPublishers().size(), 1);
        producer.close();

        // set peer-clusters : r3->r2
        admin2.clusters().updatePeerClusterNames("r3", Sets.newLinkedHashSet(Lists.newArrayList("r2")));
        producer = client3.newProducer().topic(topic2).create();
        topic = (PersistentTopic) pulsar2.getBrokerService().getOrCreateTopic(topic2).get();
        assertNotNull(topic);
        pulsar2.getBrokerService().updateRates();
        // get stats for topic1 using cluster-r3's admin3
        stats = admin3.topics().getStats(topic2);
        assertNotNull(stats);
        assertEquals(stats.getPublishers().size(), 1);
        stats = admin3.topics().getStats(topic2);
        assertNotNull(stats);
        assertEquals(stats.getPublishers().size(), 1);
        producer.close();


    }

    @Test(timeOut = 10000)
    public void testGetPeerClusters() throws Exception {

        // clean up peer-clusters
        admin1.clusters().updatePeerClusterNames("r1", null);
        admin1.clusters().updatePeerClusterNames("r2", null);
        admin1.clusters().updatePeerClusterNames("r3", null);

        final String mainClusterName = "r1";
        assertNull(admin1.clusters().getPeerClusterNames(mainClusterName));
        LinkedHashSet<String> peerClusters = Sets.newLinkedHashSet(Lists.newArrayList("r2", "r3"));
        admin1.clusters().updatePeerClusterNames(mainClusterName, peerClusters);
        retryStrategically((test) -> {
            try {
                return admin1.clusters().getPeerClusterNames(mainClusterName).size() == 1;
            } catch (PulsarAdminException e) {
                return false;
            }
        }, 5, 100);
        assertEquals(admin1.clusters().getPeerClusterNames(mainClusterName), peerClusters);
    }

    /**
     * Removing local cluster from the replication-cluster should make sure that bundle should not be loaded by the
     * cluster even if owner broker doesn't receive the watch to avoid lookup-conflict between peer-cluster.
     *
     * @throws Exception
     */
    @Test(groups = "broker")
    public void testPeerClusterInReplicationClusterListChange() throws Exception {

        // clean up peer-clusters
        admin1.clusters().updatePeerClusterNames("r1", null);
        admin1.clusters().updatePeerClusterNames("r2", null);
        admin1.clusters().updatePeerClusterNames("r3", null);

        final String serviceUrl = pulsar3.getBrokerServiceUrl();
        final String namespace1 = BrokerTestUtil.newUniqueName("pulsar/global/peer-change-repl-ns");
        admin1.namespaces().createNamespace(namespace1);
        // add replication cluster
        admin1.namespaces().setNamespaceReplicationClusters(namespace1, Sets.newHashSet("r1"));
        admin1.clusters().updatePeerClusterNames("r3", null);
        // disable tls as redirection url is prepared according tls configuration
        pulsar1.getConfiguration().setTlsEnabled(false);
        pulsar2.getConfiguration().setTlsEnabled(false);
        pulsar3.getConfiguration().setTlsEnabled(false);

        final String topic1 = "persistent://" + namespace1 + "/topic1";

        @Cleanup
        PulsarClient client3 = PulsarClient.builder().serviceUrl(serviceUrl).statsInterval(0, TimeUnit.SECONDS).build();
        // set peer-clusters : r3->r1
        admin1.clusters().updatePeerClusterNames("r3", Sets.newLinkedHashSet(Lists.newArrayList("r1")));
        admin1.clusters().updatePeerClusterNames("r1", Sets.newLinkedHashSet(Lists.newArrayList("r3")));
        Producer<byte[]> producer = client3.newProducer().topic(topic1).create();
        PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getOrCreateTopic(topic1).get();
        assertNotNull(topic);
        pulsar1.getBrokerService().updateRates();
        // get stats for topic1 using cluster-r3's admin3
        TopicStats stats = admin1.topics().getStats(topic1);
        assertNotNull(stats);
        assertEquals(stats.getPublishers().size(), 1);
        stats = admin3.topics().getStats(topic1);
        assertNotNull(stats);
        assertEquals(stats.getPublishers().size(), 1);
        producer.close();

        // change the repl cluster to peer-cluster r3 from r1
        admin1.namespaces().setNamespaceReplicationClusters(namespace1, Sets.newHashSet("r3"));
        NamespaceBundles bundles = pulsar1.getNamespaceService().getNamespaceBundleFactory()
                .getBundles(NamespaceName.get(namespace1));
        NamespaceBundle bundle = bundles.getBundles().get(0);
        retryStrategically((test) -> {
            try {
                return !pulsar1.getNamespaceService().isNamespaceBundleOwned(bundle).get();
            } catch (Exception e) {
                return false;
            }
        }, 5, 200);

        assertFalse(pulsar1.getNamespaceService().isNamespaceBundleOwned(bundle).get());
        // topic should be unloaded from broker1
        assertFalse(pulsar1.getBrokerService().getTopics().containsKey(topic1));

    }

}