/*
 * 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.connect.mirror.integration;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewPartitions;
import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.acl.AccessControlEntryFilter;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.connect.mirror.MirrorMakerConfig;
import org.apache.kafka.connect.mirror.clients.admin.FakeForwardingAdminWithLocalMetadata;
import org.apache.kafka.connect.mirror.clients.admin.FakeLocalMetadataStore;
import org.apache.kafka.connect.util.clusters.EmbeddedKafkaCluster;
import org.apache.kafka.network.SocketServerConfigs;
import org.apache.kafka.server.config.KRaftConfigs;
import org.apache.kafka.server.config.ServerConfigs;

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;

import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.stream.Collectors;

import static org.apache.kafka.connect.mirror.MirrorConnectorConfig.FORWARDING_ADMIN_CLASS;
import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX;
import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX;
import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX;
import static org.apache.kafka.test.TestUtils.waitForCondition;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * Tests MM2 is using provided ForwardingAdmin to create/alter topics, partitions and ACLs.
 */
@Tag("integration")
public class MirrorConnectorsWithCustomForwardingAdminIntegrationTest extends MirrorConnectorsIntegrationBaseTest {

    private static final int TOPIC_ACL_SYNC_DURATION_MS = 30_000;
    private static final int FAKE_LOCAL_METADATA_STORE_SYNC_DURATION_MS = 60_000;

