text:
package io.strimzi.systemtest.mirrormaker;

import io.fabric8.kubernetes.api.model.LabelSelector;
import io.fabric8.kubernetes.api.model.Secret;
import io.fabric8.kubernetes.api.model.SecretBuilder;
import io.strimzi.api.kafka.model.CertSecretSource;
import io.strimzi.api.kafka.model.KafkaMirrorMaker2;
import io.strimzi.api.kafka.model.KafkaMirrorMaker2Resources;
import io.strimzi.api.kafka.model.KafkaResources;
import io.strimzi.api.kafka.model.KafkaTopic;
import io.strimzi.api.kafka.model.PasswordSecretSource;
import io.strimzi.api.kafka.model.listener.KafkaListenerAuthenticationScramSha512;
import io.strimzi.api.kafka.model.listener.KafkaListenerAuthenticationTls;
import io.strimzi.api.kafka.model.listener.arraylistener.GenericKafkaListenerBuilder;
import io.strimzi.api.kafka.model.listener.arraylistener.KafkaListenerType;
import io.strimzi.api.kafka.model.status.KafkaMirrorMaker2Status;
import io.strimzi.api.kafka.model.template.DeploymentStrategy;
import io.strimzi.operator.cluster.model.Ca;
import io.strimzi.systemtest.AbstractST;
import io.strimzi.systemtest.BeforeAllOnce;
import io.strimzi.systemtest.Constants;
import io.strimzi.systemtest.Environment;
import io.strimzi.systemtest.annotations.IsolatedSuite;
import io.strimzi.systemtest.annotations.KRaftNotSupported;
import io.strimzi.systemtest.kafkaclients.internalClients.KafkaClients;
import io.strimzi.systemtest.kafkaclients.internalClients.KafkaClientsBuilder;
import io.strimzi.systemtest.resources.crd.KafkaResource;
import io.strimzi.systemtest.resources.operator.SetupClusterOperator;
import io.strimzi.systemtest.annotations.ParallelNamespaceTest;
import io.strimzi.systemtest.cli.KafkaCmdClient;
import io.strimzi.operator.common.model.Labels;
import io.strimzi.systemtest.resources.crd.KafkaMirrorMaker2Resource;
import io.strimzi.systemtest.resources.crd.KafkaTopicResource;
import io.strimzi.systemtest.storage.TestStorage;
import io.strimzi.systemtest.templates.crd.KafkaMirrorMaker2Templates;
import io.strimzi.systemtest.templates.crd.KafkaTemplates;
import io.strimzi.systemtest.templates.crd.KafkaTopicTemplates;
import io.strimzi.systemtest.templates.crd.KafkaUserTemplates;
import io.strimzi.systemtest.templates.specific.ScraperTemplates;
import io.strimzi.systemtest.utils.ClientUtils;
import io.strimzi.systemtest.utils.RollingUpdateUtils;
import io.strimzi.systemtest.utils.StUtils;
import io.strimzi.systemtest.utils.kafkaUtils.KafkaMirrorMaker2Utils;
import io.strimzi.systemtest.utils.kafkaUtils.KafkaTopicUtils;
import io.strimzi.systemtest.utils.kafkaUtils.KafkaUtils;
import io.strimzi.systemtest.utils.kubeUtils.controllers.DeploymentUtils;
import io.strimzi.systemtest.utils.kubeUtils.objects.PodUtils;
import io.strimzi.systemtest.utils.kubeUtils.objects.SecretUtils;
import io.strimzi.test.TestUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.extension.ExtensionContext;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static io.strimzi.systemtest.Constants.ACCEPTANCE;
import static io.strimzi.systemtest.Constants.CONNECT_COMPONENTS;
import static io.strimzi.systemtest.Constants.INTERNAL_CLIENTS_USED;
import static io.strimzi.systemtest.Constants.MIRROR_MAKER2;
import static io.strimzi.systemtest.Constants.REGRESSION;
import static io.strimzi.systemtest.Constants.SCALABILITY;
import static io.strimzi.systemtest.enums.CustomResourceStatus.Ready;
import static io.strimzi.test.k8s.KubeClusterResource.cmdKubeClient;
import static io.strimzi.test.k8s.KubeClusterResource.kubeClient;
import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.hasItem;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.valid4j.matchers.jsonpath.JsonPathMatchers.hasJsonPath;

@Tag(REGRESSION)
@Tag(MIRROR_MAKER2)
@Tag(CONNECT_COMPONENTS)
@Tag(INTERNAL_CLIENTS_USED)
@IsolatedSuite
class MirrorMaker2IsolatedST extends AbstractST {

    private static final Logger LOGGER = LogManager.getLogger(MirrorMaker2IsolatedST.class);

    @ParallelNamespaceTest
    void testMirrorMaker2(ExtensionContext extensionContext) {
        final TestStorage testStorage = new TestStorage(extensionContext, clusterOperator.getDeploymentNamespace());

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        String sourceMirroredTopicName = kafkaClusterSourceName + "." + testStorage.getTopicName();

        Map<String, Object> expectedConfig = StUtils.loadProperties("group.id=mirrormaker2-cluster\n" +
                "key.converter=org.apache.kafka.connect.converters.ByteArrayConverter\n" +
                "value.converter=org.apache.kafka.connect.converters.ByteArrayConverter\n" +
                "header.converter=org.apache.kafka.connect.converters.ByteArrayConverter\n" +
                "config.storage.topic=mirrormaker2-cluster-configs\n" +
                "status.storage.topic=mirrormaker2-cluster-status\n" +
                "offset.storage.topic=mirrormaker2-cluster-offsets\n" +
                "config.storage.replication.factor=-1\n" +
                "status.storage.replication.factor=-1\n" +
                "offset.storage.replication.factor=-1\n" +
                "config.providers=file\n" +
                "config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider\n");

        // Deploy source and target kafka
        resourceManager.createResource(extensionContext,
            KafkaTemplates.kafkaEphemeral(kafkaClusterSourceName, 1).build(),
            KafkaTemplates.kafkaEphemeral(kafkaClusterTargetName, 1).build()
        );

        // Deploy source topic
        resourceManager.createResource(extensionContext, KafkaTopicTemplates.topic(kafkaClusterSourceName, testStorage.getTopicName(), 3).build());

        KafkaClients clients = new KafkaClientsBuilder()
            .withProducerName(testStorage.getProducerName())
            .withConsumerName(testStorage.getConsumerName())
            .withBootstrapAddress(KafkaResources.plainBootstrapAddress(kafkaClusterSourceName))
            .withNamespaceName(testStorage.getNamespaceName())
            .withTopicName(testStorage.getTopicName())
            .withMessageCount(MESSAGE_COUNT)
            .build();

        // Check brokers availability
        LOGGER.info("Messages exchange - topic {}, cluster {} and message count of {}",
            testStorage.getTopicName(), kafkaClusterSourceName, MESSAGE_COUNT);

        resourceManager.createResource(extensionContext, clients.producerStrimzi(), clients.consumerStrimzi());
        ClientUtils.waitForClientsSuccess(testStorage.getProducerName(), testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        resourceManager.createResource(extensionContext, KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, false)
            .editSpec()
                .editFirstMirror()
                    .editSourceConnector()
                        .addToConfig("refresh.topics.interval.seconds", "1")
                    .endSourceConnector()
                .endMirror()
            .endSpec()
            .build());

        String podName = PodUtils.getPodNameByPrefix(testStorage.getNamespaceName(), KafkaMirrorMaker2Resources.deploymentName(testStorage.getClusterName()));
        String kafkaPodJson = TestUtils.toJsonString(kubeClient().getPod(testStorage.getNamespaceName(), podName));

        assertThat(kafkaPodJson, hasJsonPath(StUtils.globalVariableJsonPathBuilder(0, "KAFKA_CONNECT_BOOTSTRAP_SERVERS"),
                hasItem(KafkaResources.plainBootstrapAddress(kafkaClusterTargetName))));
        assertThat(StUtils.getPropertiesFromJson(0, kafkaPodJson, "KAFKA_CONNECT_CONFIGURATION"), is(expectedConfig));
        testDockerImagesForKafkaMirrorMaker2(testStorage.getClusterName(), clusterOperator.getDeploymentNamespace(), testStorage.getNamespaceName());

        verifyLabelsOnPods(testStorage.getNamespaceName(), testStorage.getClusterName(), "mirrormaker2", KafkaMirrorMaker2.RESOURCE_KIND);
        verifyLabelsForService(testStorage.getNamespaceName(), testStorage.getClusterName(), "mirrormaker2-api", KafkaMirrorMaker2.RESOURCE_KIND);
        verifyLabelsForConfigMaps(testStorage.getNamespaceName(), kafkaClusterSourceName, null, kafkaClusterTargetName);
        verifyLabelsForServiceAccounts(testStorage.getNamespaceName(), kafkaClusterSourceName, null);

        LOGGER.info("Now setting topic to {} and cluster to {} - the messages should be mirrored",
            sourceMirroredTopicName, kafkaClusterTargetName);

        clients = new KafkaClientsBuilder(clients)
            .withTopicName(sourceMirroredTopicName)
            .withBootstrapAddress(KafkaResources.plainBootstrapAddress(kafkaClusterTargetName))
            .build();

        LOGGER.info("Consumer in target cluster and topic should receive {} messages", MESSAGE_COUNT);

        resourceManager.createResource(extensionContext, clients.consumerStrimzi());
        ClientUtils.waitForClientSuccess(testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Mirrored successful");

        if (!Environment.isKRaftModeEnabled()) {
            KafkaTopic mirroredTopic = KafkaTopicResource.kafkaTopicClient().inNamespace(testStorage.getNamespaceName()).withName(sourceMirroredTopicName).get();
            assertThat(mirroredTopic.getSpec().getPartitions(), is(3));
            assertThat(mirroredTopic.getMetadata().getLabels().get(Labels.STRIMZI_CLUSTER_LABEL), is(kafkaClusterTargetName));

            // Replace source topic resource with new data and check that mm2 update target topi
            KafkaTopicResource.replaceTopicResourceInSpecificNamespace(testStorage.getTopicName(), kt -> kt.getSpec().setPartitions(8), testStorage.getNamespaceName());
            KafkaTopicUtils.waitForKafkaTopicPartitionChange(testStorage.getNamespaceName(), sourceMirroredTopicName, 8);
        }
    }

    /**
     * Test mirroring messages by MirrorMaker 2.0 over tls transport using mutual tls auth
     */
    @SuppressWarnings({"checkstyle:MethodLength"})
    @ParallelNamespaceTest
    @Tag(ACCEPTANCE)
    void testMirrorMaker2TlsAndTlsClientAuth(ExtensionContext extensionContext) {
        final TestStorage testStorage = new TestStorage(extensionContext, clusterOperator.getDeploymentNamespace());

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        String sourceMirroredTopicName = kafkaClusterSourceName + "." + testStorage.getTopicName();

        String kafkaUserSourceName = testStorage.getUserName() + "-source";
        String kafkaUserTargetName = testStorage.getUserName() + "-target";

        // Deploy source kafka with tls listener and mutual tls auth
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaEphemeral(kafkaClusterSourceName, 1, 1)
            .editSpec()
                .editKafka()
                    .withListeners(new GenericKafkaListenerBuilder()
                        .withName(Constants.TLS_LISTENER_DEFAULT_NAME)
                        .withPort(9093)
                        .withType(KafkaListenerType.INTERNAL)
                        .withTls(true)
                        .withAuth(new KafkaListenerAuthenticationTls())
                        .build())
                .endKafka()
            .endSpec()
            .build());

        // Deploy target kafka with tls listener and mutual tls auth
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaEphemeral(kafkaClusterTargetName, 1, 1)
            .editSpec()
                .editKafka()
                    .withListeners(new GenericKafkaListenerBuilder()
                        .withName(Constants.TLS_LISTENER_DEFAULT_NAME)
                        .withPort(9093)
                        .withType(KafkaListenerType.INTERNAL)
                        .withTls(true)
                        .withAuth(new KafkaListenerAuthenticationTls())
                        .build())
                .endKafka()
            .endSpec()
            .build());

