How to Migrate On-Prem Kafka Clusters to Confluent Cloud 2026 and AWS MSK 3.0 Without Downtime

import logging
import sys
import time
from dataclasses import dataclass
from typing import Dict, List, Optional

from confluent_kafka import KafkaException
from confluent_kafka.admin import AdminClient, NewTopic, ConfigResource, ConfigSource

# Configure structured logging for migration audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

@dataclass
class ClusterValidationConfig:
    bootstrap_servers: str
    security_protocol: str = "SSL"
    ssl_cafile: str = "/etc/kafka/secrets/ca-cert.pem"
    ssl_certfile: str = "/etc/kafka/secrets/client-cert.pem"
    ssl_keyfile: str = "/etc/kafka/secrets/client-key.pem"
    required_min_brokers: int = 3
    required_kafka_version: str = "3.6.0"

class OnPremKafkaValidator:
    def __init__(self, config: ClusterValidationConfig):
        self.config = config
        self.admin_client = self._init_admin_client()
        self.validation_errors: List[str] = []

    def _init_admin_client(self) -> AdminClient:
        """Initialize Kafka AdminClient with TLS config for on-prem cluster"""
        admin_config = {
            "bootstrap.servers": self.config.bootstrap_servers,
            "security.protocol": self.config.security_protocol,
            "ssl.ca.location": self.config.ssl_cafile,
            "ssl.certificate.location": self.config.ssl_certfile,
            "ssl.key.location": self.config.ssl_keyfile,
            "client.id": "migration-validator-001"
        }
        try:
            return AdminClient(admin_config)
        except KafkaException as e:
            logger.error(f"Failed to initialize AdminClient: {e}")
            sys.exit(1)

    def validate_broker_count(self) -> bool:
        """Check that on-prem cluster has at least required_min_brokers"""
        try:
            cluster_metadata = self.admin_client.list_topics(timeout=10)
            broker_count = len(cluster_metadata.brokers)
            logger.info(f"Detected {broker_count} brokers in on-prem cluster")
            if broker_count < self.config.required_min_brokers:
                self.validation_errors.append(
                    f"Insufficient brokers: {broker_count} < {self.config.required_min_brokers}"
                )
                return False
            return True
        except KafkaException as e:
            self.validation_errors.append(f"Broker count check failed: {e}")
            return False

    def validate_kafka_version(self) -> bool:
        """Verify all brokers run required Kafka version"""
        try:
            cluster_metadata = self.admin_client.list_topics(timeout=10)
            for broker_id, broker_metadata in cluster_metadata.brokers.items():
                # Broker metadata version string format: x.y.z-commit
                broker_version = broker_metadata.version if hasattr(broker_metadata, "version") else "unknown"
                if not broker_version.startswith(self.config.required_kafka_version):
                    self.validation_errors.append(
                        f"Broker {broker_id} runs unsupported version {broker_version}"
                    )
                    return False
            logger.info(f"All brokers run required version {self.config.required_kafka_version}")
            return True
        except KafkaException as e:
            self.validation_errors.append(f"Version check failed: {e}")
            return False

    def validate_topic_compatibility(self) -> bool:
        """Check that all topics have replication factor >= 2 for migration safety"""
        try:
            topics = self.admin_client.list_topics(timeout=10).topics
            for topic_name, topic_metadata in topics.items():
                if topic_metadata.partitions:
                    for partition_id, partition_metadata in topic_metadata.partitions.items():
                        replication_factor = len(partition_metadata.replicas)
                        if replication_factor < 2:
                            self.validation_errors.append(
                                f"Topic {topic_name} partition {partition_id} has replication factor {replication_factor}"
                            )
            if any("replication factor" in error for error in self.validation_errors):
                return False
            logger.info(f"Validated {len(topics)} topics for replication factor compliance")
            return True
        except KafkaException as e:
            self.validation_errors.append(f"Topic validation failed: {e}")
            return False

    def run_all_validations(self) -> bool:
        """Execute all pre-migration checks and return pass/fail status"""
        logger.info("Starting on-prem Kafka cluster pre-migration validation")
        checks = [
            self.validate_broker_count,
            self.validate_kafka_version,
            self.validate_topic_compatibility
        ]
        all_passed = True
        for check in checks:
            if not check():
                all_passed = False
        if not all_passed:
            logger.error(f"Validation failed with {len(self.validation_errors)} errors:")
            for error in self.validation_errors:
                logger.error(f" - {error}")
            return False
        logger.info("All pre-migration validations passed successfully")
        return True

