기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.
# 자체 관리형 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` 오프셋 동기화 모드를 사용하여 Replicator(자체 관리형 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` 오프셋 동기화 모드를 사용하여 역방향 Replicator(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"
}}]'
```
## Replicator 상태 확인
`describe-replicator` CLI 명령을 사용하여 Replicator의 상태를 확인합니다.
```
aws kafka describe-replicator \
--replicator-arn arn:aws:kafka:us-east-1:123456789012:replicator/my-replicator/xxx
```
복제기는 `CREATING` → `RUNNING` 상태를 통해 진행됩니다. 복제기가 `RUNNING` 상태에 도달할 때까지 약 30분 정도 기다립니다.