{
  "name": "my-rds-source",
  "kafkaConnectVersion": "2.7.1",
  "capacity": {
    "provisionedThroughput": {
      "readCapacityUnits": 10,
      "writeCapacityUnits": 10
    }
  },
  "plugin": {
    "customPluginArn": "arn:aws:kafkaconnect:us-east-1:123456789012:custom-plugin/my-rds-connector"
  },
  "connectorConfiguration": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "my-rds-instance.cluster-xxxx.us-east-1.rds.amazonaws.com",
    "database.port": "5432",
    "database.user": "admin",
    "database.password": "secretsmanager:arn:aws:secretsmanager:us-east-1:123456789012:secret:db-cred:password::",
    "database.dbname": "mydb",
    "database.server.name": "myserver",
    "plugin.name": "pgoutput",
    "topic.prefix": "cdc"
  }
}

{
  "name": "my-s3-sink",
  "kafkaConnectVersion": "2.7.1",
  "capacity": {
    "provisionedThroughput": {
      "readCapacityUnits": 10,
      "writeCapacityUnits": 10
    }
  },
  "plugin": {
    "customPluginArn": "arn:aws:kafkaconnect:us-east-1:123456789012:custom-plugin/my-s3-sink"
  },
  "connectorConfiguration": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "s3.bucket.name": "my-kafka-data-bucket",
    "s3.region": "us-east-1",
    "topics": "cdc.server1.users",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd",
    "flush.size": "1000",
    "rotate.interval.ms": "60000"
  }
}