if __name__ == "__main__":
    # Load config from environment variables or config file in production
    validator_config = ClusterValidationConfig(
        bootstrap_servers="on-prem-kafka-01:9093,on-prem-kafka-02:9093,on-prem-kafka-03:9093",
        required_min_brokers=15  # Match our 15-broker on-prem cluster
    )
    validator = OnPremKafkaValidator(validator_config)
    if not validator.run_all_validations():
        sys.exit(1)
    logger.info("Pre-migration validation complete. Proceeding to migration setup.")

import json
import logging
import os
import sys
from typing import Dict, Any, Optional
from pathlib import Path

from confluent_kafka import KafkaException
from confluent_kafka.admin import AdminClient

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class MirrorMaker2ConfigGenerator:
    """Generates versioned MirrorMaker 2.0 configs for Confluent Cloud 2026 and AWS MSK 3.0"""

    def __init__(self, target_platform: str, on_prem_config: Dict[str, Any], target_config: Dict[str, Any]):
        """
        :param target_platform: Either "confluent_cloud_2026" or "aws_msk_3.0"
        :param on_prem_config: Config for source on-prem Kafka cluster
        :param target_config: Config for target managed Kafka cluster
        """
        self.target_platform = target_platform
        self.on_prem_config = on_prem_config
        self.target_config = target_config
        self._validate_target_platform()

    def _validate_target_platform(self) -> None:
        valid_platforms = ["confluent_cloud_2026", "aws_msk_3.0"]
        if self.target_platform not in valid_platforms:
            raise ValueError(f"Invalid target platform {self.target_platform}. Valid: {valid_platforms}")

    def _get_on_prem_source_config(self) -> Dict[str, str]:
        """Return source connector config for on-prem Kafka cluster"""
        return {
            "name": "on-prem-kafka-source",
            "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
            "source.cluster.alias": "on-prem",
            "source.bootstrap.servers": self.on_prem_config["bootstrap_servers"],
            "source.security.protocol": self.on_prem_config.get("security_protocol", "SSL"),
            "source.ssl.ca.location": self.on_prem_config.get("ssl_cafile"),
            "source.ssl.certificate.location": self.on_prem_config.get("ssl_certfile"),
            "source.ssl.key.location": self.on_prem_config.get("ssl_keyfile"),
            "topics": ".*",  # Replicate all topics, filter later if needed
            "tasks.max": str(self.on_prem_config.get("tasks_max", 30)),  # 2 tasks per broker for 15-broker cluster
            "replication.factor": "3",  # Match target cluster replication factor
            "offset-syncs.topic.replication.factor": "3",
            "heartbeats.topic.replication.factor": "3",
            "checkpoints.topic.replication.factor": "3"
        }

    def _get_confluent_cloud_target_config(self) -> Dict[str, str]:
        """Add Confluent Cloud 2026 specific configs (MRC, private link)"""
        if self.target_platform != "confluent_cloud_2026":
            return {}
        return {
            "target.bootstrap.servers": self.target_config["bootstrap_servers"],
            "target.security.protocol": "SASL_SSL",
            "target.sasl.mechanism": "PLAIN",
            "target.sasl.username": self.target_config["api_key"],
            "target.sasl.password": self.target_config["api_secret"],
            "target.confluent.mrc.enabled": "true",  # Enable Multi-Region Cluster for 2026
            "target.confluent.private.link.enabled": "true",  # Use Private Link for on-prem connectivity
            "target.replication.factor": "3",
            "target.confluent.tiering.enabled": "false"  # Disable tiering for high-throughput topics
        }

    def _get_msk_target_config(self) -> Dict[str, str]:
        """Add AWS MSK 3.0 specific configs (tiered storage, IAM auth)"""
        if self.target_platform != "aws_msk_3.0":
            return {}
        return {
            "target.bootstrap.servers": self.target_config["bootstrap_servers"],
            "target.security.protocol": "SSL",
            "target.ssl.ca.location": self.target_config.get("ssl_cafile", "/etc/ssl/certs/aws-msk-ca.pem"),
            "target.ssl.certificate.location": self.target_config.get("ssl_certfile"),
            "target.ssl.key.location": self.target_config.get("ssl_keyfile"),
            "target.aws.iam.auth.enabled": "true",  # Use MSK 3.0 IAM authentication
            "target.aws.tiered.storage.enabled": "true",  # Enable MSK 3.0 tiered storage
            "target.replication.factor": "3",
            "target.aws.tiered.storage.segment.bytes": "1073741824"  # 1GB segments for tiering
        }

    def generate_full_config(self) -> Dict[str, str]:
        """Generate complete MirrorMaker 2.0 connector config"""
        full_config = self._get_on_prem_source_config()
        if self.target_platform == "confluent_cloud_2026":
            full_config.update(self._get_confluent_cloud_target_config())
        elif self.target_platform == "aws_msk_3.0":
            full_config.update(self._get_msk_target_config())

        # Add common target configs
        full_config.update({
            "sync.topic.acls.enabled": "false",  # Disable ACL sync for initial migration
            "emit.checkpoints.interval.seconds": "30",
            "emit.heartbeats.interval.seconds": "10",
            "offset-syncs.commit.interval.seconds": "60"
        })
        return full_config

    def write_to_file(self, output_path: str) -> None:
        """Write generated config to JSON file for Kafka Connect deployment"""
        try:
            config = self.generate_full_config()
            output_file = Path(output_path)
            output_file.parent.mkdir(parents=True, exist_ok=True)
            with open(output_file, "w") as f:
                json.dump(config, f, indent=2)
            logger.info(f"Wrote MirrorMaker 2.0 config to {output_path}")
        except IOError as e:
            logger.error(f"Failed to write config file: {e}")
            sys.exit(1)

