kcp-kafka-copy-paste

Intro

Simplify and streamline your Kafka migration journey to Confluent Cloud!
kcp helps you migrate your Kafka setups to Confluent Cloud by providing tools to:
- Scan and identify resources in existing Kafka deployments.
- Create reports for migration planning and cost analysis.
- Generate migration assets and infrastructure configurations.
Kcp is our new, open-sourced migration toolkit that automates the most painful parts of moving from MSK to Confluent Cloud.
- Migrate in Days: kcp automates the manual, error-prone tasks of discovery and planning, provisioning, and data migration, turning the hands-on migration activity time from 4-6 weeks into a 2-3 days process.
- Automated Discovery & Costing: A single command scans the customer's entire MSK cluster, providing a detailed inventory and a real cost model based on actual usage to use in the TCO calculator.
- Terraform Automation: kcp generates pre-filled Terraform and Ansible scripts to automatically provision the equivalent Confluent Cloud resources.
- Data and Component Migration: kcp securely replicates data via external Cluster Linking and automates the conversion and migration of components like ACLs, connectors, and more.

Workflow

The typical migration flow:

Discover / scan — kcp discover (MSK) or kcp scan clusters (MSK or Apache Kafka) to build kcp-state.json.
Report — kcp report costs and kcp report metrics for cost and utilization analysis. Alternatively, use the kcp ui for fine-grained analysis.
Generate migration assets for data migration — kcp create-asset target-infra, migration-infra, migrate-topics, migrate-schemas, migrate-acls, migrate-connectors.
Initialize and execute client switchover — kcp migration init followed by kcp migration execute.

Quick Start

Step 1: Discover MSK Clusters

Find all MSK clusters in your AWS account:

# Install kcp
curl -fsSL https://raw.githubusercontent.com/confluentinc/kcp/main/install.sh | sh

# Discover clusters in a region
kcp discover --region us-east-2

# Or scan all regions
kcp discover --all-regions

Output: Lists all MSK cluster names and ARNs

Step 2: Create Credentials File

kcp needs to connect to your MSK cluster. Create msk-credentials.yaml:

regions:
- name: us-east-2
  clusters:
  - name: my-cluster
    arn: arn:aws:kafka:us-east-2:123456789:cluster/my-cluster/uuid
    auth_method:
      # For unauthenticated clusters (port 9092)
      unauthenticated_plaintext:
        use: true

      # OR for SASL/SCRAM (port 9096)
      # sasl_scram:
      #   use: true
      #   username: "admin"
      #   password: "password-from-secrets-manager"
      #   mechanism: SHA512

      # OR for IAM (port 9098) - NOT supported for zero-cut migrations
      # iam:
      #   use: true

Get SASL/SCRAM credentials from AWS Secrets Manager:

# Find secret name
aws secretsmanager list-secrets --region us-east-2 \
  --query 'SecretList[?contains(Name, `msk`)].Name' --output table

# Get credentials
aws secretsmanager get-secret-value \
  --secret-id "AmazonMSK_cluster-name_scram" \
  --region us-east-2 \
  --query SecretString --output text | jq

Step 3: Scan the Cluster

Deep analysis of your MSK cluster:

kcp scan clusters \
  --source-type msk \
  --state-file kcp-state.json \
  --credentials-file msk-credentials.yaml \
  --verbose

What it scans:

All topics, partitions, configs
Consumer groups and lag
ACLs and permissions
Throughput and storage metrics
Schema Registry (if configured)
Connectors (if using MSK Connect)

Output: kcp-state.json (contains complete cluster inventory)

Common issues:

# Timeout errors → MSK cluster has private endpoints
# Solution 1: Run kcp from EC2 in the same VPC
ssh ec2-user@bastion-host
# Install kcp and run scan from there

# Solution 2: Enable public access (if allowed)
# Check security group allows your IP on port 9096

Step 4: Analyze Costs & Metrics

# View in UI (interactive dashboard)
kcp ui
# Upload kcp-state.json
# Shows: Costs, Cluster details, Metrics, Topics, Connectors, ACLs, Clients

# Or generate CLI reports
kcp report costs --state-file kcp-state.json
kcp report metrics --state-file kcp-state.json

Step 5: Generate Migration Assets