        // Deploy topic
        resourceManager.createResource(extensionContext,
            KafkaTopicTemplates.topic(kafkaClusterSourceName, testStorage.getTopicName(), 3).build(),
            KafkaUserTemplates.tlsUser(kafkaClusterSourceName, kafkaUserSourceName).build(),
            KafkaUserTemplates.tlsUser(kafkaClusterTargetName, kafkaUserTargetName).build()
        );

        // Initialize CertSecretSource with certificate and secret names for source
        CertSecretSource certSecretSource = new CertSecretSource();
        certSecretSource.setCertificate("ca.crt");
        certSecretSource.setSecretName(KafkaResources.clusterCaCertificateSecretName(kafkaClusterSourceName));

        // Initialize CertSecretSource with certificate and secret names for target
        CertSecretSource certSecretTarget = new CertSecretSource();
        certSecretTarget.setCertificate("ca.crt");
        certSecretTarget.setSecretName(KafkaResources.clusterCaCertificateSecretName(kafkaClusterTargetName));

        KafkaClients clients = new KafkaClientsBuilder()
            .withProducerName(testStorage.getProducerName())
            .withConsumerName(testStorage.getConsumerName())
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterSourceName))
            .withUserName(kafkaUserSourceName)
            .withNamespaceName(testStorage.getNamespaceName())
            .withTopicName(testStorage.getTopicName())
            .withMessageCount(MESSAGE_COUNT)
            .build();

        // Check brokers availability
        LOGGER.info("Messages exchange - topic {}, cluster {} and message count of {}",
            testStorage.getTopicName(), kafkaClusterSourceName, MESSAGE_COUNT);

        resourceManager.createResource(extensionContext, clients.producerTlsStrimzi(kafkaClusterSourceName), clients.consumerTlsStrimzi(kafkaClusterSourceName));
        ClientUtils.waitForClientsSuccess(testStorage.getProducerName(), testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        resourceManager.createResource(extensionContext, KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, true)
            .editSpec()
                .editMatchingCluster(spec -> spec.getAlias().equals(kafkaClusterSourceName))
                    .withNewKafkaClientAuthenticationTls()
                        .withNewCertificateAndKey()
                            .withSecretName(kafkaUserSourceName)
                            .withCertificate("user.crt")
                            .withKey("user.key")
                        .endCertificateAndKey()
                    .endKafkaClientAuthenticationTls()
                    .withNewTls()
                        .withTrustedCertificates(certSecretSource)
                    .endTls()
                .endCluster()
                .editMatchingCluster(spec -> spec.getAlias().equals(kafkaClusterTargetName))
                    .withNewKafkaClientAuthenticationTls()
                        .withNewCertificateAndKey()
                            .withSecretName(kafkaUserTargetName)
                            .withCertificate("user.crt")
                            .withKey("user.key")
                        .endCertificateAndKey()
                    .endKafkaClientAuthenticationTls()
                    .withNewTls()
                        .withTrustedCertificates(certSecretTarget)
                    .endTls()
                .endCluster()
            .endSpec()
            .build());

        LOGGER.info("Changing to mirrored topic - topic {}, cluster {}, user {}", sourceMirroredTopicName, kafkaClusterTargetName, kafkaClusterTargetName);

        clients = new KafkaClientsBuilder(clients)
            .withTopicName(sourceMirroredTopicName)
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterTargetName))
            .withUserName(kafkaUserTargetName)
            .build();

        LOGGER.info("Now messages should be mirrored to target topic and cluster");
        resourceManager.createResource(extensionContext, clients.consumerTlsStrimzi(kafkaClusterTargetName));
        ClientUtils.waitForClientSuccess(testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);
        LOGGER.info("Messages successfully mirrored");

        if (!Environment.isKRaftModeEnabled()) {
            KafkaTopicUtils.waitForKafkaTopicCreation(testStorage.getNamespaceName(), sourceMirroredTopicName);
            KafkaTopic mirroredTopic = KafkaTopicResource.kafkaTopicClient().inNamespace(testStorage.getNamespaceName()).withName(sourceMirroredTopicName).get();

            assertThat(mirroredTopic.getSpec().getPartitions(), is(3));
            assertThat(mirroredTopic.getMetadata().getLabels().get(Labels.STRIMZI_CLUSTER_LABEL), is(kafkaClusterTargetName));
        }
    }

    /**
     * Test mirroring messages by MirrorMaker 2.0 over tls transport using scram-sha-512 auth
     */
    @ParallelNamespaceTest
    @KRaftNotSupported("Scram-sha is not supported by KRaft mode and is used in this test case")
    void testMirrorMaker2TlsAndScramSha512Auth(ExtensionContext extensionContext) {
        final TestStorage testStorage = new TestStorage(extensionContext, clusterOperator.getDeploymentNamespace());

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        String sourceMirroredTopicName = kafkaClusterSourceName + "." + testStorage.getTopicName();

        String kafkaUserSourceName = testStorage.getUserName() + "-source";
        String kafkaUserTargetName = testStorage.getUserName() + "-target";

        resourceManager.createResource(extensionContext,
            KafkaTemplates.kafkaEphemeral(kafkaClusterSourceName, 1, 1)
            .editSpec()
                .editKafka()
                    .withListeners(new GenericKafkaListenerBuilder()
                            .withName(Constants.TLS_LISTENER_DEFAULT_NAME)
                            .withPort(9093)
                            .withType(KafkaListenerType.INTERNAL)
                            .withTls(true)
                            .withAuth(new KafkaListenerAuthenticationScramSha512())
                            .build())
                .endKafka()
            .endSpec()
            .build(),
            KafkaTemplates.kafkaEphemeral(kafkaClusterTargetName, 1, 1)
                .editSpec()
                    .editKafka()
                        .withListeners(new GenericKafkaListenerBuilder()
                            .withName(Constants.TLS_LISTENER_DEFAULT_NAME)
                            .withPort(9093)
                            .withType(KafkaListenerType.INTERNAL)
                            .withTls(true)
                            .withAuth(new KafkaListenerAuthenticationScramSha512())
                            .build())
                    .endKafka()
                .endSpec()
                .build()
        );

        resourceManager.createResource(extensionContext,
            KafkaTopicTemplates.topic(kafkaClusterSourceName, testStorage.getTopicName(), 3).build(),
            KafkaUserTemplates.scramShaUser(kafkaClusterSourceName, kafkaUserSourceName).build(),
            KafkaUserTemplates.scramShaUser(kafkaClusterTargetName, kafkaUserTargetName).build()
        );

        // Initialize PasswordSecretSource to set this as PasswordSecret in MirrorMaker2 spec
        PasswordSecretSource passwordSecretSource = new PasswordSecretSource();
        passwordSecretSource.setSecretName(kafkaUserSourceName);
        passwordSecretSource.setPassword("password");

        // Initialize PasswordSecretSource to set this as PasswordSecret in MirrorMaker2 spec
        PasswordSecretSource passwordSecretTarget = new PasswordSecretSource();
        passwordSecretTarget.setSecretName(kafkaUserTargetName);
        passwordSecretTarget.setPassword("password");

        // Initialize CertSecretSource with certificate and secret names for source
        CertSecretSource certSecretSource = new CertSecretSource();
        certSecretSource.setCertificate("ca.crt");
        certSecretSource.setSecretName(KafkaResources.clusterCaCertificateSecretName(kafkaClusterSourceName));

        // Initialize CertSecretSource with certificate and secret names for target
        CertSecretSource certSecretTarget = new CertSecretSource();
        certSecretTarget.setCertificate("ca.crt");
        certSecretTarget.setSecretName(KafkaResources.clusterCaCertificateSecretName(kafkaClusterTargetName));

        KafkaClients clients = new KafkaClientsBuilder()
            .withProducerName(testStorage.getProducerName())
            .withConsumerName(testStorage.getConsumerName())
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterSourceName))
            .withUserName(kafkaUserSourceName)
            .withNamespaceName(testStorage.getNamespaceName())
            .withTopicName(testStorage.getTopicName())
            .withMessageCount(MESSAGE_COUNT)
            .build();

        // Check brokers availability
        LOGGER.info("Messages exchange - topic {}, cluster {} and message count of {}",
            testStorage.getTopicName(), kafkaClusterSourceName, MESSAGE_COUNT);

        resourceManager.createResource(extensionContext, clients.producerScramShaTlsStrimzi(kafkaClusterSourceName), clients.consumerScramShaTlsStrimzi(kafkaClusterSourceName));
        ClientUtils.waitForClientsSuccess(testStorage.getProducerName(), testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        resourceManager.createResource(extensionContext, KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, true)
            .editSpec()
                .editMatchingCluster(spec -> spec.getAlias().equals(kafkaClusterSourceName))
                    .withNewKafkaClientAuthenticationScramSha512()
                        .withUsername(kafkaUserSourceName)
                        .withPasswordSecret(passwordSecretSource)
                    .endKafkaClientAuthenticationScramSha512()
                    .withNewTls()
                        .withTrustedCertificates(certSecretSource)
                    .endTls()
                .endCluster()
                .editMatchingCluster(spec -> spec.getAlias().equals(kafkaClusterTargetName))
                    .withNewKafkaClientAuthenticationScramSha512()
                        .withUsername(kafkaUserTargetName)
                        .withPasswordSecret(passwordSecretTarget)
                    .endKafkaClientAuthenticationScramSha512()
                    .withNewTls()
                        .withTrustedCertificates(certSecretTarget)
                    .endTls()
                .endCluster()
            .endSpec()
            .build());

        LOGGER.info("Changing to mirrored topic - topic {}, cluster {}, user {}", sourceMirroredTopicName, kafkaClusterTargetName, kafkaClusterTargetName);

        clients = new KafkaClientsBuilder(clients)
            .withTopicName(sourceMirroredTopicName)
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterTargetName))
            .withUserName(kafkaUserTargetName)
            .build();

        LOGGER.info("Now messages should be mirrored to target topic and cluster");
        resourceManager.createResource(extensionContext, clients.consumerScramShaTlsStrimzi(kafkaClusterTargetName));
        ClientUtils.waitForClientSuccess(testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);
        LOGGER.info("Messages successfully mirrored");

        KafkaTopicUtils.waitForKafkaTopicCreation(testStorage.getNamespaceName(), sourceMirroredTopicName);
        KafkaTopic mirroredTopic = KafkaTopicResource.kafkaTopicClient().inNamespace(testStorage.getNamespaceName()).withName(sourceMirroredTopicName).get();

        assertThat(mirroredTopic.getSpec().getPartitions(), is(3));
        assertThat(mirroredTopic.getMetadata().getLabels().get(Labels.STRIMZI_CLUSTER_LABEL), is(kafkaClusterTargetName));
    }

    @ParallelNamespaceTest
    @Tag(SCALABILITY)
    void testScaleMirrorMaker2UpAndDownToZero(ExtensionContext extensionContext) {
        final TestStorage testStorage = new TestStorage(extensionContext, clusterOperator.getDeploymentNamespace());

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        String mm2DepName = KafkaMirrorMaker2Resources.deploymentName(testStorage.getClusterName());

        // Deploy source kafka
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaEphemeral(kafkaClusterSourceName, 1, 1).build());
        // Deploy target kafka
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaEphemeral(kafkaClusterTargetName, 1, 1).build());

        resourceManager.createResource(extensionContext, KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, false).build());

        int scaleTo = 2;
        long mm2ObsGen = KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(testStorage.getNamespaceName()).withName(testStorage.getClusterName()).get().getStatus().getObservedGeneration();
        String mm2GenName = kubeClient().listPods(testStorage.getNamespaceName(), testStorage.getClusterName(), Labels.STRIMZI_KIND_LABEL, KafkaMirrorMaker2.RESOURCE_KIND).get(0).getMetadata().getGenerateName();

        LOGGER.info("-------> Scaling KafkaMirrorMaker2 up <-------");

        LOGGER.info("Scaling subresource replicas to {}", scaleTo);

        cmdKubeClient(testStorage.getNamespaceName()).scaleByName(KafkaMirrorMaker2.RESOURCE_KIND, testStorage.getClusterName(), scaleTo);
        DeploymentUtils.waitForDeploymentAndPodsReady(testStorage.getNamespaceName(), KafkaMirrorMaker2Resources.deploymentName(testStorage.getClusterName()), scaleTo);

        LOGGER.info("Check if replicas is set to {}, naming prefix should be same and observed generation higher", scaleTo);
        List<String> mm2Pods = kubeClient(testStorage.getNamespaceName()).listPodNames(testStorage.getNamespaceName(), testStorage.getClusterName(), Labels.STRIMZI_KIND_LABEL, KafkaMirrorMaker2.RESOURCE_KIND);

        assertThat(mm2Pods.size(), is(scaleTo));
        assertThat(KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(testStorage.getNamespaceName()).withName(testStorage.getClusterName()).get().getSpec().getReplicas(), is(scaleTo));
        assertThat(KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(testStorage.getNamespaceName()).withName(testStorage.getClusterName()).get().getStatus().getReplicas(), is(scaleTo));
        /*
        observed generation should be higher than before scaling -> after change of spec and successful reconciliation,
        the observed generation is increased
        */
        long actualObsGen = KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(testStorage.getNamespaceName()).withName(testStorage.getClusterName()).get().getMetadata().getGeneration();

        assertTrue(mm2ObsGen < actualObsGen);
        for (String pod : mm2Pods) {
            assertTrue(pod.contains(mm2GenName));
        }

        mm2ObsGen = actualObsGen;

        LOGGER.info("-------> Scaling KafkaMirrorMaker2 down <-------");

        LOGGER.info("Scaling MirrorMaker2 to zero");
        KafkaMirrorMaker2Resource.replaceKafkaMirrorMaker2ResourceInSpecificNamespace(testStorage.getClusterName(), mm2 -> mm2.getSpec().setReplicas(0), testStorage.getNamespaceName());

        PodUtils.waitForPodsReady(testStorage.getNamespaceName(), kubeClient(testStorage.getNamespaceName()).getDeploymentSelectors(mm2DepName), 0, true, () -> { });

        mm2Pods = kubeClient().listPodNames(testStorage.getClusterName(), Labels.STRIMZI_KIND_LABEL, KafkaMirrorMaker2.RESOURCE_KIND);
        KafkaMirrorMaker2Status mm2Status = KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(testStorage.getNamespaceName()).withName(testStorage.getClusterName()).get().getStatus();
        actualObsGen = KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(testStorage.getNamespaceName()).withName(testStorage.getClusterName()).get().getMetadata().getGeneration();

        assertThat(mm2Pods.size(), is(0));
        assertThat(mm2Status.getConditions().get(0).getType(), is(Ready.toString()));
        assertThat(actualObsGen, is(not(mm2ObsGen)));

        TestUtils.waitFor("Until mirror maker 2 status url is null", Constants.GLOBAL_POLL_INTERVAL, Constants.GLOBAL_TIMEOUT, () -> {
            KafkaMirrorMaker2Status mm2StatusUrl = KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(testStorage.getNamespaceName()).withName(testStorage.getClusterName()).get().getStatus();
            return mm2StatusUrl.getUrl() == null;
        });
    }

    @ParallelNamespaceTest
    void testMirrorMaker2CorrectlyMirrorsHeaders(ExtensionContext extensionContext) {
        final TestStorage testStorage = new TestStorage(extensionContext, clusterOperator.getDeploymentNamespace());

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        String sourceMirroredTopicName = kafkaClusterSourceName + "." + testStorage.getTopicName();

        // Deploy source kafka
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaEphemeral(kafkaClusterSourceName, 1, 1).build());
        // Deploy target kafka
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaEphemeral(kafkaClusterTargetName, 1, 1).build());
        // Deploy Topic for example clients
        resourceManager.createResource(extensionContext, KafkaTopicTemplates.topic(kafkaClusterSourceName, testStorage.getTopicName()).build());

        resourceManager.createResource(extensionContext, KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, false)
                .editSpec()
                    .editMirror(0)
                        .editSourceConnector()
                            .addToConfig("refresh.topics.interval.seconds", "1")
                        .endSourceConnector()
                    .endMirror()
                .endSpec().build());

        // Deploying example clients for checking if mm2 will mirror messages with headers

        KafkaClients targetKafkaClientsJob = new KafkaClientsBuilder()
            .withConsumerName(testStorage.getConsumerName())
            .withBootstrapAddress(KafkaResources.plainBootstrapAddress(kafkaClusterTargetName))
            .withTopicName(sourceMirroredTopicName)
            .withMessageCount(MESSAGE_COUNT)
            .withDelayMs(1000)
            .build();

        resourceManager.createResource(extensionContext, targetKafkaClientsJob.consumerStrimzi());

        KafkaClients sourceKafkaClientsJob = new KafkaClientsBuilder()
            .withProducerName(testStorage.getProducerName())
            .withBootstrapAddress(KafkaResources.plainBootstrapAddress(kafkaClusterSourceName))
            .withTopicName(testStorage.getTopicName())
            .withMessageCount(MESSAGE_COUNT)
            .withHeaders("header_key_one=header_value_one, header_key_two=header_value_two")
            .withDelayMs(1000)
            .build();

        resourceManager.createResource(extensionContext, sourceKafkaClientsJob.producerStrimzi());

        ClientUtils.waitForClientsSuccess(testStorage.getProducerName(), testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT, false);

        LOGGER.info("Checking log of {} job if the headers are correct", testStorage.getConsumerName());
        String header1 = "key: header_key_one, value: header_value_one";
        String header2 = "key: header_key_two, value: header_value_two";
        String log = StUtils.getLogFromPodByTime(testStorage.getNamespaceName(), kubeClient(testStorage.getNamespaceName()).listPodsByPrefixInName(testStorage.getConsumerName()).get(0).getMetadata().getName(), "", MESSAGE_COUNT + "s");
        assertThat(log, containsString(header1));
        assertThat(log, containsString(header2));
    }

    /*
     * This test is using the Kafka Identity Replication policy. This is what should be used by all new users.
     */
    @ParallelNamespaceTest
    void testIdentityReplicationPolicy(ExtensionContext extensionContext) {
        final TestStorage testStorage = new TestStorage(extensionContext, clusterOperator.getDeploymentNamespace());

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        resourceManager.createResource(extensionContext,
            KafkaTemplates.kafkaEphemeral(kafkaClusterSourceName, 1, 1).build(),
            KafkaTemplates.kafkaEphemeral(kafkaClusterTargetName, 1, 1).build(),
            ScraperTemplates.scraperPod(testStorage.getNamespaceName(), testStorage.getScraperName()).build()
        );

        final String scraperPodName =  kubeClient().listPodsByPrefixInName(testStorage.getNamespaceName(), testStorage.getScraperName()).get(0).getMetadata().getName();

        // Create topic
        resourceManager.createResource(extensionContext, KafkaTopicTemplates.topic(kafkaClusterSourceName, testStorage.getTopicName(), 3).build());

        resourceManager.createResource(extensionContext, KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, false)
            .editSpec()
                .editMirror(0)
                    .editSourceConnector()
                        .addToConfig("replication.policy.class", "org.apache.kafka.connect.mirror.IdentityReplicationPolicy")
                        .addToConfig("refresh.topics.interval.seconds", "1")
                    .endSourceConnector()
                .endMirror()
            .endSpec()
            .build());

        LOGGER.info("Sending and receiving messages via {}", kafkaClusterSourceName);

        KafkaClients clients = new KafkaClientsBuilder()
            .withProducerName(testStorage.getProducerName())
            .withConsumerName(testStorage.getConsumerName())
            .withBootstrapAddress(KafkaResources.plainBootstrapAddress(kafkaClusterSourceName))
            .withNamespaceName(testStorage.getNamespaceName())
            .withTopicName(testStorage.getTopicName())
            .withMessageCount(MESSAGE_COUNT)
            .build();

        resourceManager.createResource(extensionContext, clients.producerStrimzi(), clients.consumerStrimzi());
        ClientUtils.waitForClientsSuccess(testStorage.getProducerName(), testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Changing to {} and will try to receive messages", kafkaClusterTargetName);

        clients = new KafkaClientsBuilder(clients)
            .withBootstrapAddress(KafkaResources.plainBootstrapAddress(kafkaClusterTargetName))
            .build();

        resourceManager.createResource(extensionContext, clients.consumerStrimzi());
        ClientUtils.waitForClientSuccess(testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        if (!Environment.isKRaftModeEnabled()) {
            LOGGER.info("Checking if the mirrored topic name is same as the original one");

            List<String> kafkaTopics = KafkaCmdClient.listTopicsUsingPodCli(testStorage.getNamespaceName(), scraperPodName, KafkaResources.plainBootstrapAddress(kafkaClusterTargetName));
            assertNotNull(kafkaTopics.stream().filter(kafkaTopic -> kafkaTopic.equals(testStorage.getTopicName())).findAny());

            String kafkaTopicSpec = KafkaCmdClient.describeTopicUsingPodCli(testStorage.getNamespaceName(), scraperPodName, KafkaResources.plainBootstrapAddress(kafkaClusterTargetName), testStorage.getTopicName());
            assertThat(kafkaTopicSpec, containsString("Topic: " + testStorage.getTopicName()));
            assertThat(kafkaTopicSpec, containsString("PartitionCount: 3"));
        }
    }

    /*
     * This test is using the Strimzi Identity Replication policy. This is needed for backwards compatibility for users
     * who might still have it configured.
     *
     * This ST should be deleted once we drop the Strimzi policy completely.
     */
    @ParallelNamespaceTest
    void testStrimziIdentityReplicationPolicy(ExtensionContext extensionContext) {
        final TestStorage testStorage = new TestStorage(extensionContext, clusterOperator.getDeploymentNamespace());

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        resourceManager.createResource(extensionContext,
            KafkaTemplates.kafkaEphemeral(kafkaClusterSourceName, 1, 1).build(),
            KafkaTemplates.kafkaEphemeral(kafkaClusterTargetName, 1, 1).build(),
            ScraperTemplates.scraperPod(testStorage.getNamespaceName(), testStorage.getScraperName()).build()
        );

        final String scraperPodName =  kubeClient().listPodsByPrefixInName(testStorage.getNamespaceName(), testStorage.getScraperName()).get(0).getMetadata().getName();

        // Create topic
        resourceManager.createResource(extensionContext, KafkaTopicTemplates.topic(kafkaClusterSourceName, testStorage.getTopicName(), 3).build());

        resourceManager.createResource(extensionContext, KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, false)
            .editSpec()
                .editMirror(0)
                    .editSourceConnector()
                        .addToConfig("replication.policy.class", "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy")
                        .addToConfig("refresh.topics.interval.seconds", "1")
                    .endSourceConnector()
                .endMirror()
            .endSpec()
            .build());

        LOGGER.info("Sending and receiving messages via {}", kafkaClusterSourceName);

        KafkaClients clients = new KafkaClientsBuilder()
            .withProducerName(testStorage.getProducerName())
            .withConsumerName(testStorage.getConsumerName())
            .withBootstrapAddress(KafkaResources.plainBootstrapAddress(kafkaClusterSourceName))
            .withNamespaceName(testStorage.getNamespaceName())
            .withTopicName(testStorage.getTopicName())
            .withMessageCount(MESSAGE_COUNT)
            .build();

        resourceManager.createResource(extensionContext, clients.producerStrimzi(), clients.consumerStrimzi());
        ClientUtils.waitForClientsSuccess(testStorage.getProducerName(), testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Changing to {} and will try to receive messages", kafkaClusterTargetName);

        clients = new KafkaClientsBuilder(clients)
            .withBootstrapAddress(KafkaResources.plainBootstrapAddress(kafkaClusterTargetName))
            .build();

        resourceManager.createResource(extensionContext, clients.consumerStrimzi());
        ClientUtils.waitForClientSuccess(testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        if (!Environment.isKRaftModeEnabled()) {
            LOGGER.info("Checking if the mirrored topic name is same as the original one");

            List<String> kafkaTopics = KafkaCmdClient.listTopicsUsingPodCli(testStorage.getNamespaceName(), scraperPodName, KafkaResources.plainBootstrapAddress(kafkaClusterTargetName));
            assertNotNull(kafkaTopics.stream().filter(kafkaTopic -> kafkaTopic.equals(testStorage.getTopicName())).findAny());

            String kafkaTopicSpec = KafkaCmdClient.describeTopicUsingPodCli(testStorage.getNamespaceName(), scraperPodName, KafkaResources.plainBootstrapAddress(kafkaClusterTargetName), testStorage.getTopicName());
            assertThat(kafkaTopicSpec, containsString("Topic: " + testStorage.getTopicName()));
            assertThat(kafkaTopicSpec, containsString("PartitionCount: 3"));
        }
    }

    @ParallelNamespaceTest
    void testConfigureDeploymentStrategy(ExtensionContext extensionContext) {
        final TestStorage testStorage = new TestStorage(extensionContext, clusterOperator.getDeploymentNamespace());

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        // Deploy source kafka
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaEphemeral(kafkaClusterSourceName, 1, 1).build());
        // Deploy target kafka
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaEphemeral(kafkaClusterTargetName, 1, 1).build());

        resourceManager.createResource(extensionContext, KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, false)
            .editSpec()
                .editOrNewTemplate()
                    .editOrNewDeployment()
                        .withDeploymentStrategy(DeploymentStrategy.RECREATE)
                    .endDeployment()
                .endTemplate()
            .endSpec()
            .build());

        String mm2DepName = KafkaMirrorMaker2Resources.deploymentName(testStorage.getClusterName());

        LOGGER.info("Adding label to MirrorMaker2 resource, the CR should be recreated");
        KafkaMirrorMaker2Resource.replaceKafkaMirrorMaker2ResourceInSpecificNamespace(testStorage.getClusterName(),
            mm2 -> mm2.getMetadata().setLabels(Collections.singletonMap("some", "label")), testStorage.getNamespaceName());
        DeploymentUtils.waitForDeploymentAndPodsReady(testStorage.getNamespaceName(), mm2DepName, 1);

        KafkaMirrorMaker2 kmm2 = KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(testStorage.getNamespaceName()).withName(testStorage.getClusterName()).get();

        LOGGER.info("Checking that observed gen. is still on 1 (recreation) and new label is present");
        assertThat(kmm2.getStatus().getObservedGeneration(), is(1L));
        assertThat(kmm2.getMetadata().getLabels().toString(), containsString("some=label"));
        assertThat(kmm2.getSpec().getTemplate().getDeployment().getDeploymentStrategy(), is(DeploymentStrategy.RECREATE));

        LOGGER.info("Changing deployment strategy to {}", DeploymentStrategy.ROLLING_UPDATE);
        KafkaMirrorMaker2Resource.replaceKafkaMirrorMaker2ResourceInSpecificNamespace(testStorage.getClusterName(),
            mm2 -> mm2.getSpec().getTemplate().getDeployment().setDeploymentStrategy(DeploymentStrategy.ROLLING_UPDATE), testStorage.getNamespaceName());
        KafkaMirrorMaker2Utils.waitForKafkaMirrorMaker2Ready(testStorage.getNamespaceName(), testStorage.getClusterName());

        LOGGER.info("Adding another label to MirrorMaker2 resource, pods should be rolled");
        KafkaMirrorMaker2Resource.replaceKafkaMirrorMaker2ResourceInSpecificNamespace(testStorage.getClusterName(), mm2 -> mm2.getMetadata().getLabels().put("another", "label"), testStorage.getNamespaceName());
        DeploymentUtils.waitForDeploymentAndPodsReady(testStorage.getNamespaceName(), mm2DepName, 1);

        LOGGER.info("Checking that observed gen. higher (rolling update) and label is changed");
        kmm2 = KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(testStorage.getNamespaceName()).withName(testStorage.getClusterName()).get();
        assertThat(kmm2.getStatus().getObservedGeneration(), is(2L));
        assertThat(kmm2.getMetadata().getLabels().toString(), containsString("another=label"));
        assertThat(kmm2.getSpec().getTemplate().getDeployment().getDeploymentStrategy(), is(DeploymentStrategy.ROLLING_UPDATE));
    }

    @ParallelNamespaceTest
    @KRaftNotSupported("This the is failing with KRaft and need more investigation")
    void testRestoreOffsetsInConsumerGroup(ExtensionContext extensionContext) {
        final TestStorage testStorage = new TestStorage(extensionContext, clusterOperator.getDeploymentNamespace());

        final String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        final String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        final String sourceMirroredTopicName = kafkaClusterSourceName + "." + testStorage.getTopicName();
        final String consumerGroup = ClientUtils.generateRandomConsumerGroup();

        final String sourceProducerName = testStorage.getProducerName() + "-source";
        final String sourceConsumerName = testStorage.getConsumerName() + "-source";

        final String targetProducerName = testStorage.getProducerName() + "-target";
        final String targetConsumerName = testStorage.getProducerName() + "-target";

        resourceManager.createResource(extensionContext,
            KafkaTemplates.kafkaPersistent(kafkaClusterSourceName, 1).build(),
            KafkaTemplates.kafkaPersistent(kafkaClusterTargetName, 1).build()
        );

        Map<String, Object> sourceConnectorConfig = new HashMap<>();
        sourceConnectorConfig.put("refresh.topics.interval.seconds", "1");
        sourceConnectorConfig.put("replication.factor", "1");
        sourceConnectorConfig.put("offset-syncs.topic.replication.factor", "1");

        Map<String, Object> checkpointConnectorConfig = new HashMap<>();
        checkpointConnectorConfig.put("refresh.groups.interval.seconds", "1");
        checkpointConnectorConfig.put("sync.group.offsets.enabled", "true");
        checkpointConnectorConfig.put("sync.group.offsets.interval.seconds", "1");
        checkpointConnectorConfig.put("emit.checkpoints.enabled", "true");
        checkpointConnectorConfig.put("emit.checkpoints.interval.seconds", "1");
        checkpointConnectorConfig.put("checkpoints.topic.replication.factor", "1");

        Map<String, Object> heartbeatConnectorConfig = new HashMap<>();
        heartbeatConnectorConfig.put("heartbeats.topic.replication.factor", "1");

        resourceManager.createResource(extensionContext,
            KafkaMirrorMaker2Templates.kafkaMirrorMaker2(kafkaClusterSourceName, kafkaClusterTargetName, kafkaClusterSourceName, 1, false)
                .editSpec()
                    .editFirstMirror()
                        .editSourceConnector()
                            .addToConfig(sourceConnectorConfig)
                        .endSourceConnector()
                        .editCheckpointConnector()
                            .addToConfig(checkpointConnectorConfig)
                        .endCheckpointConnector()
                        .editHeartbeatConnector()
                            .addToConfig(heartbeatConnectorConfig)
                        .endHeartbeatConnector()
                        .withTopicsPattern(".*")
                        .withGroupsPattern(".*")
                    .endMirror()
                .endSpec()
                .build(),
            KafkaMirrorMaker2Templates.kafkaMirrorMaker2(kafkaClusterTargetName, kafkaClusterSourceName, kafkaClusterTargetName, 1, false)
                .editSpec()
                    .editFirstMirror()
                        .editSourceConnector()
                            .addToConfig(sourceConnectorConfig)
                        .endSourceConnector()
                        .editCheckpointConnector()
                            .addToConfig(checkpointConnectorConfig)
                        .endCheckpointConnector()
                        .editHeartbeatConnector()
                            .addToConfig(heartbeatConnectorConfig)
                        .endHeartbeatConnector()
                        .withTopicsPattern(".*")
                        .withGroupsPattern(".*")
                    .endMirror()
                .endSpec()
                .build(),
            KafkaTopicTemplates.topic(kafkaClusterSourceName, testStorage.getTopicName(), 3).build()
        );

        KafkaClients initialInternalClientSourceJob = new KafkaClientsBuilder()
                .withProducerName(sourceProducerName)
                .withConsumerName(sourceConsumerName)
                .withBootstrapAddress(KafkaResources.plainBootstrapAddress(kafkaClusterSourceName))
                .withTopicName(testStorage.getTopicName())
                .withMessageCount(MESSAGE_COUNT)
                .withMessage("Producer A")
                .withConsumerGroup(consumerGroup)
                .withNamespaceName(testStorage.getNamespaceName())
                .build();

        KafkaClients initialInternalClientTargetJob = new KafkaClientsBuilder()
                .withProducerName(targetProducerName)
                .withConsumerName(targetConsumerName)
                .withBootstrapAddress(KafkaResources.plainBootstrapAddress(kafkaClusterTargetName))
                .withTopicName(sourceMirroredTopicName)
                .withMessageCount(MESSAGE_COUNT)
                .withConsumerGroup(consumerGroup)
                .withNamespaceName(testStorage.getNamespaceName())
                .build();

        LOGGER.info("Send & receive {} messages to/from Source cluster.", MESSAGE_COUNT);
        resourceManager.createResource(extensionContext,
            initialInternalClientSourceJob.producerStrimzi(),
            initialInternalClientSourceJob.consumerStrimzi());

        ClientUtils.waitForClientsSuccess(sourceProducerName, sourceConsumerName, testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Send {} messages to Source cluster.", MESSAGE_COUNT);
        KafkaClients internalClientSourceJob = new KafkaClientsBuilder(initialInternalClientSourceJob).withMessage("Producer B").build();

        resourceManager.createResource(extensionContext,
            internalClientSourceJob.producerStrimzi());
        ClientUtils.waitForClientSuccess(sourceProducerName, testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Wait 1 second as 'sync.group.offsets.interval.seconds=1'. As this is insignificant wait, we're skipping it");

        LOGGER.info("Receive {} messages from mirrored topic on Target cluster.", MESSAGE_COUNT);
        resourceManager.createResource(extensionContext,
            initialInternalClientTargetJob.consumerStrimzi());
        ClientUtils.waitForClientSuccess(targetConsumerName, testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Send 50 messages to Source cluster");
        internalClientSourceJob = new KafkaClientsBuilder(internalClientSourceJob).withMessageCount(50).withMessage("Producer C").build();
        resourceManager.createResource(extensionContext,
            internalClientSourceJob.producerStrimzi());
        ClientUtils.waitForClientSuccess(sourceProducerName, testStorage.getNamespaceName(), 50);

        LOGGER.info("Wait 1 second as 'sync.group.offsets.interval.seconds=1'. As this is insignificant wait, we're skipping it");
        LOGGER.info("Receive 10 msgs from source cluster");
        internalClientSourceJob = new KafkaClientsBuilder(internalClientSourceJob).withMessageCount(10).withAdditionalConfig("max.poll.records=10").build();
        resourceManager.createResource(extensionContext,
            internalClientSourceJob.consumerStrimzi());
        ClientUtils.waitForClientSuccess(sourceConsumerName, testStorage.getNamespaceName(), 10);

        LOGGER.info("Wait 1 second as 'sync.group.offsets.interval.seconds=1'. As this is insignificant wait, we're skipping it");

        LOGGER.info("Receive 40 msgs from mirrored topic on Target cluster");
        KafkaClients internalClientTargetJob = new KafkaClientsBuilder(initialInternalClientTargetJob).withMessageCount(40).build();
        resourceManager.createResource(extensionContext,
            internalClientTargetJob.consumerStrimzi());
        ClientUtils.waitForClientSuccess(targetConsumerName, testStorage.getNamespaceName(), 40);

        LOGGER.info("There should be no more messages to read. Try to consume at least 1 message. " +
                "This client job should fail on timeout.");
        resourceManager.createResource(extensionContext,
            initialInternalClientTargetJob.consumerStrimzi());
        assertDoesNotThrow(() -> ClientUtils.waitForClientTimeout(targetConsumerName, testStorage.getNamespaceName(), 1));

        LOGGER.info("As it's Active-Active MM2 mode, there should be no more messages to read from Source cluster" +
                " topic. This client job should fail on timeout.");
        resourceManager.createResource(extensionContext,
            initialInternalClientSourceJob.consumerStrimzi());
        assertDoesNotThrow(() -> ClientUtils.waitForClientTimeout(sourceConsumerName, testStorage.getNamespaceName(), 1));
    }

    @ParallelNamespaceTest
    void testKafkaMirrorMaker2ReflectsConnectorsState(ExtensionContext extensionContext) {
        final TestStorage testStorage = new TestStorage(extensionContext, clusterOperator.getDeploymentNamespace());

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";
        String errorMessage = "One or more connectors are in FAILED state";

        resourceManager.createResource(extensionContext,
            KafkaTemplates.kafkaEphemeral(kafkaClusterSourceName, 1, 1).build(),
            KafkaTemplates.kafkaEphemeral(kafkaClusterTargetName, 1, 1).build());

        resourceManager.createResource(extensionContext, false,
            KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, false)
                .editSpec()
                    .editMatchingCluster(spec -> spec.getAlias().equals(kafkaClusterSourceName))
                        // typo in bootstrap name, connectors should not connect and MM2 should be in NotReady state with error
                        .withBootstrapServers(KafkaResources.bootstrapServiceName(kafkaClusterSourceName) + ".:9092")
                    .endCluster()
                .endSpec()
                .build());

        KafkaMirrorMaker2Utils.waitForKafkaMirrorMaker2StatusMessage(testStorage.getNamespaceName(), testStorage.getClusterName(), errorMessage);

        KafkaMirrorMaker2Resource.replaceKafkaMirrorMaker2ResourceInSpecificNamespace(testStorage.getClusterName(), mm2 ->
            mm2.getSpec().getClusters().stream().filter(mm2ClusterSpec -> mm2ClusterSpec.getAlias().equals(kafkaClusterSourceName))
                .findFirst().get().setBootstrapServers(KafkaUtils.namespacedPlainBootstrapAddress(kafkaClusterSourceName, testStorage.getNamespaceName())), testStorage.getNamespaceName());

        KafkaMirrorMaker2Utils.waitForKafkaMirrorMaker2Ready(testStorage.getNamespaceName(), testStorage.getClusterName());

        KafkaMirrorMaker2Status kmm2Status = KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(testStorage.getNamespaceName()).withName(testStorage.getClusterName()).get().getStatus();
        assertFalse(kmm2Status.getConditions().stream().anyMatch(condition -> condition.getMessage() != null && condition.getMessage().contains(errorMessage)));
    }

    /**
     * Test mirroring messages by MirrorMaker 2.0 over tls transport using scram-sha-512 auth
     * while user Scram passwords, CA cluster and clients certificates are changed.
     */
    @ParallelNamespaceTest
    @KRaftNotSupported("Scram-sha is not supported by KRaft mode and is used in this test case")
    @SuppressWarnings({"checkstyle:MethodLength"})
    void testKMM2RollAfterSecretsCertsUpdateScramSha(ExtensionContext extensionContext) {
        TestStorage testStorage = new TestStorage(extensionContext);

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        String sourceMirroredTopicName = kafkaClusterSourceName + "." + testStorage.getTopicName();

        String kafkaUserSourceName = testStorage.getUserName() + "-source";
        String kafkaUserTargetName = testStorage.getUserName() + "-target";

        // Deploy source kafka with tls listener and SCRAM-SHA authentication
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaPersistent(kafkaClusterSourceName, 1, 1)
            .editSpec()
                .editKafka()
                    .withListeners(
                        new GenericKafkaListenerBuilder()
                            .withName(Constants.TLS_LISTENER_DEFAULT_NAME)
                            .withPort(9093)
                            .withType(KafkaListenerType.INTERNAL)
                            .withTls(true)
                            .withAuth(new KafkaListenerAuthenticationScramSha512())
                            .build()
                    )
                .endKafka()
            .endSpec()
            .build());

        // Deploy target kafka with tls listeners with tls and SCRAM-SHA authentication
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaPersistent(kafkaClusterTargetName, 1, 1)
            .editSpec()
                .editKafka()
                    .withListeners(
                        new GenericKafkaListenerBuilder()
                            .withName(Constants.TLS_LISTENER_DEFAULT_NAME)
                            .withPort(9093)
                            .withType(KafkaListenerType.INTERNAL)
                            .withTls(true)
                            .withAuth(new KafkaListenerAuthenticationScramSha512())
                            .build()
                    )
                .endKafka()
            .endSpec()
            .build());

        resourceManager.createResource(extensionContext,
            KafkaTopicTemplates.topic(kafkaClusterSourceName, testStorage.getTopicName()).build(),
            KafkaUserTemplates.scramShaUser(kafkaClusterSourceName, kafkaUserSourceName).build(),
            KafkaUserTemplates.scramShaUser(kafkaClusterTargetName, kafkaUserTargetName).build()
        );

        // Initialize PasswordSecretSource to set this as PasswordSecret in Source/Target MirrorMaker2 spec
        PasswordSecretSource passwordSecretSource = new PasswordSecretSource();
        passwordSecretSource.setSecretName(kafkaUserSourceName);
        passwordSecretSource.setPassword("password");

        PasswordSecretSource passwordSecretTarget = new PasswordSecretSource();
        passwordSecretTarget.setSecretName(kafkaUserTargetName);
        passwordSecretTarget.setPassword("password");

        // Initialize CertSecretSource with certificate and secret names for source
        CertSecretSource certSecretSource = new CertSecretSource();
        certSecretSource.setCertificate("ca.crt");
        certSecretSource.setSecretName(KafkaResources.clusterCaCertificateSecretName(kafkaClusterSourceName));

        // Initialize CertSecretSource with certificate and secret names for target
        CertSecretSource certSecretTarget = new CertSecretSource();
        certSecretTarget.setCertificate("ca.crt");
        certSecretTarget.setSecretName(KafkaResources.clusterCaCertificateSecretName(kafkaClusterTargetName));

        resourceManager.createResource(extensionContext, KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, true)
            .editSpec()
                .editMatchingCluster(spec -> spec.getAlias().equals(kafkaClusterSourceName))
                    .withNewKafkaClientAuthenticationScramSha512()
                        .withUsername(kafkaUserSourceName)
                        .withPasswordSecret(passwordSecretSource)
                    .endKafkaClientAuthenticationScramSha512()
                    .withNewTls()
                        .withTrustedCertificates(certSecretSource)
                    .endTls()
                .endCluster()
                .editMatchingCluster(spec -> spec.getAlias().equals(kafkaClusterTargetName))
                    .withNewKafkaClientAuthenticationScramSha512()
                        .withUsername(kafkaUserTargetName)
                        .withPasswordSecret(passwordSecretTarget)
                    .endKafkaClientAuthenticationScramSha512()
                    .withNewTls()
                        .withTrustedCertificates(certSecretTarget)
                    .endTls()
                .endCluster()
            .endSpec()
            .build());

        KafkaClients clients = new KafkaClientsBuilder()
            .withProducerName(testStorage.getProducerName())
            .withConsumerName(testStorage.getConsumerName())
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterSourceName))
            .withUserName(kafkaUserSourceName)
            .withNamespaceName(testStorage.getNamespaceName())
            .withTopicName(testStorage.getTopicName())
            .withMessageCount(MESSAGE_COUNT)
            .build();

        resourceManager.createResource(extensionContext, clients.producerScramShaTlsStrimzi(kafkaClusterSourceName), clients.consumerScramShaTlsStrimzi(kafkaClusterSourceName));
        ClientUtils.waitForClientsSuccess(testStorage.getProducerName(), testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        clients = new KafkaClientsBuilder(clients)
            .withTopicName(sourceMirroredTopicName)
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterTargetName))
            .withUserName(kafkaUserTargetName)
            .build();

        LOGGER.info("Now messages should be mirrored to target topic and cluster");

        resourceManager.createResource(extensionContext, clients.consumerScramShaTlsStrimzi(kafkaClusterTargetName));
        ClientUtils.waitForClientSuccess(testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Messages successfully mirrored");

        String kmm2DeploymentName = KafkaMirrorMaker2Resources.deploymentName(testStorage.getClusterName());
        Map<String, String> mmSnapshot = DeploymentUtils.depSnapshot(testStorage.getNamespaceName(), kmm2DeploymentName);

        LOGGER.info("Changing KafkaUser sha-password on KMM2 Source and make sure it rolled");

        Secret passwordSource = new SecretBuilder()
            .withNewMetadata()
                .withName(kafkaUserSourceName)
            .endMetadata()
            .addToData("password", "c291cmNlLXBhc3N3b3Jk")
            .build();

        kubeClient().patchSecret(testStorage.getNamespaceName(), kafkaUserSourceName, passwordSource);
        mmSnapshot = DeploymentUtils.waitTillDepHasRolled(testStorage.getNamespaceName(), kmm2DeploymentName, 1, mmSnapshot);

        LOGGER.info("Changing KafkaUser sha-password on KMM2 Target");
        Secret passwordTarget = new SecretBuilder()
            .withNewMetadata()
                .withName(kafkaUserTargetName)
            .endMetadata()
            .addToData("password", "dGFyZ2V0LXBhc3N3b3Jk")
            .build();

        kubeClient().patchSecret(testStorage.getNamespaceName(), kafkaUserTargetName, passwordTarget);
        DeploymentUtils.waitTillDepHasRolled(testStorage.getNamespaceName(), kmm2DeploymentName, 1, mmSnapshot);

        clients = new KafkaClientsBuilder(clients)
            .withTopicName(testStorage.getTopicName())
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterSourceName))
            .withUserName(kafkaUserSourceName)
            .build();

        resourceManager.createResource(extensionContext, clients.producerScramShaTlsStrimzi(kafkaClusterSourceName));
        ClientUtils.waitForClientSuccess(testStorage.getProducerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        clients = new KafkaClientsBuilder(clients)
            .withTopicName(sourceMirroredTopicName)
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterTargetName))
            .withUserName(kafkaUserTargetName)
            .build();

        LOGGER.info("Now messages should be mirrored to target topic and cluster");

        resourceManager.createResource(extensionContext, clients.consumerScramShaTlsStrimzi(kafkaClusterTargetName));
        ClientUtils.waitForClientSuccess(testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Messages successfully mirrored");
    }

    @ParallelNamespaceTest
    @SuppressWarnings({"checkstyle:MethodLength"})
    void testKMM2RollAfterSecretsCertsUpdateTLS(ExtensionContext extensionContext) {
        TestStorage testStorage = new TestStorage(extensionContext);

        String kafkaClusterSourceName = testStorage.getClusterName() + "-source";
        String kafkaClusterTargetName = testStorage.getClusterName() + "-target";

        String sourceMirroredTopicName = kafkaClusterSourceName + "." + testStorage.getTopicName();

        String kafkaUserSourceName = testStorage.getUserName() + "-source";
        String kafkaUserTargetName = testStorage.getUserName() + "-target";

        // Deploy source kafka with tls listener and mutual tls auth
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaPersistent(kafkaClusterSourceName, 1)
            .editSpec()
                .editKafka()
                    .addToConfig("min.insync.replicas", 1)
                    .withListeners(new GenericKafkaListenerBuilder()
                        .withName(Constants.TLS_LISTENER_DEFAULT_NAME)
                        .withPort(9093)
                        .withType(KafkaListenerType.INTERNAL)
                        .withTls(true)
                        .withAuth(new KafkaListenerAuthenticationTls())
                        .build())
                .endKafka()
            .endSpec()
            .build());

        // Deploy target kafka with tls listener and mutual tls auth
        resourceManager.createResource(extensionContext, KafkaTemplates.kafkaPersistent(kafkaClusterTargetName, 1)
            .editSpec()
                .editKafka()
                    .addToConfig("min.insync.replicas", 1)
                    .withListeners(new GenericKafkaListenerBuilder()
                        .withName(Constants.TLS_LISTENER_DEFAULT_NAME)
                        .withPort(9093)
                        .withType(KafkaListenerType.INTERNAL)
                        .withTls(true)
                        .withAuth(new KafkaListenerAuthenticationTls())
                        .build())
                .endKafka()
            .endSpec()
            .build());

        resourceManager.createResource(extensionContext,
            KafkaTopicTemplates.topic(kafkaClusterSourceName, testStorage.getTopicName(), 3).build(),
            KafkaUserTemplates.tlsUser(kafkaClusterSourceName, kafkaUserSourceName).build(),
            KafkaUserTemplates.tlsUser(kafkaClusterTargetName, kafkaUserTargetName).build()
        );

        // Initialize CertSecretSource with certificate and secret names for source
        CertSecretSource certSecretSource = new CertSecretSource();
        certSecretSource.setCertificate("ca.crt");
        certSecretSource.setSecretName(KafkaResources.clusterCaCertificateSecretName(kafkaClusterSourceName));

        // Initialize CertSecretSource with certificate and secret names for target
        CertSecretSource certSecretTarget = new CertSecretSource();
        certSecretTarget.setCertificate("ca.crt");
        certSecretTarget.setSecretName(KafkaResources.clusterCaCertificateSecretName(kafkaClusterTargetName));

        resourceManager.createResource(extensionContext, KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getClusterName(), kafkaClusterTargetName, kafkaClusterSourceName, 1, true)
            .editSpec()
                .editMatchingCluster(spec -> spec.getAlias().equals(kafkaClusterSourceName))
                    .withNewKafkaClientAuthenticationTls()
                        .withNewCertificateAndKey()
                            .withSecretName(kafkaUserSourceName)
                            .withCertificate("user.crt")
                            .withKey("user.key")
                        .endCertificateAndKey()
                    .endKafkaClientAuthenticationTls()
                    .withNewTls()
                        .withTrustedCertificates(certSecretSource)
                    .endTls()
                .endCluster()
                .editMatchingCluster(spec -> spec.getAlias().equals(kafkaClusterTargetName))
                    .withNewKafkaClientAuthenticationTls()
                        .withNewCertificateAndKey()
                            .withSecretName(kafkaUserTargetName)
                            .withCertificate("user.crt")
                            .withKey("user.key")
                        .endCertificateAndKey()
                    .endKafkaClientAuthenticationTls()
                    .withNewTls()
                        .withTrustedCertificates(certSecretTarget)
                    .endTls()
                .endCluster()
                .editFirstMirror()
                        .editSourceConnector()
                            .addToConfig("refresh.topics.interval.seconds", 1)
                        .endSourceConnector()
                .endMirror()
            .endSpec()
            .build());

        String mm2DeploymentName = KafkaMirrorMaker2Resources.deploymentName(testStorage.getClusterName());
        Map<String, String> mmSnapshot = DeploymentUtils.depSnapshot(testStorage.getNamespaceName(), mm2DeploymentName);

        KafkaClients clients = new KafkaClientsBuilder()
            .withProducerName(testStorage.getProducerName())
            .withConsumerName(testStorage.getConsumerName())
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterSourceName))
            .withUserName(kafkaUserSourceName)
            .withNamespaceName(testStorage.getNamespaceName())
            .withTopicName(testStorage.getTopicName())
            .withMessageCount(MESSAGE_COUNT)
            .build();

        resourceManager.createResource(extensionContext, clients.producerTlsStrimzi(kafkaClusterSourceName), clients.consumerTlsStrimzi(kafkaClusterSourceName));
        ClientUtils.waitForClientsSuccess(testStorage.getProducerName(), testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        clients = new KafkaClientsBuilder(clients)
            .withTopicName(sourceMirroredTopicName)
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterTargetName))
            .withUserName(kafkaUserTargetName)
            .build();

        LOGGER.info("Consumer in target cluster and topic should receive {} messages", MESSAGE_COUNT);

        resourceManager.createResource(extensionContext, clients.consumerTlsStrimzi(kafkaClusterTargetName));
        ClientUtils.waitForClientSuccess(testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Messages successfully mirrored");

        LabelSelector zkSourceSelector = KafkaResource.getLabelSelector(kafkaClusterSourceName, KafkaResources.zookeeperStatefulSetName(kafkaClusterSourceName));
        LabelSelector kafkaSourceSelector = KafkaResource.getLabelSelector(kafkaClusterSourceName, KafkaResources.kafkaStatefulSetName(kafkaClusterSourceName));

        LabelSelector zkTargetSelector = KafkaResource.getLabelSelector(kafkaClusterTargetName, KafkaResources.zookeeperStatefulSetName(kafkaClusterTargetName));
        LabelSelector kafkaTargetSelector = KafkaResource.getLabelSelector(kafkaClusterTargetName, KafkaResources.kafkaStatefulSetName(kafkaClusterTargetName));

        Map<String, String> kafkaSourcePods = PodUtils.podSnapshot(testStorage.getNamespaceName(), kafkaSourceSelector);
        Map<String, String> zkSourcePods = PodUtils.podSnapshot(testStorage.getNamespaceName(), zkSourceSelector);
        Map<String, String> eoSourcePods = DeploymentUtils.depSnapshot(testStorage.getNamespaceName(), KafkaResources.entityOperatorDeploymentName(kafkaClusterSourceName));
        Map<String, String> kafkaTargetPods = PodUtils.podSnapshot(testStorage.getNamespaceName(), kafkaTargetSelector);
        Map<String, String> zkTargetPods = PodUtils.podSnapshot(testStorage.getNamespaceName(), zkTargetSelector);
        Map<String, String> eoTargetPods = DeploymentUtils.depSnapshot(testStorage.getNamespaceName(), KafkaResources.entityOperatorDeploymentName(kafkaClusterTargetName));

        LOGGER.info("Renew Clients CA secret for Source cluster via annotation");
        String sourceClientsCaSecretName = KafkaResources.clientsCaCertificateSecretName(kafkaClusterSourceName);
        SecretUtils.annotateSecret(testStorage.getNamespaceName(), sourceClientsCaSecretName, Ca.ANNO_STRIMZI_IO_FORCE_RENEW, "true");
        kafkaSourcePods = RollingUpdateUtils.waitTillComponentHasRolledAndPodsReady(testStorage.getNamespaceName(), kafkaSourceSelector, 1, kafkaSourcePods);
        mmSnapshot = DeploymentUtils.waitTillDepHasRolled(testStorage.getNamespaceName(), mm2DeploymentName, 1, mmSnapshot);

        LOGGER.info("Renew Clients CA secret for Target cluster via annotation");
        String targetClientsCaSecretName = KafkaResources.clientsCaCertificateSecretName(kafkaClusterTargetName);
        SecretUtils.annotateSecret(testStorage.getNamespaceName(), targetClientsCaSecretName, Ca.ANNO_STRIMZI_IO_FORCE_RENEW, "true");
        kafkaTargetPods = RollingUpdateUtils.waitTillComponentHasRolledAndPodsReady(testStorage.getNamespaceName(), kafkaTargetSelector, 1, kafkaTargetPods);
        mmSnapshot = DeploymentUtils.waitTillDepHasRolled(testStorage.getNamespaceName(), mm2DeploymentName, 1, mmSnapshot);

        LOGGER.info("Send and receive messages after clients certs were removed");
        clients = new KafkaClientsBuilder(clients)
            .withTopicName(testStorage.getTopicName())
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterSourceName))
            .withUserName(kafkaUserSourceName)
            .build();

        resourceManager.createResource(extensionContext, clients.producerTlsStrimzi(kafkaClusterSourceName));
        ClientUtils.waitForClientSuccess(testStorage.getProducerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Consumer in target cluster and topic should receive {} messages", MESSAGE_COUNT);

        clients = new KafkaClientsBuilder(clients)
            .withTopicName(sourceMirroredTopicName)
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterTargetName))
            .withUserName(kafkaUserTargetName)
            .build();

        resourceManager.createResource(extensionContext, clients.consumerTlsStrimzi(kafkaClusterTargetName));
        ClientUtils.waitForClientSuccess(testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Messages successfully mirrored");

        LOGGER.info("Renew Cluster CA secret for Source clusters via annotation");
        String sourceClusterCaSecretName = KafkaResources.clusterCaCertificateSecretName(kafkaClusterSourceName);
        SecretUtils.annotateSecret(testStorage.getNamespaceName(), sourceClusterCaSecretName, Ca.ANNO_STRIMZI_IO_FORCE_RENEW, "true");

        if (!Environment.isKRaftModeEnabled()) {
            RollingUpdateUtils.waitTillComponentHasRolledAndPodsReady(testStorage.getNamespaceName(), zkSourceSelector, 1, zkSourcePods);
        }
        RollingUpdateUtils.waitTillComponentHasRolledAndPodsReady(testStorage.getNamespaceName(), kafkaSourceSelector, 1, kafkaSourcePods);
        DeploymentUtils.waitTillDepHasRolled(testStorage.getNamespaceName(), KafkaResources.entityOperatorDeploymentName(kafkaClusterSourceName), 1, eoSourcePods);
        mmSnapshot = DeploymentUtils.waitTillDepHasRolled(testStorage.getNamespaceName(), mm2DeploymentName, 1, mmSnapshot);

        LOGGER.info("Renew Cluster CA secret for Target clusters via annotation");
        String targetClusterCaSecretName = KafkaResources.clusterCaCertificateSecretName(kafkaClusterTargetName);
        SecretUtils.annotateSecret(testStorage.getNamespaceName(), targetClusterCaSecretName, Ca.ANNO_STRIMZI_IO_FORCE_RENEW, "true");

        if (!Environment.isKRaftModeEnabled()) {
            RollingUpdateUtils.waitTillComponentHasRolledAndPodsReady(testStorage.getNamespaceName(), zkTargetSelector, 1, zkTargetPods);
        }
        RollingUpdateUtils.waitTillComponentHasRolledAndPodsReady(testStorage.getNamespaceName(), kafkaTargetSelector, 1, kafkaTargetPods);
        DeploymentUtils.waitTillDepHasRolled(testStorage.getNamespaceName(), KafkaResources.entityOperatorDeploymentName(kafkaClusterTargetName), 1, eoTargetPods);
        DeploymentUtils.waitTillDepHasRolled(testStorage.getNamespaceName(), mm2DeploymentName, 1, mmSnapshot);

        LOGGER.info("Send and receive messages after clients certs were removed");
        clients = new KafkaClientsBuilder(clients)
            .withTopicName(testStorage.getTopicName())
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterSourceName))
            .withUserName(kafkaUserSourceName)
            .build();

        resourceManager.createResource(extensionContext, clients.producerTlsStrimzi(kafkaClusterSourceName));
        ClientUtils.waitForClientSuccess(testStorage.getProducerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Consumer in target cluster and topic should receive {} messages", MESSAGE_COUNT);

        clients = new KafkaClientsBuilder(clients)
            .withTopicName(sourceMirroredTopicName)
            .withBootstrapAddress(KafkaResources.tlsBootstrapAddress(kafkaClusterTargetName))
            .withUserName(kafkaUserTargetName)
            .build();

        resourceManager.createResource(extensionContext, clients.consumerTlsStrimzi(kafkaClusterTargetName));
        ClientUtils.waitForClientSuccess(testStorage.getConsumerName(), testStorage.getNamespaceName(), MESSAGE_COUNT);

        LOGGER.info("Messages successfully mirrored");
    }

    private void testDockerImagesForKafkaMirrorMaker2(String clusterName, String clusterOperatorNamespace, String mirrorMakerNamespace) {
        LOGGER.info("Verifying docker image names");
        // we must use INFRA_NAMESPACE because there is CO deployed
        Map<String, String> imgFromDeplConf = getImagesFromConfig(clusterOperatorNamespace);
        // Verifying docker image for kafka mirrormaker2
        String mirrormaker2ImageName = PodUtils.getFirstContainerImageNameFromPod(mirrorMakerNamespace, kubeClient(mirrorMakerNamespace).listPods(mirrorMakerNamespace, clusterName, Labels.STRIMZI_KIND_LABEL, KafkaMirrorMaker2.RESOURCE_KIND)
            .get(0).getMetadata().getName());

        String mirrormaker2Version = KafkaMirrorMaker2Resource.kafkaMirrorMaker2Client().inNamespace(mirrorMakerNamespace).withName(clusterName).get().getSpec().getVersion();
        if (mirrormaker2Version == null) {
            mirrormaker2Version = Environment.ST_KAFKA_VERSION;
        }

        assertThat(TestUtils.parseImageMap(imgFromDeplConf.get(KAFKA_MIRROR_MAKER_2_IMAGE_MAP)).get(mirrormaker2Version), is(mirrormaker2ImageName));
        LOGGER.info("Docker images verified");
    }

    @BeforeAll
    void setup(ExtensionContext extensionContext) {
        clusterOperator.unInstall();
        clusterOperator = new SetupClusterOperator.SetupClusterOperatorBuilder()
            .withExtensionContext(BeforeAllOnce.getSharedExtensionContext())
            .withNamespace(Constants.INFRA_NAMESPACE)
            .withWatchingNamespaces(Constants.WATCH_ALL_NAMESPACES)
            .withOperationTimeout(Constants.CO_OPERATION_TIMEOUT_SHORT)
            .createInstallation()
            .runInstallation();
    }
}