if __name__ == "__main__":
    # Example on-prem config (replace with production values)
    on_prem_config = {
        "bootstrap_servers": "on-prem-kafka-01:9093,on-prem-kafka-02:9093",
        "security_protocol": "SSL",
        "ssl_cafile": "/etc/kafka/secrets/ca-cert.pem",
        "ssl_certfile": "/etc/kafka/secrets/client-cert.pem",
        "ssl_keyfile": "/etc/kafka/secrets/client-key.pem",
        "tasks_max": 30
    }

    # Example Confluent Cloud 2026 target config
    confluent_target_config = {
        "bootstrap_servers": "pkc-xxxxx.us-east-1.aws.confluent.cloud:9092",
        "api_key": os.environ.get("CONFLUENT_API_KEY"),
        "api_secret": os.environ.get("CONFLUENT_API_SECRET")
    }

    # Generate Confluent Cloud config
    confluent_generator = MirrorMaker2ConfigGenerator(
        target_platform="confluent_cloud_2026",
        on_prem_config=on_prem_config,
        target_config=confluent_target_config
    )
    confluent_generator.write_to_file("/tmp/mm2-confluent-cloud-2026-config.json")

    # Example AWS MSK 3.0 target config
    msk_target_config = {
        "bootstrap_servers": "b-1.mskcluster.xyz.c3.kafka.us-east-1.amazonaws.com:9094",
        "ssl_cafile": "/etc/ssl/certs/aws-msk-ca.pem",
        "ssl_certfile": "/etc/kafka/secrets/msk-client-cert.pem",
        "ssl_keyfile": "/etc/kafka/secrets/msk-client-key.pem"
    }

    # Generate MSK config
    msk_generator = MirrorMaker2ConfigGenerator(
        target_platform="aws_msk_3.0",
        on_prem_config=on_prem_config,
        target_config=msk_target_config
    )
    msk_generator.write_to_file("/tmp/mm2-msk-3.0-config.json")

    logger.info("MirrorMaker 2.0 config generation complete for both targets.")

import logging
import os
import sys
import time
from typing import Dict, List, Optional
from dataclasses import dataclass

from confluent_kafka import Consumer, KafkaException, TopicPartition
from confluent_kafka.admin import AdminClient

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

@dataclass
class CutoverConfig:
    on_prem_bootstrap: str
    target_bootstrap: str
    consumer_group_id: str
    topic_pattern: str = ".*"
    max_allowed_lag: int = 100  # Max 100 offsets lag during cutover
    cutover_dns_record: str = "kafka.prod.example.com"
    dns_ttl: int = 60  # 60 second TTL for fast rollback

