本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
CreateReplicator 自建的 Kafka 集群的 API 示例
向前复制（已配置 Self-managed Kafka 到 MSK）
使用以下 Amazon CLI 命令创建复制器，将数据从您的自我管理的 Kafka 集群复制到 Amazon MSK 预置集群。
使用 SASL/SCRAM 身份验证：
aws kafka create-replicator \ --replicator-name my-selfmanaged-to-msk-replicator \ --description "Replicating from self-managed Kafka to MSK Provisioned" \ --service-execution-role-arn arn:aws:iam::123456789012:role/MSKReplicatorRole \ --kafka-clusters '[ { "apacheKafkaCluster": { "bootstrapBrokerString": "broker1:9094,broker2:9094,broker3:9094", "apacheKafkaClusterId": "<self-managed-cluster-id>" }, "clientAuthentication": { "saslScram": { "mechanism": "SHA256", "secretArn": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-scram-creds" } }, "encryptionInTransit": { "encryptionType": "TLS", "rootCaCertificate": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-scram-creds" }, "vpcConfig": { "subnetIds": ["subnet-aaa","subnet-bbb","subnet-ccc"], "securityGroupIds": ["sg-xxxxxxxxxxxxxxxxx"] } }, { "amazonMskCluster": { "mskClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx" }, "vpcConfig": { "subnetIds": ["subnet-ddd","subnet-eee","subnet-fff"], "securityGroupIds": ["sg-yyyyyyyyy"] } }]' \ --replication-info-list '[{ "sourceKafkaClusterId": "<self-managed-cluster-id>", "targetKafkaClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx", "targetCompressionType": "NONE", "topicReplication": { "topicsToReplicate": [".*"], "topicNameConfiguration": {"type": "IDENTICAL"}, "startingPosition": {"type": "EARLIEST"}, "detectAndCopyNewTopics": true, "copyTopicConfigurations": true, "copyAccessControlListsForTopics": true }, "consumerGroupReplication": { "consumerGroupsToReplicate": [".*"], "detectAndCopyNewConsumerGroups": true, "synchroniseConsumerGroupOffsets": true }}]'
使用 mTLS 身份验证：
aws kafka create-replicator \ --replicator-name my-selfmanaged-to-msk-replicator \ --description "Replicating from self-managed Kafka to MSK Provisioned" \ --service-execution-role-arn arn:aws:iam::123456789012:role/MSKReplicatorRole \ --kafka-clusters '[ { "apacheKafkaCluster": { "bootstrapBrokerString": "broker1:9095,broker2:9095,broker3:9095", "apacheKafkaClusterId": "<self-managed-cluster-id>" }, "clientAuthentication": { "MTLS": { "secretArn": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-mtls-creds" } }, "encryptionInTransit": { "encryptionType": "TLS", "rootCaCertificate": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-ca-cert" }, "vpcConfig": { "subnetIds": ["subnet-aaa","subnet-bbb","subnet-ccc"], "securityGroupIds": ["sg-xxxxxxxxxxxxxxxxx"] } }, { "amazonMskCluster": { "mskClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx" }, "vpcConfig": { "subnetIds": ["subnet-ddd","subnet-eee","subnet-fff"], "securityGroupIds": ["sg-yyyyyyyyy"] } }]' \ --replication-info-list '[{ "sourceKafkaClusterId": "<self-managed-cluster-id>", "targetKafkaClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx", "targetCompressionType": "NONE", "topicReplication": { "topicsToReplicate": [".*"], "topicNameConfiguration": {"type": "IDENTICAL"}, "startingPosition": {"type": "EARLIEST"}, "detectAndCopyNewTopics": true, "copyTopicConfigurations": true, "copyAccessControlListsForTopics": true }, "consumerGroupReplication": { "consumerGroupsToReplicate": [".*"], "detectAndCopyNewConsumerGroups": true, "synchroniseConsumerGroupOffsets": true }}]'
使用 SASL/OAUTHBEARER (OAuth) 身份验证：
以下示例使用客户端凭证机制，该机制client_secret从引用的密钥中读取client_id和tokenRequestSecretArn。该saslOAuthBearer方块取代了中的saslScram或MTLS方块clientAuthentication。请仅指定clientCredentialsiamJwtBearer、或中的一个clientCredentialsAssertion。
aws kafka create-replicator \ --replicator-name my-selfmanaged-to-msk-replicator \ --description "Replicating from self-managed Kafka to MSK Provisioned" \ --service-execution-role-arn arn:aws:iam::123456789012:role/MSKReplicatorRole \ --kafka-clusters '[ { "apacheKafkaCluster": { "bootstrapBrokerString": "broker1:9096,broker2:9096,broker3:9096", "apacheKafkaClusterId": "<self-managed-cluster-id>" }, "clientAuthentication": { "saslOAuthBearer": { "tokenEndpointUrl": "https://idp.example.com/oauth/token", "clientCredentials": { "tokenRequestSecretArn": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-oauth-creds" }, "tokenEndpointAuthenticationMethod": "POST", "scope": "kafka", "tokenEndpointTlsCertificateArn": "arn:aws:secretsmanager:us-east-1:123456789012:secret:idp-ca-cert" } }, "encryptionInTransit": { "encryptionType": "TLS", "rootCaCertificate": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-ca-cert" }, "vpcConfig": { "subnetIds": ["subnet-aaa","subnet-bbb","subnet-ccc"], "securityGroupIds": ["sg-xxxxxxxxxxxxxxxxx"] } }, { "amazonMskCluster": { "mskClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx" }, "vpcConfig": { "subnetIds": ["subnet-ddd","subnet-eee","subnet-fff"], "securityGroupIds": ["sg-yyyyyyyyy"] } }]' \ --replication-info-list '[{ "sourceKafkaClusterId": "<self-managed-cluster-id>", "targetKafkaClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx", "targetCompressionType": "NONE", "topicReplication": { "topicsToReplicate": [".*"], "topicNameConfiguration": {"type": "IDENTICAL"}, "startingPosition": {"type": "EARLIEST"}, "detectAndCopyNewTopics": true, "copyTopicConfigurations": true, "copyAccessControlListsForTopics": true }, "consumerGroupReplication": { "consumerGroupsToReplicate": [".*"], "detectAndCopyNewConsumerGroups": true, "synchroniseConsumerGroupOffsets": true }}]'
要改用无密机制，请将该clientCredentials方块替换为以下其中一种。IAM JWT 承载机制使用服务执行角色的 Amazon 身份作为断言。它需要audience（您的 IDP 期望的aud索赔价值）和signingAlgorithm（RS256或ES384）：
"clientAuthentication": { "saslOAuthBearer": { "tokenEndpointUrl": "https://idp.example.com/oauth/token", "iamJwtBearer": { "audience": "msk-replicator-client", "signingAlgorithm": "RS256" }, "tokenEndpointAuthenticationMethod": "NONE", "scope": "kafka" } }
客户端凭证断言机制使用服务执行角色的签名 JWT 作为. client_assertion 它还要求audience和signingAlgorithm，并且tokenEndpointAuthenticationMethod需要设置为NONE：
"clientAuthentication": { "saslOAuthBearer": { "tokenEndpointUrl": "https://idp.example.com/oauth/token", "clientCredentialsAssertion": { "audience": "msk-replicator-client", "signingAlgorithm": "RS256" }, "tokenEndpointAuthenticationMethod": "NONE", "scope": "kafka" } }
该示例包括tokenEndpointTlsCertificateArn，它是可选的，仅当您的 IDP 出示私有 CA 颁发的证书时才需要；如果您的 IDP 使用公共信任证书，则省略该证书。如果您的 Kafka 提供商在握手期间需要 SASL 扩展，请将它们作为extension.<name>条目添加到您的密钥中，而不是 API 请求中。请参阅将凭据存储在 Amazon Secrets Manager。
双向复制示例
要设置双向复制以实现回滚功能，必须将正向和反向复制器都设置为consumerGroupOffsetSyncMode。ENHANCED这样可以确保消费者组偏移量以支持双向无缝切换的方式进行同步。
重要
在两个复制器中以相同的方式注册每个集群。如果一个集群在一个复制器中注册为亚马逊 MSK 集群（amazonMskCluster使用其集群 ARN），则也要在另一个复制器中将其注册为亚马逊 MSK 集群。如果它注册为自管理的 Apache Kafka 集群（apacheKafkaCluster使用引导代理字符串），则在两者中使用相同的类型。这样做可确保 MSK Replicator 中的一致跟踪，以防止复制循环。
使用 SASL/SCRAM 身份验证：
使用偏移同步模式创建正向复制器（自管理 Kafka 到 MSK 已配置）：ENHANCED
aws kafka create-replicator \ --replicator-name my-selfmanaged-to-msk-replicator \ --description "Replicating from self-managed Kafka to MSK Provisioned" \ --service-execution-role-arn arn:aws:iam::123456789012:role/MSKReplicatorRole \ --kafka-clusters '[ { "apacheKafkaCluster": { "bootstrapBrokerString": "broker1:9094,broker2:9094,broker3:9094", "apacheKafkaClusterId": "<self-managed-cluster-id>" }, "clientAuthentication": { "saslScram": { "mechanism": "SHA256", "secretArn": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-scram-creds" } }, "encryptionInTransit": { "encryptionType": "TLS", "rootCaCertificate": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-ca-cert" }, "vpcConfig": { "subnetIds": ["subnet-aaa","subnet-bbb","subnet-ccc"], "securityGroupIds": ["sg-xxxxxxxxxxxxxxxxx"] } }, { "amazonMskCluster": { "mskClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx" }, "vpcConfig": { "subnetIds": ["subnet-ddd","subnet-eee","subnet-fff"], "securityGroupIds": ["sg-yyyyyyyyy"] } }]' \ --replication-info-list '[{ "sourceKafkaClusterId": "<self-managed-cluster-id>", "targetKafkaClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx", "targetCompressionType": "NONE", "topicReplication": { "topicsToReplicate": [".*"], "topicNameConfiguration": {"type": "IDENTICAL"}, "startingPosition": {"type": "EARLIEST"}, "detectAndCopyNewTopics": true, "copyTopicConfigurations": true, "copyAccessControlListsForTopics": true }, "consumerGroupReplication": { "consumerGroupsToReplicate": [".*"], "detectAndCopyNewConsumerGroups": true, "synchroniseConsumerGroupOffsets": true, "consumerGroupOffsetSyncMode": "ENHANCED" }}]'
然后使用偏移同步模式创建反向复制器（MSK 配置为自建的 Kafka）：ENHANCED
aws kafka create-replicator \ --replicator-name my-msk-to-selfmanaged-replicator \ --description "Reverse replication for rollback" \ --service-execution-role-arn arn:aws:iam::123456789012:role/MSKReplicatorRole \ --kafka-clusters '[ { "amazonMskCluster": { "mskClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx" }, "vpcConfig": { "subnetIds": ["subnet-ddd","subnet-eee","subnet-fff"], "securityGroupIds": ["sg-yyyyyyyyy"] } }, { "apacheKafkaCluster": { "bootstrapBrokerString": "broker1:9094,broker2:9094,broker3:9094", "apacheKafkaClusterId": "<self-managed-cluster-id>" }, "clientAuthentication": { "saslScram": { "mechanism": "SHA256", "secretArn": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-scram-creds" } }, "encryptionInTransit": { "encryptionType": "TLS", "rootCaCertificate": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-ca-cert" }, "vpcConfig": { "subnetIds": ["subnet-aaa","subnet-bbb","subnet-ccc"], "securityGroupIds": ["sg-xxxxxxxxxxxxxxxxx"] } }]' \ --replication-info-list '[{ "sourceKafkaClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx", "targetKafkaClusterId": "<self-managed-cluster-id>", "targetCompressionType": "NONE", "topicReplication": { "topicsToReplicate": [".*"], "topicNameConfiguration": {"type": "IDENTICAL"}, "startingPosition": {"type": "LATEST"}, "detectAndCopyNewTopics": true, "copyTopicConfigurations": true, "copyAccessControlListsForTopics": true }, "consumerGroupReplication": { "consumerGroupsToReplicate": [".*"], "detectAndCopyNewConsumerGroups": true, "synchroniseConsumerGroupOffsets": true, "consumerGroupOffsetSyncMode": "ENHANCED" }}]'
使用 mTLS 身份验证：
使用偏移同步模式创建正向复制器（自管理 Kafka 到 MSK 已配置）：ENHANCED
aws kafka create-replicator \ --replicator-name my-selfmanaged-to-msk-replicator \ --description "Replicating from self-managed Kafka to MSK Provisioned" \ --service-execution-role-arn arn:aws:iam::123456789012:role/MSKReplicatorRole \ --kafka-clusters '[ { "apacheKafkaCluster": { "bootstrapBrokerString": "broker1:9095,broker2:9095,broker3:9095", "apacheKafkaClusterId": "<self-managed-cluster-id>" }, "clientAuthentication": { "MTLS": { "secretArn": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-mtls-creds" } }, "encryptionInTransit": { "encryptionType": "TLS", "rootCaCertificate": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-ca-cert" }, "vpcConfig": { "subnetIds": ["subnet-aaa","subnet-bbb","subnet-ccc"], "securityGroupIds": ["sg-xxxxxxxxxxxxxxxxx"] } }, { "amazonMskCluster": { "mskClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx" }, "vpcConfig": { "subnetIds": ["subnet-ddd","subnet-eee","subnet-fff"], "securityGroupIds": ["sg-yyyyyyyyy"] } }]' \ --replication-info-list '[{ "sourceKafkaClusterId": "<self-managed-cluster-id>", "targetKafkaClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx", "targetCompressionType": "NONE", "topicReplication": { "topicsToReplicate": [".*"], "topicNameConfiguration": {"type": "IDENTICAL"}, "startingPosition": {"type": "EARLIEST"}, "detectAndCopyNewTopics": true, "copyTopicConfigurations": true, "copyAccessControlListsForTopics": true }, "consumerGroupReplication": { "consumerGroupsToReplicate": [".*"], "detectAndCopyNewConsumerGroups": true, "synchroniseConsumerGroupOffsets": true, "consumerGroupOffsetSyncMode": "ENHANCED" }}]'
然后使用偏移同步模式创建反向复制器（MSK 配置为自建的 Kafka）：ENHANCED
aws kafka create-replicator \ --replicator-name my-msk-to-selfmanaged-replicator \ --description "Reverse replication for rollback" \ --service-execution-role-arn arn:aws:iam::123456789012:role/MSKReplicatorRole \ --kafka-clusters '[ { "amazonMskCluster": { "mskClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx" }, "vpcConfig": { "subnetIds": ["subnet-ddd","subnet-eee","subnet-fff"], "securityGroupIds": ["sg-yyyyyyyyy"] } }, { "apacheKafkaCluster": { "bootstrapBrokerString": "broker1:9095,broker2:9095,broker3:9095", "apacheKafkaClusterId": "<self-managed-cluster-id>" }, "clientAuthentication": { "MTLS": { "secretArn": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-mtls-creds" } }, "encryptionInTransit": { "encryptionType": "TLS", "rootCaCertificate": "arn:aws:secretsmanager:us-east-1:123456789012:secret:kafka-ca-cert" }, "vpcConfig": { "subnetIds": ["subnet-aaa","subnet-bbb","subnet-ccc"], "securityGroupIds": ["sg-xxxxxxxxxxxxxxxxx"] } }]' \ --replication-info-list '[{ "sourceKafkaClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-provisioned/xxx", "targetKafkaClusterId": "<self-managed-cluster-id>", "targetCompressionType": "NONE", "topicReplication": { "topicsToReplicate": [".*"], "topicNameConfiguration": {"type": "IDENTICAL"}, "startingPosition": {"type": "LATEST"}, "detectAndCopyNewTopics": true, "copyTopicConfigurations": true, "copyAccessControlListsForTopics": true }, "consumerGroupReplication": { "consumerGroupsToReplicate": [".*"], "detectAndCopyNewConsumerGroups": true, "synchroniseConsumerGroupOffsets": true, "consumerGroupOffsetSyncMode": "ENHANCED" }}]'
验证复制器状态
使用 describe-replicator CLI 命令检查复制器的状态：
aws kafka describe-replicator \ --replicator-arn arn:aws:kafka:us-east-1:123456789012:replicator/my-replicator/xxx
复制器将通过 CREATING → RUNNING 状态。等待大约 30 分钟让复制器进入RUNNING状态。