redpajama_set_name: RedPajamaGithub


If you have chemicals that you would like to export to another country for cash return, what exact country should you attempt to export them to? You have a broad choice of countries that would likely welcome the chemicals that you have, but we would eagerly suggest that you don't overlook Malaysia. This country in Southeast Asia can be too easy to overlook for people seeking to make money through use of services from one of the UK's most esteemed chemical transport companies, Wake Marine. Below, we list particularly good incentives to strongly consider Malaysia.
We will start by pointing out that, in recent decades, Malaysia has particularly thrived economically and, therefore, is particularly likely to have money to give you in return for taking chemicals off your hands. More precisely, for nearly five decades, the country's GDP has grown by an average of 6.5% per annum. Malaysia currently has what is known as a newly industrialised market economy, and its economy is ranked as Southeast Asia's third biggest and the world's 28th biggest. Malaysia has also been tipped to become a developed country within the next few years.
Chemicals appear in the list of goods that Malaysia imports especially large amounts of, making the country seem especially apt to chemical transport to. Wake Marine is also one of the most suitable chemical shipping companies to export to Malaysia with assistance from; you can find, elsewhere on this website, plenty of information about why. Though Malaysia's top five import partners – according to 2012 estimates – are China, Singapore, Japan, the United States and Thailand, the country is likely to also welcome chemicals of the right type and quality from the UK.
The country's demand for chemicals is undoubtedly likely encouraged by its relatively large population. This population has been recorded in Malaysia's 2010 census as 28,334,135, making Malaysia the world's 42nd most populous country. The population also continues growing at the rate of 2.4% annually, which suggests that you could greatly benefit from repeatedly exporting more chemicals to the country long into the future. You could find, to your pleasant surprise, that a wide range of chemical products, including lube oil, are warmly welcomed in this Southeast Asia country - and don't underestimate how easily you could get such products transported with assistance from the liquid transport business Wake Marine.