    /*
     * enable ACL on brokers.
     */
    protected static void enableAclAuthorizer(Properties brokerProps) {
        brokerProps.put(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, "CONTROLLER:SASL_PLAINTEXT,EXTERNAL:SASL_PLAINTEXT");
        brokerProps.put(ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG, "org.apache.kafka.metadata.authorizer.StandardAuthorizer");
        brokerProps.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, "PLAIN");
        brokerProps.put(BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG, "PLAIN");
        brokerProps.put(KRaftConfigs.SASL_MECHANISM_CONTROLLER_PROTOCOL_CONFIG, "PLAIN");
        String listenerSaslJaasConfig = "org.apache.kafka.common.security.plain.PlainLoginModule required "
                + "username=\"super\" "
                + "password=\"super_pwd\" "
                + "user_connector=\"connector_pwd\" "
                + "user_super=\"super_pwd\";";
        brokerProps.put("listener.name.external.plain.sasl.jaas.config", listenerSaslJaasConfig);
        brokerProps.put("listener.name.controller.plain.sasl.jaas.config", listenerSaslJaasConfig);
        brokerProps.put("super.users", "User:super");
    }

    /*
     * return superUser auth config.
     */
    protected static Map<String, String> superUserConfig() {
        Map<String, String> superUserClientConfig = new HashMap<>();
        superUserClientConfig.put("sasl.mechanism", "PLAIN");
        superUserClientConfig.put("security.protocol", "SASL_PLAINTEXT");
        superUserClientConfig.put("sasl.jaas.config",
                "org.apache.kafka.common.security.plain.PlainLoginModule required "
                        + "username=\"super\" "
                        + "password=\"super_pwd\";");
        return superUserClientConfig;
    }

    /*
     * return connect user auth config.
     */
    protected static Map<String, String> connectorUserConfig() {
        Map<String, String> connectUserClientConfig = new HashMap<>();
        connectUserClientConfig.put("sasl.mechanism", "PLAIN");
        connectUserClientConfig.put("security.protocol", "SASL_PLAINTEXT");
        connectUserClientConfig.put("sasl.jaas.config",
                "org.apache.kafka.common.security.plain.PlainLoginModule required "
                        + "username=\"connector\" "
                        + "password=\"connector_pwd\";");
        return connectUserClientConfig;
    }

    /*
     * delete all acls from the input kafka cluster
     */
    private static void deleteAllACLs(EmbeddedKafkaCluster cluster) throws Exception {
        try (final Admin adminClient = cluster.createAdminClient()) {
            Set<String> topicsToBeDeleted = adminClient.listTopics().names().get();
            List<AclBindingFilter> aclBindingFilters = topicsToBeDeleted.stream().map(topic -> new AclBindingFilter(
                            new ResourcePatternFilter(ResourceType.TOPIC, topic, PatternType.ANY),
                            AccessControlEntryFilter.ANY
                    )
            ).collect(Collectors.toList());
            adminClient.deleteAcls(aclBindingFilters);
        }
    }

    /*
     * retrieve the acl details based on the input cluster for given topic name.
     */
    protected static Collection<AclBinding> getAclBindings(EmbeddedKafkaCluster cluster, String topic) throws Exception {
        try (Admin client = cluster.createAdminClient()) {
            ResourcePatternFilter topicFilter = new ResourcePatternFilter(ResourceType.TOPIC,
                    topic, PatternType.ANY);
            return client.describeAcls(new AclBindingFilter(topicFilter, AccessControlEntryFilter.ANY)).values().get();
        }
    }

    @BeforeEach
    public void startClusters() throws Exception {
        enableAclAuthorizer(primaryBrokerProps);
        additionalPrimaryClusterClientsConfigs.putAll(superUserConfig());
        primaryWorkerProps.putAll(superUserConfig());

        enableAclAuthorizer(backupBrokerProps);
        additionalBackupClusterClientsConfigs.putAll(superUserConfig());
        backupWorkerProps.putAll(superUserConfig());

        Map<String, String> additionalConfig = new HashMap<>(superUserConfig()) {{
                put(FORWARDING_ADMIN_CLASS, FakeForwardingAdminWithLocalMetadata.class.getName());
            }};

        superUserConfig().forEach((property, value) -> {
            additionalConfig.put(CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + property, value);
            additionalConfig.put(CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX + property, value);
            additionalConfig.put("consumer." + property, value);
            additionalConfig.put("producer." + property, value);
        });

        connectorUserConfig().forEach((property, value) -> {
            additionalConfig.put("admin." + property, value);
            additionalConfig.put(CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX + property, value);
        });

        startClusters(additionalConfig);

        try (Admin adminClient = primary.kafka().createAdminClient()) {
            adminClient.createAcls(List.of(
                    new AclBinding(
                            new ResourcePattern(ResourceType.TOPIC, "*", PatternType.LITERAL),
                            new AccessControlEntry("User:connector", "*", AclOperation.ALL, AclPermissionType.ALLOW)
                    )
            )).all().get();
        }
        try (Admin adminClient = backup.kafka().createAdminClient()) {
            adminClient.createAcls(List.of(
                    new AclBinding(
                            new ResourcePattern(ResourceType.TOPIC, "*", PatternType.LITERAL),
                            new AccessControlEntry("User:connector", "*", AclOperation.ALL, AclPermissionType.ALLOW)
                    )
            )).all().get();
        }
    }

    @AfterEach
    public void shutdownClusters() throws Exception {
        deleteAllACLs(primary.kafka());
        deleteAllACLs(backup.kafka());
        FakeLocalMetadataStore.clear();
        super.shutdownClusters();
    }

    @Test
    public void testReplicationIsCreatingTopicsUsingProvidedForwardingAdmin() throws Exception {
        produceMessages(primaryProducer, "test-topic-1");
        produceMessages(backupProducer, "test-topic-1");
        String consumerGroupName = "consumer-group-testReplication";
        Map<String, Object> consumerProps = Map.of("group.id", consumerGroupName);
        // warm up consumers before starting the connectors so we don't need to wait for discovery
        warmUpConsumer(consumerProps);

        mm2Config = new MirrorMakerConfig(mm2Props);

        waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS);
        waitUntilMirrorMakerIsRunning(primary, CONNECTOR_LIST, mm2Config, BACKUP_CLUSTER_ALIAS, PRIMARY_CLUSTER_ALIAS);

        // make sure the topic is auto-created in the other cluster
        waitForTopicCreated(primary, "backup.test-topic-1");
        waitForTopicCreated(backup, "primary.test-topic-1");
        waitForTopicCreated(primary, "mm2-offset-syncs.backup.internal");
        waitForTopicCreated(primary, "backup.checkpoints.internal");
        waitForTopicCreated(primary, "backup.heartbeats");
        waitForTopicCreated(backup, "mm2-offset-syncs.primary.internal");
        waitForTopicCreated(backup, "primary.checkpoints.internal");
        waitForTopicCreated(backup, "primary.heartbeats");

        // expect to use FakeForwardingAdminWithLocalMetadata to create remote topics and internal topics into local store
        waitForTopicToPersistInFakeLocalMetadataStore("backup.test-topic-1");
        waitForTopicToPersistInFakeLocalMetadataStore("primary.test-topic-1");
        waitForTopicToPersistInFakeLocalMetadataStore("mm2-offset-syncs.backup.internal");
        waitForTopicToPersistInFakeLocalMetadataStore("backup.checkpoints.internal");
        waitForTopicToPersistInFakeLocalMetadataStore("backup.heartbeats");
        waitForTopicToPersistInFakeLocalMetadataStore("mm2-offset-syncs.primary.internal");
        waitForTopicToPersistInFakeLocalMetadataStore("primary.checkpoints.internal");
        waitForTopicToPersistInFakeLocalMetadataStore("primary.heartbeats");
        waitForTopicToPersistInFakeLocalMetadataStore("heartbeats");
    }

    @Test
    public void testCreatePartitionsUseProvidedForwardingAdmin() throws Exception {
        mm2Config = new MirrorMakerConfig(mm2Props);
        produceMessages(backupProducer, "test-topic-1");
        produceMessages(primaryProducer, "test-topic-1");
        String consumerGroupName = "consumer-group-testReplication";
        Map<String, Object> consumerProps = Map.of("group.id", consumerGroupName);
        // warm up consumers before starting the connectors so we don't need to wait for discovery
        warmUpConsumer(consumerProps);

        waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS);
        waitUntilMirrorMakerIsRunning(primary, CONNECTOR_LIST, mm2Config, BACKUP_CLUSTER_ALIAS, PRIMARY_CLUSTER_ALIAS);

        // make sure the topic is auto-created in the other cluster
        waitForTopicCreated(primary, "backup.test-topic-1");
        waitForTopicCreated(backup, "primary.test-topic-1");

        // make sure to remote topics are created using FakeForwardingAdminWithLocalMetadata
        waitForTopicToPersistInFakeLocalMetadataStore("backup.test-topic-1");
        waitForTopicToPersistInFakeLocalMetadataStore("primary.test-topic-1");

        // increase number of partitions
        Map<String, NewPartitions> newPartitions = Map.of("test-topic-1", NewPartitions.increaseTo(NUM_PARTITIONS + 1));
        try (Admin adminClient = primary.kafka().createAdminClient()) {
            adminClient.createPartitions(newPartitions).all().get();
        }

        // make sure partition is created in the other cluster
        waitForTopicPartitionCreated(backup, "primary.test-topic-1", NUM_PARTITIONS + 1);

        // expect to use FakeForwardingAdminWithLocalMetadata to update number of partitions in local store
        waitForTopicConfigPersistInFakeLocalMetaDataStore("primary.test-topic-1", "partitions", String.valueOf(NUM_PARTITIONS + 1));
    }

    @Test
    public void testSyncTopicConfigUseProvidedForwardingAdmin() throws Exception {
        mm2Props.put("sync.topic.configs.enabled", "true");
        mm2Config = new MirrorMakerConfig(mm2Props);
        produceMessages(backupProducer, "test-topic-1");
        produceMessages(primaryProducer, "test-topic-1");
        String consumerGroupName = "consumer-group-testReplication";
        Map<String, Object> consumerProps = Map.of("group.id", consumerGroupName);
        // warm up consumers before starting the connectors so we don't need to wait for discovery
        warmUpConsumer(consumerProps);

        waitUntilMirrorMakerIsRunning(primary, CONNECTOR_LIST, mm2Config, BACKUP_CLUSTER_ALIAS, PRIMARY_CLUSTER_ALIAS);
        waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS);

        // make sure the topic is auto-created in the other cluster
        waitForTopicCreated(primary, "backup.test-topic-1");
        waitForTopicCreated(backup, "primary.test-topic-1");

        // make sure the topic config is synced into the other cluster
        assertEquals(TopicConfig.CLEANUP_POLICY_COMPACT, getTopicConfig(backup.kafka(), "primary.test-topic-1", TopicConfig.CLEANUP_POLICY_CONFIG),
            "topic config was synced");

        // expect to use FakeForwardingAdminWithLocalMetadata to create remote topics into local store
        waitForTopicToPersistInFakeLocalMetadataStore("backup.test-topic-1");
        waitForTopicToPersistInFakeLocalMetadataStore("primary.test-topic-1");

        // expect to use FakeForwardingAdminWithLocalMetadata to update topic config in local store
        waitForTopicConfigPersistInFakeLocalMetaDataStore("primary.test-topic-1", TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
    }

    @Test
    public void testSyncTopicACLsUseProvidedForwardingAdmin() throws Exception {
        mm2Props.put("sync.topic.acls.enabled", "true");
        mm2Props.put("sync.topic.acls.interval.seconds", "1");
        mm2Config = new MirrorMakerConfig(mm2Props);
        List<AclBinding> aclBindings = List.of(
                new AclBinding(
                        new ResourcePattern(ResourceType.TOPIC, "test-topic-1", PatternType.LITERAL),
                        new AccessControlEntry("User:dummy", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW)
                )
        );
        try (Admin adminClient = primary.kafka().createAdminClient()) {
            adminClient.createAcls(aclBindings).all().get();
        }
        try (Admin adminClient = backup.kafka().createAdminClient()) {
            adminClient.createAcls(aclBindings).all().get();
        }

        waitUntilMirrorMakerIsRunning(primary, CONNECTOR_LIST, mm2Config, BACKUP_CLUSTER_ALIAS, PRIMARY_CLUSTER_ALIAS);
        waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS);

        // make sure the topic is auto-created in the other cluster
        waitForTopicCreated(primary, "backup.test-topic-1");
        waitForTopicCreated(backup, "primary.test-topic-1");

        // expect to use FakeForwardingAdminWithLocalMetadata to update topic ACLs on remote cluster
        AclBinding expectedACLOnBackupCluster = new AclBinding(
                new ResourcePattern(ResourceType.TOPIC, "primary.test-topic-1", PatternType.LITERAL),
                new AccessControlEntry("User:dummy", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW)
        );
        AclBinding expectedACLOnPrimaryCluster = new AclBinding(
                new ResourcePattern(ResourceType.TOPIC, "backup.test-topic-1", PatternType.LITERAL),
                new AccessControlEntry("User:dummy", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW)
        );

        // In some rare cases replica topics are created before ACLs are synced, so retry logic is necessary
        waitForCondition(
                () -> {
                    assertTrue(getAclBindings(backup.kafka(), "primary.test-topic-1").contains(expectedACLOnBackupCluster), "topic ACLs are not synced on backup cluster");
                    assertTrue(getAclBindings(primary.kafka(), "backup.test-topic-1").contains(expectedACLOnPrimaryCluster), "topic ACLs are not synced on primary cluster");
                    return true;
                },
                TOPIC_ACL_SYNC_DURATION_MS,
                "Topic ACLs were not synced in time"
        );

        // expect to use FakeForwardingAdminWithLocalMetadata to update topic ACLs in FakeLocalMetadataStore.allAcls
        assertTrue(FakeLocalMetadataStore.aclBindings("dummy").containsAll(List.of(expectedACLOnBackupCluster, expectedACLOnPrimaryCluster)));
    }

    void waitForTopicToPersistInFakeLocalMetadataStore(String topicName) throws InterruptedException {
        waitForCondition(() -> FakeLocalMetadataStore.containsTopic(topicName), FAKE_LOCAL_METADATA_STORE_SYNC_DURATION_MS,
            "Topic: " + topicName + " didn't get created in the FakeLocalMetadataStore"
        );
    }

    void waitForTopicConfigPersistInFakeLocalMetaDataStore(String topicName, String configName, String expectedConfigValue) throws InterruptedException {
        waitForCondition(() -> FakeLocalMetadataStore.topicConfig(topicName).getOrDefault(configName, "").equals(expectedConfigValue), FAKE_LOCAL_METADATA_STORE_SYNC_DURATION_MS,
            "Topic: " + topicName + "'s configs don't have " + configName + ":" + expectedConfigValue
        );
    }
}