# Generate Terraform for Confluent Cloud resources
kcp create-asset target-infra \
    --state-file kcp-state.json \
    --source-cluster-id "arn:aws:kafka:us-east-2:492737776546:cluster/kcp-msk-cluster/abc-6841-402f-b8d1-abc-3" \
    --needs-environment \
    --env-name "kcp-migration-demo" \
    --needs-cluster \
    --cluster-name "migrated-from-msk" \
    --cluster-type "dedicated" \
    --cluster-availability "SINGLE_ZONE" \
    --cluster-cku 1 \
    --output-dir "target_infra"

#  Key parameters:
#   - --state-file kcp-state.json - Uses scan data
#   - --source-cluster-id - Which MSK cluster to migrate from
#   - --needs-environment + --env-name - Create new CC environment
#   - --needs-cluster + --cluster-name - Create new CC cluster
#   - --cluster-type dedicated - Dedicated cluster (vs enterprise)
#   - --cluster-cku 1 - 1 Confluent Kafka Unit sizing
#   - --output-dir target_infra - Where to write Terraform files

# Generate Cluster Linking config (data replication)
kcp create-asset migration-infra --state-file kcp-state.json

# Generate topic migration scripts
kcp create-asset migrate-topics --state-file kcp-state.json
kcp create-asset migrate-topics --state-file kcp-state.json

# Migrate ACLs to CC RBAC
kcp create-asset migrate-acls --state-file kcp-state.json

# Migrate connectors
kcp create-asset migrate-connectors --state-file kcp-state.json

# Migrate schemas
kcp create-asset migrate-schemas --state-file kcp-state.json

What you get:

Terraform files to provision CC cluster
Ansible playbooks for Cluster Linking
Topic configs matching MSK settings
ACL → RBAC conversion (IAM ACLs require manual conversion)

Step 6: Execute Migration

# Initialize zero-cut migration
kcp migration init --state-file kcp-state.json

# Execute cutover (when ready)
kcp migration execute --state-file kcp-state.json

Common Workflows

Network Access Troubleshooting

Most MSK clusters are private (VPC-only). Check bootstrap server visibility:

aws kafka get-bootstrap-brokers \
  --cluster-arn "$CLUSTER_ARN" \
  --region us-east-2

Output shows:

BootstrapBrokerString (PLAINTEXT, port 9092)
BootstrapBrokerStringTls (TLS, port 9094)
BootstrapBrokerStringSaslScram (SASL/SCRAM, port 9096)
BootstrapBrokerStringSaslIam (IAM, port 9098)

If addresses show 172.x.x.x → Private endpoints:

# Option 1: Run from EC2 bastion in same VPC
kcp create-asset bastion-host --state-file kcp-state.json
# Generates EC2 Terraform

# Option 2: Check if cluster has public access enabled
aws kafka describe-cluster-v2 \
  --cluster-arn "$CLUSTER_ARN" \
  --region us-east-2 \
  --query 'ClusterInfo.Provisioned.BrokerNodeGroupInfo.ConnectivityInfo.PublicAccess.Type'

MSK Authentication Methods

See Amazon MSK Authentication for detailed comparison of:

Unauthenticated (PLAINTEXT)
TLS mutual authentication (mTLS)
SASL/SCRAM
IAM authentication

Critical for migration: IAM auth is NOT supported by Confluent Cloud Gateway, so zero-cut migrations require enabling SASL/SCRAM or unauthenticated access on MSK first.

CLI Help

A comprehensive CLI tool for planning and executing kafka cluster migrations to confluent cloud. Docs: https://confluentinc.github.io/kcp/0.8.7/

Usage:
  kcp [command]

Available Commands:
  completion   Generate the autocompletion script for the specified shell
  create-asset Generate infrastructure and migration assets
  discover     Multi-region, multi cluster discovery scan of AWS MSK
  docs         Show the documentation URL for this build
  help         Help about any command
  migration    Commands for migrating using CPC Gateway.
  report       Generate reports (costs, metrics, migration plan) from kcp scan data
  scan         Scan AWS resources for migration planning
  state        Operate on kcp-state.json files
  ui           Start the UI
  update       Update the kcp binary to the latest version
  version      Show version information

Flags:
  -h, --help      help for kcp
      --verbose   Enable verbose logging to console

