/*
 * 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;

import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.utils.Utils;

import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;

import static org.apache.kafka.connect.mirror.TestUtils.assertEqualsExceptClientId;
import static org.apache.kafka.connect.mirror.TestUtils.makeProps;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class MirrorCheckpointConfigTest {

    @Test
    public void testTaskConfigConsumerGroups() {
        List<String> groups = Arrays.asList("consumer-1", "consumer-2", "consumer-3");
        MirrorCheckpointConfig config = new MirrorCheckpointConfig(makeProps());
        Map<String, String> props = config.taskConfigForConsumerGroups(groups, 1);
        MirrorCheckpointTaskConfig taskConfig = new MirrorCheckpointTaskConfig(props);
        assertEquals(taskConfig.taskConsumerGroups(), new HashSet<>(groups),
                "Setting consumer groups property configuration failed");
    }

    @Test
    public void testGroupMatching() {
        MirrorCheckpointConfig config = new MirrorCheckpointConfig(makeProps("groups", "group1"));
        assertTrue(config.groupFilter().shouldReplicateGroup("group1"),
                "topic1 group matching property configuration failed");
        assertFalse(config.groupFilter().shouldReplicateGroup("group2"),
                "topic2 group matching property configuration failed");
    }

    @Test
    public void testNonMutationOfConfigDef() {
        // Sanity check to make sure that these properties are actually defined for the task config,
        // and that the task config class has been loaded and statically initialized by the JVM
        ConfigDef taskConfigDef = MirrorCheckpointTaskConfig.TASK_CONFIG_DEF;
        assertTrue(
                taskConfigDef.names().contains(MirrorCheckpointConfig.TASK_CONSUMER_GROUPS),
                MirrorCheckpointConfig.TASK_CONSUMER_GROUPS + " should be defined for task ConfigDef"
        );

        // Ensure that the task config class hasn't accidentally modified the connector config
        assertFalse(
                MirrorCheckpointConfig.CONNECTOR_CONFIG_DEF.names().contains(MirrorCheckpointConfig.TASK_CONSUMER_GROUPS),
                MirrorCheckpointConfig.TASK_CONSUMER_GROUPS + " should not be defined for connector ConfigDef"
        );
    }

    @Test
    public void testConsumerConfigsForOffsetSyncsTopic() {
        Map<String, String> connectorProps = makeProps(
                "source.consumer.max.partition.fetch.bytes", "1",
                "target.consumer.heartbeat.interval.ms", "1",
                "consumer.max.poll.interval.ms", "1",
                "fetch.min.bytes", "1"
        );
        MirrorCheckpointConfig config = new MirrorCheckpointConfig(connectorProps);
        Map<String, Object> sourceConsumerConfig = config.sourceConsumerConfig("test");
        Map<String, Object> offsetSyncsTopicSourceConsumerConfig = config.offsetSyncsTopicConsumerConfig();
        assertEqualsExceptClientId(sourceConsumerConfig, offsetSyncsTopicSourceConsumerConfig);
        assertEquals("source1->target2|ConnectorName|test", sourceConsumerConfig.get("client.id"));
        assertEquals(
                "source1->target2|ConnectorName|" + MirrorCheckpointConfig.OFFSET_SYNCS_SOURCE_CONSUMER_ROLE,
                offsetSyncsTopicSourceConsumerConfig.get("client.id"));
        connectorProps.put("offset-syncs.topic.location", "target");
        config = new MirrorCheckpointConfig(connectorProps);
        Map<String, Object> targetConsumerConfig = config.targetConsumerConfig("test");
        Map<String, Object> offsetSyncsTopicTargetConsumerConfig = config.offsetSyncsTopicConsumerConfig();
        assertEqualsExceptClientId(targetConsumerConfig, offsetSyncsTopicTargetConsumerConfig);
        assertEquals("source1->target2|ConnectorName|test", targetConsumerConfig.get("client.id"));
        assertEquals(
                "source1->target2|ConnectorName|" + MirrorCheckpointConfig.OFFSET_SYNCS_TARGET_CONSUMER_ROLE,
                offsetSyncsTopicTargetConsumerConfig.get("client.id"));
    }

    @Test
    public void testSkipValidationIfConnectorDisabled() {
        Map<String, String> configValues = MirrorCheckpointConfig.validate(makeProps(
                MirrorConnectorConfig.ENABLED, "false",
                MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED, "false",
                MirrorCheckpointConfig.SYNC_GROUP_OFFSETS_ENABLED, "false"));
        assertTrue(configValues.isEmpty());

        configValues = MirrorCheckpointConfig.validate(makeProps(
                MirrorConnectorConfig.ENABLED, "false",
                MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED, "true",
                MirrorCheckpointConfig.EMIT_OFFSET_SYNCS_ENABLED, "false"));
        assertTrue(configValues.isEmpty());
    }

    @Test
    public void testValidateIfConnectorEnabled() {
        Map<String, String> configValues = MirrorCheckpointConfig.validate(makeProps(
                MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED, "false",
                MirrorCheckpointConfig.SYNC_GROUP_OFFSETS_ENABLED, "false"));
        assertEquals(configValues.keySet(), Collections.singleton(MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED));

        configValues = MirrorCheckpointConfig.validate(makeProps(MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED, "true",
                MirrorCheckpointConfig.EMIT_OFFSET_SYNCS_ENABLED, "false"));
        assertEquals(configValues.keySet(), Utils.mkSet(MirrorCheckpointConfig.EMIT_OFFSET_SYNCS_ENABLED));

        configValues = MirrorCheckpointConfig.validate(makeProps(MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED, "true",
                MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED, "true",
                MirrorCheckpointConfig.EMIT_OFFSET_SYNCS_ENABLED, "true"));
        assertTrue(configValues.isEmpty());
    }
}
