Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
CreateReplicator Exemples d'API pour les clusters Kafka autogérés
Réplication directe (Self-managed Kafka vers MSK Provisioned)
Utilisez la commande AWS CLI suivante pour créer un réplicateur qui réplique les données de votre cluster Kafka autogéré vers un cluster Amazon MSK provisioned.
Utilisation de SASL/SCRAM l'authentification :
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 }}]'
Utilisation de l'authentification 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 }}]'
Exemple de réplication bidirectionnelle
Pour configurer la réplication bidirectionnelle à des fins de restauration, les réplicateurs avant et arrière doivent être créés avec consumerGroupOffsetSyncMode la valeur définie sur. ENHANCED Cela garantit que les compensations des groupes de consommateurs sont synchronisées de manière à permettre une transition fluide dans les deux sens.
Utilisation de SASL/SCRAM l'authentification :
Créez le réplicateur direct (Kafka to MSK provisioned autogéré) avec le mode de synchronisation offset : 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" }}]'
Créez ensuite le réplicateur inverse (MSK provisionné pour Kafka autogéré) également avec le mode de synchronisation offset : 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" }}]'
Utilisation de l'authentification mTLS :
Créez le réplicateur direct (Kafka to MSK provisioned autogéré) avec le mode de synchronisation offset : 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" }}]'
Créez ensuite le réplicateur inverse (MSK provisionné pour Kafka autogéré) également avec le mode de synchronisation offset : 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" }}]'
Vérifier l'état du réplicateur
Vérifiez l'état de votre réplicateur à l'aide de la commande describe-replicator CLI :
aws kafka describe-replicator \ --replicator-arn arn:aws:kafka:us-east-1:123456789012:replicator/my-replicator/xxx
Le réplicateur passera par CREATING → RUNNING états. Attendez environ 30 minutes pour que le réplicateur atteigne RUNNING son statut.