Daisy was one of those dogs that everyone loved, and I was no exception. I met her and her 'mom' Paula in the early 2000s and looked forward to their visits to my practice. Sometimes there was nothing wrong with Daisy, but Paula enjoyed coming for a little chat, especially after her husband passed away.
One wintery weekend I was away visiting my sister when Paula called.
She sounded alarmed because Daisy, who was absolutely fine an hour ago, was now in severe distress. Paula told me Daisy ate her full meal and played in the backyard with no signs of a problem and now her breathing was shallow and her abdomen was severely distended. She had all the signs of stomach bloat.
I immediately recognized Paula was describing the symptoms of Gastric Dilation Volvulus (GDV), which is also known as stomach bloat. I asked her not to wait a minute and take her to the local emergency hospital. Sadly, Daisy died in surgery and Paula and I were devastated.
Even many years later, I can't stop feeling that sinking sensation I had when I learned that Daisy died. I also know there are many dogs each year who die from this mostly preventable condition.
My hope is Chapter 14 of the Health and Longevity Course for Dogs will help you protect your dog and you'll share it with others to help them do the same.
Most veterinary textbooks state that GDV is a serious, life-threatening condition caused by fermentation of food, gas build-up and twist, or torsion, of the stomach, which causes severe distention and blood circulation collapse and shock.
However, these books never mention two factors that, as I have discovered, play an important role in GDV occurrence.
In nature, dogs have evolved to digest the carcasses of prey. They are equipped with a highly effective set of teeth for ripping, shearing, crushing and crunching meat and bones. Their digestive tract is ideal for digesting raw meat protein and the acidic pH of their stomach protects them from pathogenic bacteria. Stomach acids are also the reason why wild and domestic canines are capable of digesting raw bones.
Unfortunately, most dogs today are fed kibble, highly processed and ground food that turns into a slimy porridge soon after it reaches the stomach.
Over time the stomach muscles become very weak. The muscles don't need to work as hard due to the lack of natural roughage in the food, such as raw bones and meat. 'Kibble stomach' is the equivalent of a lazy and weak couch potato that is more predisposed to gastric dilation.
As if this was not enough, kibble and processed food are much more likely to cause fermentation and gas build-up, which increases the chance of bloat even more. Grain and carbohydrate-based kibble are especially prone to producing gas.
Did you know that dogs should never be fed fruit and protein together? The main reason is the digestive time for fruit and protein is different.
Fruit digestion time in the stomach is much shorter than protein. If they are fed together, the stomach does not empty until the protein portion is digested, which can cause fermentation and gas build-up.
I have heard some people say the fermentation process causes alcohol production, which can make your dog drunk, but I have never seen this.
Ideally, one should feed fruit at least one hour or longer before a meal and at least four hours after eating.
The general consensus is you should wait at least four hours after eating to exercise your dog. When your dog's stomach is full, it is much more likely to twist and turn when they make a sudden movement.
There are still many people feeding their dogs and taking them for a walk and a play in a park and you and I need to caution them about the risks.
This article gives more details on exercising dogs safely.
In Chapter 5 and 6 of the Health and Longevity Course for Dogs, I mention in detail why natural vitamins, plant-based minerals, amino acids and omega oils are key to optimal organ function. We now know that the body can't produce many of these nutrients on its own and they need to come from the soil.
Unfortunately, agricultural soils are severely nutrient depleted and those deficiencies are passed on to our dogs. Such deficiencies can also reduce the vitality, strength and mobility of the stomach wall and logically increases the chances of bloat.
In Chapter 7 of the Health and Longevity Course for Dogs I write about the energy flow of the body and how important it is to organ function.
After years of observation, I have found a very close connection between the thoracic-lumbar junction of the spine and the stomach. The stomach association points are right behind the last rib on both sides of the spine.
Dogs with a tendency for stomach problems often have inflammation and sensitivity in this area and sometimes this inflammation leads to spondylosis, a form of spinal arthritis.
The interesting part of this is that my emergency clinic colleagues confirmed that they often see spondylosis at the stomach association point.
This means that regular monthly check-ups are an important part of GDV prevention, especially if you see your dog is sensitive along the spine behind the last rib.
I find many modalities are helpful and the results depend greatly on the skills of the practitioner. Physiotherapy, chiropractic, osteopathy, massage, intramuscular needle stimulation treatment (IMS) and acupuncture are just a few examples of practitioners you can consult.
What to do if you suspect bloat?
First, panting is not a sign of bloat on its own. However, if it comes with severe distress, pale gums, stomach distention and excessive salivation, you must rush your dog to the nearest vet or emergency clinic immediately.
As a first aid measure on the way to the hospital, I recommend giving the homeopathic remedy Nux Vomica or Carbo Vegetabilis in 30C or 200C potency.
After your arrival, a light sedation may be required and a stomach tube will be inserted to release the gas immediately. This may also be achieved by using a large needle directly through the abdominal wall.
The most life-threatening part of GDV is a circulatory collapse and shock and these steps can be lifesaving.
After this, radiographs will usually follow and in some cases, an emergency surgery is required.
Is preventive surgery gastropexy good for preventing GDV?
Gastropexy is a commonly recommended preventive procedure, where the stomach wall is attached to the inside of the rib cage to prevent the stomach from flipping.
Personally, I'm not in favor of gastropexy because it restricts the natural movement and function of the stomach. Also, any surgical intervention affects the body's energy channels and the natural state of the body is the best.
I feel very grateful that by following the precautions mentioned in the article, not a single patient of mine suffered from GDV since Daisy passed away. In her case, it was the exercise right after feeding that was the cause.
The best way to protect your dog from GDV is to feed a natural raw or cooked diet and raw bones, give essential supplements, feed fruit separately from protein, provide safe exercise and ensure good spinal energy flow.


