翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# セルフマネージド 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": ""
},
"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": "",
"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": ""
},
"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": "",
"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": ""
},
"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": "",
"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 分かかります。