Use "kcp [command] --help" for more information about a command.

Troubleshooting Common Issues

Issue 1: Authentication Failed (SASL/SCRAM)

Error: SASL Authentication failed: Authentication failed during authentication due to invalid credentials

Root causes:

Wrong username/password in credentials file
Secret not associated with the cluster
Wrong secret being used

Solution:

# Step 1: Find which secrets are actually associated with the cluster
aws kafka list-scram-secrets \
  --cluster-arn "arn:aws:kafka:region:account:cluster/name/uuid" \
  --region us-east-2

# Step 2: Get credentials from the CORRECT secret
aws secretsmanager get-secret-value \
  --secret-id "SECRET_NAME_FROM_STEP_1" \
  --region us-east-2 \
  --query SecretString --output text | jq

# Step 3: Update msk-credentials.yaml with correct username/password

Real example: Cluster had secret AmazonMSK_kcp-ec2-user-secrets-manager associated, but we were using AmazonMSK_kcp-msk-tao-cluster_scram which was NOT associated.

Issue 2: Network Timeout (Private Endpoints)

Error: dial tcp 172.32.44.231:9092: i/o timeout

Root cause: MSK cluster has private VPC endpoints, not accessible from local machine

Solution:

# Check if cluster has public access
aws kafka describe-cluster-v2 \
  --cluster-arn "$CLUSTER_ARN" \
  --region us-east-2 \
  --query 'ClusterInfo.Provisioned.BrokerNodeGroupInfo.ConnectivityInfo.PublicAccess.Type'

# If PUBLIC access enabled, check security groups
aws kafka describe-cluster \
  --cluster-arn "$CLUSTER_ARN" \
  --region us-east-2 \
  --query 'ClusterInfo.BrokerNodeGroupInfo.SecurityGroups' \
  --output text

# Add your IP to security group
MY_IP=$(curl -s ifconfig.me)
aws ec2 authorize-security-group-ingress \
  --group-id sg-xxxxx \
  --protocol tcp \
  --port 9196 \
  --cidr $MY_IP/32 \
  --region us-east-2

If private only: Run kcp from EC2 instance in same VPC

# Generate bastion host Terraform
kcp create-asset bastion-host --state-file kcp-state.json

# SSH to bastion and run kcp from there
ssh ec2-user@bastion-ip

Issue 3: Scanned 0 Clusters

Error: Scan completed successfully. Scanned 0 cluster(s)

Root causes:

Wrong credentials file format
Missing regions: top-level key
Network/auth issue (check verbose logs)

Solution:

Correct format:

regions:  # REQUIRED top-level key
- name: us-east-2
  clusters:
  - name: cluster-name
    arn: arn:aws:kafka:region:account:cluster/name/uuid
    auth_method:
      sasl_scram:
        use: true
        username: "user"
        password: "pass"
        mechanism: SHA512

Wrong format (missing regions:):

clusters:  # ❌ WRONG - will scan 0 clusters
  - name: cluster-name

Issue 4: Terraform Variable Missing

Error: No value for required variable subnet_cidr_ranges

Root cause: kcp generates Terraform expecting private link setup even when not needed

Solution:

Add to inputs.auto.tfvars:

subnet_cidr_ranges = []

Or regenerate without --needs-private-link flag.

Issue 5: AWS Session Expired

Error: Your session has expired. Please reauthenticate using 'aws login'

Solution:

# Re-authenticate
aws login

# Or export temporary credentials
export AWS_ACCESS_KEY_ID="..."
export AWS_SECRET_ACCESS_KEY="..."
export AWS_SESSION_TOKEN="..."

Migration Validation Checklist

Pre-Migration Baseline

Capture MSK state before migration:

# 1. Topic inventory
kafka-topics --bootstrap-server $MSK_BOOTSTRAP --list > msk-topics-baseline.txt
kafka-topics --bootstrap-server $MSK_BOOTSTRAP --describe > msk-topics-details.txt

# 2. Consumer groups
kafka-consumer-groups --bootstrap-server $MSK_BOOTSTRAP --list > msk-groups-baseline.txt

