翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
セルフマネージド Kafka クラスターの CreateReplicator API の例
レプリケーションを転送 (セルフマネージド Kafka を MSK Express に転送)
次の AWS CLI コマンドを使用して、セルフマネージド Kafka クラスターから Amazon MSK Express クラスターにデータをレプリケートするレプリケーターを作成します。
aws kafka create-replicator \ --replicator-name my-selfmanaged-to-msk-replicator \ --description "Replicating from self-managed Kafka to MSK Express" \ --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-express/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-express/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。これにより、コンシューマーグループのオフセットは、どちらの方向でもシームレスなカットオーバーをサポートする方法で同期されます。
ENHANCED オフセット同期モードでフォワードレプリケーター (セルフマネージド Kafka から MSK Express) を作成します。
aws kafka create-replicator \ --replicator-name my-selfmanaged-to-msk-replicator \ --description "Replicating from self-managed Kafka to MSK Express" \ --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-express/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-express/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" }}]'
次に、ENHANCEDオフセット同期モードを使用してリバースレプリケーター (MSK Express からセルフマネージド Kafka) も作成します。
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-express/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-express/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状態に進みます。レプリケーターが RUNNING ステータスになるまで約 30 分かかります。