text:
Select Movie / All Movies Sword Art Online Progressive: Aria Of A Starless Night The King's Man Sing 2 Spider-Man: No Way Home Makmum 2 The 355 Resident Evil: Welcome To Raccoon City House Of Gucci The French Dispatch G Storm Till We Meet Again Spider-Man: No Way Home (D-BOX) Free! The Final Stroke - The First Volume Embrace Again The Policeman's Lineage Velan Enna Solla Pogirai Paw Patrol: The Movie No Time To Die Venom: Let There Be Carnage Wrath Of Man Nightmare Alley Belle The Matrix Resurrections Spider-Man: No Way Home (IMAX) The Matrix Resurrections (IMAX) The King's Man (IMAX) Encanto Theal Naai Sekar The Medium Ghostbusters: Afterlife 83 Spider-Man: No Way Home (ATMOS) Prebet Sapu Spider-Man: No Way Home (Family Friendly) Sing 2 (Family Friendly) Paw Patrol: The Movie (Family Friendly)
Disney headquarters say no to "Beauty and the Beast" cuts?
Writer: Florey DM
Say adieu to LeFou, "Beauty and the Beast" may never be screened in Malaysia.
15 Mar – The fate of "Beauty and the Beast" in local cinemas is very much undetermined at this point but if latest word is to be believed, the movie may never make it to the big screens in Malaysia.
While yesterday it was cleared by the Malaysian Censorship Board (LPF) that the movie is good to go and can be screened in cinemas nationwide as long as the "gay moments" were cut from the movie, Bloomberg recently updated that in an email statement, Walt Disney Co. revealed the following:
"The film has not been and will not be cut for Malaysia."
Rather than allow the movie to suffer an approximately 4 minutes and 30 seconds of cuts, the company has decided to just pull the movie from the Malaysian market.
The Motion Picture Association of America stated that Malaysia is not one of the biggest film markets in the world, hence Disney's lack of apprehension in dropping the movie from the country's listings.
However, when contacted by Cinema Online regarding the supposed "no show", Walt Disney Malaysia stated that there is currently "no confirmation on anything as of yet".
It can be assumed that the movie is still undergoing internal review, as what was announced by the Walt Disney Malaysia two days ago, and its true fate at the local market has yet to be determined.
Cinema Online, 15 March 2017
Beauty And The Beast (30 Mar 2017)
Latest Trailers:
Be enchanted by the "Beauty and the Beast" characters
Emma Watson dazzles as Belle while Dan Stevens reveals both his Beast and Prince looks
"Beauty and the Beast" greenlit, awaiting release date
No new release date has been set despite approval from the Malaysian Censorship Board (LPF)
"Beauty and the Beast" crosses 1 billion mark
Meanwhile director Bill Condon already has his eyes set on another movie remake
"Logan" actor was cut from "Beauty and the Beast"
Did you know that this "Logan" actor was in the live-action "Beauty and the Beast"?
"Beauty and the Beast" postponed in Malaysia
Disney has confirmed that the movie is delayed, no longer releasing on this 16 March
Disney Malaysia appeals "Beauty and the Beast" cut
The movie is currently postponed until 30 March while awaiting decision from LPF
Final appeal for "Beauty and the Beast" censorship taking place today
LPF will decide today whether the Disney movie will be screened in Malaysia or not
"Beauty and the Beast" is back on in Malaysia!
Disney has confirmed that the movie can be screened in local cinemas without any cuts
Josh Gad, Luke Evans returning for "Beauty and the Beast" prequel
A prequel series focusing on Gaston and LeFou has been ordered by Disney+
10 things you should know before watching "Beauty and the Beast"
We list some things that have and haven't changed about the live adaptation