class TrafficCutoverManager:
    def __init__(self, config: CutoverConfig):
        self.config = config
        self.on_prem_admin = self._init_admin(self.config.on_prem_bootstrap, is_target=False)
        self.target_admin = self._init_admin(self.config.target_bootstrap, is_target=True)
        self.cutover_errors: List[str] = []

    def _init_admin(self, bootstrap: str, is_target: bool) -> AdminClient:
        """Initialize AdminClient for source or target cluster"""
        admin_config = {
            "bootstrap.servers": bootstrap,
            "security.protocol": "SSL" if not is_target else "SASL_SSL",
            "client.id": "cutover-manager-001"
        }
        if is_target:
            # Confluent Cloud uses SASL_SSL, MSK uses SSL; adjust per target
            admin_config["sasl.mechanism"] = "PLAIN"
            admin_config["sasl.username"] = os.environ.get("TARGET_API_KEY")
            admin_config["sasl.password"] = os.environ.get("TARGET_API_SECRET")
        else:
            admin_config["ssl.ca.location"] = "/etc/kafka/secrets/ca-cert.pem"
            admin_config["ssl.certificate.location"] = "/etc/kafka/secrets/client-cert.pem"
            admin_config["ssl.key.location"] = "/etc/kafka/secrets/client-key.pem"
        try:
            return AdminClient(admin_config)
        except KafkaException as e:
            logger.error(f"Failed to init admin for {bootstrap}: {e}")
            sys.exit(1)

    def validate_offset_alignment(self) -> bool:
        """Check that target cluster has all messages from source for consumer group"""
        try:
            # Get consumer group offsets from on-prem
            on_prem_offsets = self._get_consumer_group_offsets(self.on_prem_admin)
            # Get consumer group offsets from target (MirrorMaker syncs offsets)
            target_offsets = self._get_consumer_group_offsets(self.target_admin)

            for topic_partition, on_prem_offset in on_prem_offsets.items():
                target_offset = target_offsets.get(topic_partition)
                if not target_offset:
                    self.cutover_errors.append(f"Target missing offset for {topic_partition}")
                    return False
                if target_offset < on_prem_offset:
                    self.cutover_errors.append(
                        f"Target offset {target_offset} < on-prem offset {on_prem_offset} for {topic_partition}"
                    )
                    return False
            logger.info(f"Offset alignment validated for {len(on_prem_offsets)} topic partitions")
            return True
        except KafkaException as e:
            self.cutover_errors.append(f"Offset validation failed: {e}")
            return False

    def _get_consumer_group_offsets(self, admin: AdminClient) -> Dict[TopicPartition, int]:
        """Fetch committed offsets for a consumer group"""
        try:
            committed = admin.list_consumer_group_offsets([self.config.consumer_group_id])
            offsets = {}
            for group_id, group_offsets in committed.items():
                for topic_partition, offset_metadata in group_offsets.items():
                    if offset_metadata.offset != -1:  # -1 means no committed offset
                        offsets[TopicPartition(topic_partition.topic, topic_partition.partition)] = offset_metadata.offset
            return offsets
        except KafkaException as e:
            raise KafkaException(f"Failed to fetch offsets: {e}")

    def check_consumer_lag(self) -> bool:
        """Verify consumer lag on target cluster is below max allowed"""
        try:
            consumer_config = {
                "bootstrap.servers": self.config.target_bootstrap,
                "group.id": self.config.consumer_group_id,
                "security.protocol": "SASL_SSL",
                "sasl.mechanism": "PLAIN",
                "sasl.username": os.environ.get("TARGET_API_KEY"),
                "sasl.password": os.environ.get("TARGET_API_SECRET"),
                "enable.auto.commit": False
            }
            consumer = Consumer(consumer_config)
            # Get end offsets for all topics in pattern
            metadata = self.target_admin.list_topics(timeout=10)
            topic_partitions = []
            for topic_name, topic_meta in metadata.topics.items():
                if re.match(self.config.topic_pattern, topic_name):
                    for partition_id in topic_meta.partitions.keys():
                        topic_partitions.append(TopicPartition(topic_name, partition_id))
            # Get end offsets
            end_offsets = consumer.end_offsets(topic_partitions)
            # Get committed offsets
            committed_offsets = self._get_consumer_group_offsets(self.target_admin)
            # Calculate lag
            total_lag = 0
            for tp, end_offset in end_offsets.items():
                committed_offset = committed_offsets.get(tp, 0)
                lag = end_offset - committed_offset
                total_lag += lag
            logger.info(f"Total consumer lag on target: {total_lag}")
            if total_lag > self.config.max_allowed_lag:
                self.cutover_errors.append(f"Lag {total_lag} exceeds max {self.config.max_allowed_lag}")
                return False
            return True
        except Exception as e:
            self.cutover_errors.append(f"Lag check failed: {e}")
            return False

    def execute_dns_cutover(self) -> bool:
        """Switch DNS record to point to target cluster (stub for AWS Route53)"""
        try:
            # In production, use boto3 for Route53 or Cloudflare API
            logger.info(f"Cutting over DNS {self.config.cutover_dns_record} to target cluster")
            # Stub implementation: log instead of actual API call
            logger.info(f"DNS TTL set to {self.config.dns_ttl}s for fast rollback")
            logger.info(f"DNS record updated successfully")
            return True
        except Exception as e:
            self.cutover_errors.append(f"DNS cutover failed: {e}")
            return False

    def run_cutover(self) -> bool:
        """Execute full cutover process with rollback on failure"""
        logger.info("Starting traffic cutover process")
        checks = [self.validate_offset_alignment, self.check_consumer_lag]
        for check in checks:
            if not check():
                logger.error(f"Cutover check failed: {self.cutover_errors}")
                self.rollback()
                return False
        if not self.execute_dns_cutover():
            self.rollback()
            return False
        logger.info("Traffic cutover complete. Monitoring for 5 minutes.")
        time.sleep(300)  # Monitor for 5 minutes
        # Final lag check post cutover
        if not self.check_consumer_lag():
            self.rollback()
            return False
        logger.info("Cutover validated successfully. Migration complete.")
        return True

    def rollback(self) -> None:
        """Revert DNS to on-prem cluster"""
        logger.warning("Initiating rollback to on-prem cluster")
        # Stub for DNS rollback
        logger.info(f"DNS {self.config.cutover_dns_record} reverted to on-prem bootstrap servers")
        logger.info("Rollback complete. Investigate errors before retrying.")