# 3. Consumer lag
for group in $(kafka-consumer-groups --bootstrap-server $MSK_BOOTSTRAP --list); do
  echo "=== $group ===" >> msk-lag-baseline.txt
  kafka-consumer-groups --bootstrap-server $MSK_BOOTSTRAP --group $group --describe >> msk-lag-baseline.txt
done

# 4. ACLs
kafka-acls --bootstrap-server $MSK_BOOTSTRAP --list > msk-acls-baseline.txt

# 5. kcp baseline report
kcp report metrics --state-file kcp-state.json > msk-metrics-baseline.txt

Post-Migration Validation

After migration, verify:

1. Topic Count & Names

# Create CC kafka.properties
cat > kafka.properties <<EOF
bootstrap.servers=pkc-xxxxx.us-east-2.aws.confluent.cloud:9092
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="API_KEY" password="API_SECRET";
EOF

# Compare topic counts
MSK_COUNT=$(kafka-topics --bootstrap-server $MSK_BOOTSTRAP --list | wc -l)
CC_COUNT=$(kafka-topics --command-config kafka.properties --bootstrap-server $CC_BOOTSTRAP --list | wc -l)

echo "MSK Topics: $MSK_COUNT"
echo "CC Topics: $CC_COUNT"

# Find missing topics
comm -23 \
  <(kafka-topics --bootstrap-server $MSK_BOOTSTRAP --list | sort) \
  <(kafka-topics --command-config kafka.properties --bootstrap-server $CC_BOOTSTRAP --list | sort) \
  > missing-topics.txt

if [ -s missing-topics.txt ]; then
  echo "⚠️ MISSING TOPICS:"
  cat missing-topics.txt
else
  echo "✅ All topics migrated"
fi

2. Partition Counts

for topic in $(kafka-topics --bootstrap-server $MSK_BOOTSTRAP --list); do
  MSK_PARTS=$(kafka-topics --bootstrap-server $MSK_BOOTSTRAP --describe --topic $topic | grep PartitionCount | awk '{print $2}')
  CC_PARTS=$(kafka-topics --command-config kafka.properties --bootstrap-server $CC_BOOTSTRAP --describe --topic $topic 2>/dev/null | grep PartitionCount | awk '{print $2}')

  if [ "$MSK_PARTS" != "$CC_PARTS" ]; then
    echo "⚠️ MISMATCH: $topic - MSK:$MSK_PARTS CC:$CC_PARTS"
  fi
done

3. Consumer Group Lag

# Check lag on CC cluster
for group in $(kafka-consumer-groups --command-config kafka.properties --bootstrap-server $CC_BOOTSTRAP --list | head -10); do
  echo "=== $group ==="
  kafka-consumer-groups --command-config kafka.properties --bootstrap-server $CC_BOOTSTRAP --group $group --describe
done

4. RBAC Validation

# Check service account has correct permissions
confluent iam rbac role-binding list \
  --principal User:sa-xxxxx \
  --cloud-cluster lkc-xxxxx \
  --current-environment

# Expected roles:
# - CloudClusterAdmin
# - DataSteward (environment level)
# - ResourceOwner (Schema Registry)

5. Connectivity Test

# Produce test message
echo "test-message" | kafka-console-producer \
  --bootstrap-server $CC_BOOTSTRAP \
  --producer.config kafka.properties \
  --topic test-migration

# Consume test message
kafka-console-consumer \
  --bootstrap-server $CC_BOOTSTRAP \
  --consumer.config kafka.properties \
  --topic test-migration \
  --from-beginning \
  --max-messages 1

Critical Validation Checklist

Before declaring migration complete:

Rollback Decision Criteria

Rollback to MSK if:

Data loss detected (message count mismatch)
Consumer lag >10,000 and growing
Application errors >5% of requests
Critical consumer group missing
Authentication failures after cutover

Hands-On Learnings

Real Migration Exercise (MSK → CC)

What we accomplished:

✅ Scanned live MSK cluster (kcp-msk-tao-cluster)
✅ Debugged authentication (found correct SCRAM secret)
✅ Fixed network access (security group whitelisting)
✅ Generated Terraform for CC cluster
✅ Deployed to Confluent Cloud deep-test environment
✅ Created 1 CKU dedicated cluster in us-east-2

Key blockers encountered:

Wrong SCRAM secret - Cluster used AmazonMSK_kcp-ec2-user-secrets-manager but we initially tried AmazonMSK_kcp-msk-tao-cluster_scram
- Fix: Always run aws kafka list-scram-secrets first
Security group blocking port 9196 - Public cluster but security group didn't allow our IP
- Fix: Add IP to security group rules
Credentials file format - Missing top-level regions: key
- Fix: Use exact format from original_creds.yaml example
Private VPC endpoints - Most production MSK clusters are private-only
- Fix: Run kcp from EC2 bastion or enable public access

Terraform Deployment Results

Deployed resources:

# Cluster details
cluster_id                 = "lkc-o33n2j9"
cluster_name              = "kcp-migrated-msk-demo"
cluster_bootstrap_endpoint = "pkc-8ypzxr.us-east-2.aws.confluent.cloud:9092"

# Service account
service_account_id = "sa-nw0qywk"

# API keys
kafka_api_key_id     = "D2JR6DK3C6BREOQE"
kafka_api_key_secret = "cfltVAYeQ+/jUhJK6B9g4lFRGylIgWLFBb08pntAA6pcOkMvLNepHZwGB3gyLOfQ"

Time to deploy: ~6 minutes (cluster provisioning)

Cost: ~ $1.50/hour = ~$ 1,095/month (1 CKU dedicated cluster)

Best Practices Learned

Always check which secret is associated with the cluster before scanning
Test network connectivity (telnet/nc) before running full scan
Use --verbose flag to debug authentication issues
Save kcp-state.json - it contains complete cluster inventory
Use existing CC environment instead of creating new one (--env-id vs --needs-environment)
Validate Terraform plan before applying (check CKU sizing, region)
Keep API keys secure - they're in Terraform output and state file
Document baseline metrics before migration (topic count, lag, throughput)

Zero-cut Migrations

With Zero-cut Migrations, clients make one change: update the bootstrap URL to point at the gateway. That's done in advance, no urgency. When the operator is ready to cut over, could be days later, could be weeks, they run one command. KCP fences traffic, waits for lag to hit zero, promotes the topics, flips routing to Confluent Cloud, and resumes traffic. Clients resume on CC. Operator is in full control the whole time: pick a single topic, a group of topics, or the whole cluster.

Core Architecture

Zero-cut migrations route all Kafka client traffic through a Kafka-aware proxy gateway deployed in (or adjacent to) the source cluster's network. The gateway forwards to source while Cluster Linking replicates, then atomically flips to Confluent Cloud once replication lag is zero.

Three Main Components:

KCP CLI
- Orchestrates cutover via three commands: kcp migration init, kcp migration lag-check, kcp migration execute
- Can be installed as a Confluent CLI plugin
- Apache 2.0 licensed, free to use
CC Gateway (Confluent Cloud Gateway)
- Kafka protocol proxy deployed in (or adjacent to) the source cluster's network
- Handles auth translation between source credentials and Confluent Cloud credentials
- Switches routing to Confluent Cloud at cutover, all without a restart
- Deployed on Kubernetes via Confluent for Kubernetes
- Requires Confluent Platform license (Confluent Cloud Gateway Add-On)
- Org-scoped: one license covers all Confluent Cloud clusters in a customer's org
- Trial mode supports up to 4 routes
Cluster Linking
- Replicates topics from the source cluster to the destination in real time, including consumer offset synchronization
- Available on Dedicated and Enterprise cluster types only
- Consumer offset sync (consumer.offset.sync.enable=true) must be enabled before migration

Client Experience During Cutover

During cutover, clients see BROKER_NOT_AVAILABLE with message "Migration to Confluent Cloud in progress. Your client will automatically retry"

Triggers automatic retries in all standard Kafka clients (Java, Python, Go, librdkafka) - no code changes required
Brief partial downtime window of approximately 60 seconds
Single bootstrap URL change for clients - that's the entire ask

Authentication Support

Supported source → destination combinations:

mTLS → CC SASL/PLAIN, CC mTLS, CC OAuth
SASL/SCRAM → CC SASL/PLAIN, CC OAuth
Unauthenticated → all CC auth types

Not supported:

IAM (requires pre-migration to SCRAM or mTLS)
SCRAM → CC mTLS

Two auth modes via --auth-mode:

dest_swap: clients present source credentials, gateway swaps for CC
source_swap: clients present CC credentials, gateway swaps for source

CRITICAL: IAM clients must migrate to SCRAM or mTLS before gateway onboarding

Migration Execution Process

Prerequisites:

Gateway deployed in Kubernetes
Cluster Linking configured and replicating
Client already pointing at gateway bootstrap URL
Three gateway CR files prepared

Execution steps:

# 1. Initialize migration (validates setup, creates migration plan)
kcp migration init

# 2. Monitor replication lag in real-time
kcp migration lag-check

# 3. Perform cutover in four phases: Pre-flight, Block, Promote, Switch+unblock
kcp migration execute

Cutover phases:

Pre-flight: Validates setup
Block: Fences traffic to source cluster
Promote: Promotes mirror topics via Cluster Linking
Switch+unblock: Flips routing to CC and resumes traffic

Key Constraints & Limitations

IAM not supported: IAM authentication requires migration to SCRAM or mTLS first
Consumer offset sync required: Must be enabled on cluster link before migration
One-way promotion: Topic promotion via Cluster Linking is one-way
- Rollback after promotion: possible (no data loss) but the cluster link is broken
Cluster type requirement: Cluster Linking only available on Dedicated and Enterprise clusters
Licensing: CC Gateway requires Confluent Cloud Gateway Add-On license (org-scoped)

Benefits

Fully orchestrated cutover: gateway fencing, mirror topic promotion, traffic routing flip, all automated
Real-time lag and offset monitoring so you pick the right window before and during the migration
Auth swap built in: unauthenticated clusters can migrate to Confluent cloud with minimal client changes
Works for any Kafka-compatible cluster migration
Operator in full control: pick a single topic, a group of topics, or the whole cluster

Intro​

Workflow​

Quick Start​

Step 1: Discover MSK Clusters​

Step 2: Create Credentials File​

Step 3: Scan the Cluster​

Step 4: Analyze Costs & Metrics​

Step 5: Generate Migration Assets​

Step 6: Execute Migration​

Common Workflows​

Network Access Troubleshooting​

MSK Authentication Methods​

CLI Help​

Troubleshooting Common Issues​

Issue 1: Authentication Failed (SASL/SCRAM)​

Issue 2: Network Timeout (Private Endpoints)​

Issue 3: Scanned 0 Clusters​

Issue 4: Terraform Variable Missing​

Issue 5: AWS Session Expired​

Migration Validation Checklist​

Pre-Migration Baseline​

Post-Migration Validation​

1. Topic Count & Names​

2. Partition Counts​

3. Consumer Group Lag​

4. RBAC Validation​

5. Connectivity Test​

Critical Validation Checklist​

Rollback Decision Criteria​

Hands-On Learnings​

Real Migration Exercise (MSK → CC)​

Terraform Deployment Results​

Best Practices Learned​

Zero-cut Migrations​

Core Architecture​

Client Experience During Cutover​

Authentication Support​

Migration Execution Process​

Key Constraints & Limitations​

Benefits​

Links​

Intro

Workflow

Quick Start

Step 1: Discover MSK Clusters

Step 2: Create Credentials File

Step 3: Scan the Cluster

Step 4: Analyze Costs & Metrics

Step 5: Generate Migration Assets

Step 6: Execute Migration

Common Workflows

Network Access Troubleshooting

MSK Authentication Methods

CLI Help

Troubleshooting Common Issues

Issue 1: Authentication Failed (SASL/SCRAM)

Issue 2: Network Timeout (Private Endpoints)

Issue 3: Scanned 0 Clusters

Issue 4: Terraform Variable Missing

Issue 5: AWS Session Expired

Migration Validation Checklist

Pre-Migration Baseline

Post-Migration Validation

1. Topic Count & Names

2. Partition Counts

3. Consumer Group Lag

4. RBAC Validation

5. Connectivity Test

Critical Validation Checklist

Rollback Decision Criteria

Hands-On Learnings

Real Migration Exercise (MSK → CC)

Terraform Deployment Results

Best Practices Learned

Zero-cut Migrations

Core Architecture

Client Experience During Cutover

Authentication Support

Migration Execution Process

Key Constraints & Limitations

Benefits

Links