redpajama_set_name: RedPajamaCommonCrawl


Factory finishes including Gloss, Matte, Satin, Eggshell, Frozen etc. We have a vast selection of OEM and aftermarket color codes.
We do accept insurance claims in addition to our standard collision repair services. We do our best to work with most providers and estimates, and will make estimate supplements as necessary.
Every job has a unique own approach depending on current vehicle state.


text:
La trentasettesima edizione della gara maschile Junior dei Campionati del mondo di ciclocross 2015 si svolse il 31 gennaio 2015 con partenza ed arrivo da Tábor in Repubblica Ceca, su un percorso iniziale di 160 mt più un circuito di 3,11 km da ripetere 5 volte per un totale di 15,71 km. La vittoria fu appannaggio del danese Simon Andreassen, il quale terminò la gara in 42'24", alla media di 22,23 km/h, precedendo il belga Eli Iserbyt e l'olandese Max Gulickx terzo.

Partenza con 71 ciclisti provenienti da 19 nazioni, dei quali 67 portarono a termine la competizione.

Squadre partecipanti

Classifica (Top 10)

Collegamenti esterni
  Risultati su archivio UCI

Campionati del mondo di ciclocross 2015

redpajama_set_name: RedPajamaWikipedia


Unobtrusive and respectful of tradition, Tagmarshal's Pace of Play management system is the go-to-solution for clubs looking to excel in their player experience.
While member days are managed with Tagmarshal (Busy days are Thursday / Saturday / Sunday) – Monday's are only passively tracked, as the club rent out their course to a third party management company on Mondays.
These managed days are very steady – trend declining.


Chevron | Ribbon And Bows Oh My!
Our selection of chevron ribbon includes over two hundred different color options in every width. All made in the USA. This trend still continues to be an industry leader in hair bows and accessories.