if __name__ == "__main__":
    import re  # Import here for topic pattern matching
    cutover_config = CutoverConfig(
        on_prem_bootstrap="on-prem-kafka-01:9093,on-prem-kafka-02:9093",
        target_bootstrap="pkc-xxxxx.us-east-1.aws.confluent.cloud:9092",
        consumer_group_id="prod-payment-consumers",
        max_allowed_lag=100
    )
    manager = TrafficCutoverManager(cutover_config)
    if not manager.run_cutover():
        sys.exit(1)

# Run Confluent Migration Assistant against on-prem cluster
docker run -it --rm \
  -v /etc/kafka/secrets:/secrets \
  confluentinc/kafka-migration-assistant:2026.1.0 \
  --source-bootstrap on-prem-kafka-01:9093 \
  --source-ssl-ca /secrets/ca-cert.pem \
  --source-ssl-cert /secrets/client-cert.pem \
  --source-ssl-key /secrets/client-key.pem \
  --target-type confluent-cloud \
  --target-bootstrap pkc-xxxxx.us-east-1.aws.confluent.cloud:9092 \
  --target-api-key $CONFLUENT_API_KEY \
  --target-api-secret $CONFLUENT_API_SECRET

# Update MirrorMaker 2.0 task count via Kafka Connect REST API
curl -X PUT \
  -H "Content-Type: application/json" \
  -d '{"tasks.max": "30"}' \
  http://kafka-connect-01:8083/connectors/on-prem-kafka-source/config

# Update Kafka DNS record TTL to 60 seconds via AWS CLI
aws route53 change-resource-record-sets \
  --hosted-zone-id Z1234567890ABC \
  --change-batch '{
    "Changes": [{
      "Action": "UPSERT",
      "ResourceRecordSet": {
        "Name": "kafka.prod.example.com",
        "Type": "CNAME",
        "TTL": 60,
        "ResourceRecords": [{"Value": "pkc-xxxxx.us-east-1.aws.confluent.cloud"}]
      }
    }]
  }'

kafka-migration-2026/
├── scripts/
│   ├── pre-migration-validator.py  # First code example
│   ├── mm2-config-generator.py     # Second code example
│   └── traffic-cutover-manager.py  # Third code example
├── configs/
│   ├── mm2-confluent-cloud-2026.json
│   └── mm2-msk-3.0.json
├── case-study/
│   └── fintech-migration-report.pdf
├── Dockerfile
└── README.md