Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.
CreateReplicator API-Beispiele für selbstverwaltete Kafka-Cluster
Replikation weiterleiten (von Self-managed Kafka nach MSK Provisioned)
Verwenden Sie den folgenden AWS CLI-Befehl, um einen Replikator zu erstellen, der Daten aus Ihrem selbstverwalteten Kafka-Cluster auf einen von Amazon MSK bereitgestellten Cluster repliziert.
Mithilfe SASL/SCRAM der Authentifizierung:
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 }}]'
Verwendung der mTLS-Authentifizierung:
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 }}]'
Verwenden der SASL/OAUTHBEARER (OAuth-) Authentifizierung:
Das folgende Beispiel verwendet den Mechanismus für Client-Anmeldeinformationen, der ein client_id und client_secret aus dem geheimen Schlüssel liest, auf den von verwiesen wird. tokenRequestSecretArn Der saslOAuthBearer Block ersetzt den saslScram oder MTLS -Block inclientAuthentication. Geben Sie genau eines von clientCredentialsiamJwtBearer, oder anclientCredentialsAssertion.
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 }}]'
Um stattdessen einen Mechanismus ohne Geheimhaltung zu verwenden, ersetzen Sie den clientCredentials Block durch einen der folgenden. Der IAM-JWT-Bearer-Mechanismus verwendet die Identität der Dienstausführungsrolle als Zusicherung. AWS Dazu sind ein audience (der Wert des aud Anspruchs, den Ihr IDP erwartet) und ein (oder) erforderlich: signingAlgorithm RS256 ES384
"clientAuthentication": { "saslOAuthBearer": { "tokenEndpointUrl": "https://idp.example.com/oauth/token", "iamJwtBearer": { "audience": "msk-replicator-client", "signingAlgorithm": "RS256" }, "tokenEndpointAuthenticationMethod": "NONE", "scope": "kafka" } }
Der Mechanismus zur Bestätigung der Client-Anmeldeinformationen verwendet das signierte JWT der Dienstausführungsrolle als. client_assertion Es erfordert außerdem audience und und signingAlgorithm muss tokenEndpointAuthenticationMethod auf Folgendes gesetzt sein: NONE
"clientAuthentication": { "saslOAuthBearer": { "tokenEndpointUrl": "https://idp.example.com/oauth/token", "clientCredentialsAssertion": { "audience": "msk-replicator-client", "signingAlgorithm": "RS256" }, "tokenEndpointAuthenticationMethod": "NONE", "scope": "kafka" } }
Das Beispiel beinhaltettokenEndpointTlsCertificateArn, was optional ist und nur benötigt wird, wenn Ihr IDP ein von einer privaten Zertifizierungsstelle ausgestelltes Zertifikat vorlegt; lassen Sie es weg, wenn Ihr IDP ein öffentlich vertrauenswürdiges Zertifikat verwendet. Wenn Ihr Kafka-Anbieter während des Handshakes SASL-Erweiterungen benötigt, fügen Sie sie als extension.<name> Einträge in Ihrem Secret und nicht in der API-Anfrage hinzu. Siehe Speichern Sie Anmeldeinformationen in AWS Secrets Manager.
Beispiel für bidirektionale Replikation
Um die bidirektionale Replikation für die Rollback-Funktion einzurichten, müssen sowohl der Vorwärts- als auch der Rückwärts-Replikator mit set to erstellt werden. consumerGroupOffsetSyncMode ENHANCED Dadurch wird sichergestellt, dass Offsets für Verbrauchergruppen so synchronisiert werden, dass eine nahtlose Umstellung in beide Richtungen möglich ist.
Wichtig
Registrieren Sie jeden Cluster auf die gleiche Weise in beiden Replikatoren. Wenn ein Cluster in einem Replikator als Amazon MSK-Cluster registriert ist (amazonMskClusterunter Verwendung seines Cluster-ARN), registrieren Sie ihn auch im anderen Replikator als Amazon MSK-Cluster. Wenn er als selbstverwalteter Apache Kafka-Cluster registriert ist (unter Verwendung einer Bootstrap-Brokerzeichenfolge)apacheKafkaCluster, verwenden Sie in beiden Fällen denselben Typ. Dadurch wird eine konsistente Nachverfolgung in MSK Replicator gewährleistet, um Replikationsschleifen zu vermeiden.
Mithilfe der Authentifizierung SASL/SCRAM :
Erstellen Sie den Forward-Replikator (von Kafka nach MSK Provisioned selbst verwaltet) mit Offset-Sync-Modus: 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" }}]'
Erstellen Sie dann den Reverse-Replicator (MSK Provisioned bis Self-Managed Kafka) ebenfalls mit Offset-Sync-Modus: 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" }}]'
Mithilfe der mTLS-Authentifizierung:
Erstellen Sie den Forward-Replikator (von Kafka nach MSK Provisioned selbst verwaltet) mit Offset-Sync-Modus: 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" }}]'
Erstellen Sie dann den Reverse-Replicator (MSK Provisioned bis Self-Managed Kafka) ebenfalls mit Offset-Sync-Modus: 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" }}]'
Überprüfen Sie den Status des Replikators
Überprüfen Sie den Status Ihres Replikators mit dem describe-replicator CLI-Befehl:
aws kafka describe-replicator \ --replicator-arn arn:aws:kafka:us-east-1:123456789012:replicator/my-replicator/xxx
Der Replikator durchläuft CREATING → RUNNING Status. Warten Sie etwa 30 Minuten, bis der Replikator den Status erreicht RUNNING hat.