# Step 1: Package connector with dependencies
mvn clean package
mkdir -p /tmp/connector-plugin
cp target/my-custom-connector-1.0.jar /tmp/connector-plugin/
cp target/dependency/*.jar /tmp/connector-plugin/
cd /tmp/connector-plugin && zip -r my-custom-connector.zip .

# Step 2: Upload to S3
aws s3 cp my-custom-connector.zip s3://my-connectors-bucket/plugins/my-custom-connector.zip

# Step 3: Register as MSK Connect plugin
aws kafkaconnect create-custom-plugin \
  --name my-custom-connector \
  --content-type ZIP \
  --location-s3-location "{
    \"bucketArn\": \"arn:aws:s3:::my-connectors-bucket\",
    \"objectKey\": \"plugins/my-custom-connector.zip\"
  }"

# Step 4: Create connector
aws kafkaconnect create-connector \
  --connector-name my-connector \
  --connector-configuration file://connector-config.json \
  --kafkaconnect-version 2.7.1 \
  --connector-plugin "{
    \"customPluginArn\": \"arn:aws:kafkaconnect:us-east-1:123456789012:custom-plugin/my-custom-connector\"
  }" \
  --capacity "{
    \"provisionedThroughput\": {
      \"readCapacityUnits\": 10,
      \"writeCapacityUnits\": 10
    }
  }" \
  --worker-configuration "{
    \"workerConfigurationArn\": \"arn:aws:kafkaconnect:us-east-1:123456789012:worker-configuration/my-workers\"
  }" \
  --kafka-cluster "{
    \"apacheKafkaCluster\": {
      \"bootstrapServers\": \"b-1.mycluster.abc123.c2.kafka.us-east-1.amazonaws.com:9092\",
      \"vpc\": {
        \"securityGroups\": [\"sg-0123456789abcdef0\"],
        \"subnets\": [\"subnet-0123456789abcdef0\"]
      }
    }
  }"

{
  "transforms": "insertTimestamp,maskEmail",
  "transforms.insertTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.insertTimestamp.timestamp.field": "event_timestamp",
  "transforms.maskEmail.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.maskEmail.fields": "email",
  "transforms.maskEmail.replacement": "*****"
}

import boto3

def monitor_msk_connectors(cluster_arn: str):
    """Monitor MSK Connect connectors and report health status."""
    client = boto3.client('kafkaconnect')

    try:
        response = client.list_connectors(
            kafkaClusterArn=cluster_arn
        )

        for connector in response.get('connectors', []):
            name = connector['connectorName']
            state = connector['connectorState']
            print(f"Connector: {name}, State: {state}")

            if state == 'FAILED':
                print(f"  WARNING: {name} is in FAILED state")
                # Get connector status for more details
                status = client.describe_connector(
                    connectorArn=connector['connectorArn']
                )
                print(f"  Status: {status['connectorStatus']}")

    except Exception as e:
        print(f"Error monitoring connectors: {str(e)}")
        raise

import boto3

def update_connector_capacity(connector_arn: str, read_units: int, write_units: int):
    """Update MSK Connect connector capacity."""
    client = boto3.client('kafkaconnect')

    try:
        response = client.update_connector(
            connectorArn=connector_arn,
            capacityUpdate={
                'provisionedThroughput': {
                    'readCapacityUnits': read_units,
                    'writeCapacityUnits': write_units
                }
            }
        )
        print(f"Capacity updated: {response}")
    except Exception as e:
        print(f"Error updating capacity: {str(e)}")
        raise

{
  "transforms": "addTimestamp,maskPII,routeToTopic",
  "transforms.addTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addTimestamp.timestamp.field": "processed_at",
  "transforms.maskPII.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.maskPII.fields": "ssn,credit_card",
  "transforms.maskPII.replacement": "MASKED",
  "transforms.routeToTopic.type": "org.apache.kafka.connect.transforms.RegexRouter$Value",
  "transforms.routeToTopic.regex": ".*",
  "transforms.routeToTopic.replacement": "processed-${topic}"
}

{
  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "dlq-connect-errors",
  "errors.deadletterqueue.topic.replication.factor": "3",
  "errors.deadletterqueue.context.headers.enable": true,
  "errors.log.enable": true,
  "errors.log.include.messages": true,
  "errors.log.exclude.messages": false,
  "errors.retry.timeout": 30000,
  "errors.retry.delay.max.ms": 60000
}

{
  "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "key.converter.schemas.enable": true,
  "value.converter.schemas.enable": true,
  "tasks.max": 10,
  "heartbeat.interval.ms": 30000,
  "session.timeout.ms": 180000,
  "max.poll.interval.ms": 300000
}

#!/bin/bash
# MSK Connect health check script

CONNECTOR_NAME=$1
REGION=${2:-us-east-1}

echo "Checking health of connector: $CONNECTOR_NAME"

# Get connector status
STATUS=$(aws kafkaconnect describe-connector \
    --connector-arn "arn:aws:kafkaconnect:${REGION}:123456789012:connector/${CONNECTOR_NAME}" \
    --region ${REGION} \
    --query 'connectorStatus.connectorState' \
    --output text)

echo "Connector Status: $STATUS"

if [ "$STATUS" = "FAILED" ]; then
    echo "CRITICAL: Connector is in FAILED state"
    exit 1
elif [ "$STATUS" = "PAUSED" ]; then
    echo "WARNING: Connector is paused"
    exit 2
elif [ "$STATUS" = "RUNNING" ]; then
    echo "OK: Connector is running normally"
    exit 0
else
    echo "UNKNOWN: Connector status is $STATUS"
    exit 3
fi

# Create alarm for connector errors
aws cloudwatch put-metric-alarm \
    --alarm-name "MSKConnect-HighErrorRate" \
    --alarm-description "Alarm when MSK Connect error rate exceeds threshold" \
    --metric-name "ErrorCount" \
    --namespace "AWS/KafkaConnect" \
    --statistic Sum \
    --period 300 \
    --evaluation-periods 2 \
    --threshold 100 \
    --comparison-operator GreaterThanOrEqualToThreshold \
    --dimensions Name=ConnectorName,Value=my-connector \
    --alarm-actions arn:aws:sns:us-east-1:123456789012:ops-alerts \
    --region us-east-1

# Create alarm for connector status
aws cloudwatch put-metric-alarm \
    --alarm-name "MSKConnect-ConnectorFailed" \
    --alarm-description "Alarm when connector enters FAILED state" \
    --metric-name "ConnectorStatus" \
    --namespace "AWS/KafkaConnect" \
    --statistic Maximum \
    --period 300 \
    --evaluation-periods 1 \
    --threshold 1 \
    --comparison-operator GreaterThanOrEqualToThreshold \
    --dimensions Name=ConnectorName,Value=my-connector \
    --alarm-actions arn:aws:sns:us-east-1:123456789012:ops-alerts \
    --region us-east-1