# Examples to set up Amazon MSK Connect resources Examples This section includes examples to help you set up Amazon MSK Connect resources such as common third-party connectors and configuration providers. **Topics** + [ # Set up Amazon S3 sink connector ](mkc-S3sink-connector-example.md) + [ # Set up EventBridge Kafka sink connector for MSK Connect ](mkc-eventbridge-kafka-connector.md) + [ # Use Debezium source connector with configuration provider ](mkc-debeziumsource-connector-example.md) # Set up Amazon S3 sink connector This example shows how to use the Confluent [Amazon S3 sink connector](https://www.confluent.io/hub/confluentinc/kafka-connect-s3) and the AWS CLI to create an Amazon S3 sink connector in MSK Connect. 1. Copy the following JSON and paste it in a new file. Replace the placeholder strings with values that correspond to your Amazon MSK cluster's bootstrap servers connection string and the cluster's subnet and security group IDs. For information about how to set up a service execution role, see [IAM roles and policies for MSK Connect](msk-connect-iam.md). ``` { "connectorConfiguration": { "connector.class": "io.confluent.connect.s3.S3SinkConnector", "s3.region": "us-east-1", "format.class": "io.confluent.connect.s3.format.json.JsonFormat", "flush.size": "1", "schema.compatibility": "NONE", "topics": "my-test-topic", "tasks.max": "2", "partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner", "storage.class": "io.confluent.connect.s3.storage.S3Storage", "s3.bucket.name": "amzn-s3-demo-bucket" }, "connectorName": "example-S3-sink-connector", "kafkaCluster": { "apacheKafkaCluster": { "bootstrapServers": "", "vpc": { "subnets": [ "", "", "" ], "securityGroups": [""] } } }, "capacity": { "provisionedCapacity": { "mcuCount": 2, "workerCount": 4 } }, "kafkaConnectVersion": "2.7.1", "serviceExecutionRoleArn": "", "plugins": [ { "customPlugin": { "customPluginArn": "", "revision": 1 } } ], "kafkaClusterEncryptionInTransit": {"encryptionType": "PLAINTEXT"}, "kafkaClusterClientAuthentication": {"authenticationType": "NONE"} } ``` 1. Run the following AWS CLI command in the folder where you saved the JSON file in the previous step. ``` aws kafkaconnect create-connector --cli-input-json file://connector-info.json ``` The following is an example of the output that you get when you run the command successfully. ``` { "ConnectorArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector/example-S3-sink-connector/abc12345-abcd-4444-a8b9-123456f513ed-2", "ConnectorState": "CREATING", "ConnectorName": "example-S3-sink-connector" } ``` # Set up EventBridge Kafka sink connector for MSK Connect Set up EventBridge Kafka sink connector This topic shows you how to set up the [EventBridge Kafka sink connector](https://github.com/awslabs/eventbridge-kafka-connector) for MSK Connect. This connector lets you send events from your MSK cluster to EventBridge [event buses](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-bus.html). This topic describes the process for creating the required resources and configuring the connector to enable seamless data flow between Kafka and EventBridge. **Topics** + [ ## Prerequisites ](#mkc-eb-kafka-prerequisites) + [ ## Set up the resources required for MSK Connect ](#mkc-eb-kafka-set-up-resources) + [ ## Create the connector ](#mkc-eb-kafka-create-connector) + [ ## Send messages to Kafka ](#mkc-eb-kafka-send-json-encoded-messages) ## Prerequisites Before deploying the connector, make sure that you have the following resources: + **Amazon MSK cluster**: An active MSK cluster to produce and consume Kafka messages. + **Amazon EventBridge event bus**: An EventBridge event bus to receive events from the Kafka topics. + **IAM roles**: Create IAM roles with the necessary permissions for MSK Connect and the EventBridge connector. + [Access to the public internet](msk-connect-internet-access.md) from MSK Connect or a [VPC interface endpoint](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-related-service-vpc.html) for EventBridge created in the VPC and subnet of your MSK cluster. This helps you avoid traversing the public internet and without requiring NAT gateways. + A [client machine](create-serverless-cluster-client.md), such as an Amazon EC2 instance or [AWS CloudShell](https://aws.amazon.com/cloudshell/), to create topics and send records to Kafka. ## Set up the resources required for MSK Connect You create an IAM role for the connector, and then you create the connector. You also create an EventBridge rule to filter Kafka events sent to the EventBridge event bus. **Topics** + [ ### IAM role for the connector ](#mkc-eb-kafka-iam-role-connector) + [ ### An EventBridge rule for incoming events ](#mkc-eb-kafka-create-rule) ### IAM role for the connector The IAM role that you associate with the connector must have the [PutEvents](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-permissions-reference.html) permission to allow sending events to EventBridge. The following IAM policy example grants you the permissin to send events to an event bus named `example-event-bus`. Make sure that you replace the resource ARN in the following example with the ARN of your event bus. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "events:PutEvents" ], "Resource": "arn:aws:events:us-east-1:123456789012:event-bus/example-event-bus" } ] } ``` ------ In addition, you must make sure that your IAM role for the connector contains the following trust policy. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "kafkaconnect.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` ------ ### An EventBridge rule for incoming events You create [rules](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-rules.html) that match incoming events with event data criteria, known as [https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html). With an event pattern, you can define the criteria to filter incoming events, and determine which events should trigger a particular rule and subsequently be routed to a designated [target](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-targets.html). The following example of an event pattern matches Kafka events sent to the EventBridge event bus. ``` { "detail": { "topic": ["msk-eventbridge-tutorial"] } } ``` The following is an example of an event sent from Kafka to EventBridge using the Kafka sink connector. ``` { "version": "0", "id": "dbc1c73a-c51d-0c0e-ca61-ab9278974c57", "account": "123456789012", "time": "2025-03-26T10:15:00Z", "region": "us-east-1", "detail-type": "msk-eventbridge-tutorial", "source": "kafka-connect.msk-eventbridge-tutorial", "resources": [], "detail": { "topic": "msk-eventbridge-tutorial", "partition": 0, "offset": 0, "timestamp": 1742984100000, "timestampType": "CreateTime", "headers": [], "key": "order-1", "value": { "orderItems": [ "item-1", "item-2" ], "orderCreatedTime": "Wed Mar 26 10:15:00 UTC 2025" } } } ``` In the EventBridge console, [create a rule](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-create-rule.html) on the event bus using this example pattern and specify a target, such as a CloudWatch Logs group. The EventBridge console will automatically configure the necessary access policy for the CloudWatch Logs group. ## Create the connector In the following section, you create and deploy the [EventBridge Kafka sink connector](https://github.com/awslabs/eventbridge-kafka-connector) using the AWS Management Console. **Topics** + [ ### Step 1: Download the connector ](#mkc-eb-kafka-download-connector) + [ ### Step 2: Create an Amazon S3 bucket ](#mkc-eb-kafka-s3-bucket-create) + [ ### Step 3: Create a plugin in MSK Connect ](#mkc-eb-kafka-create-plugin) + [ ### Step 4: Create the connector ](#mkc-eb-kafka-create-connector) ### Step 1: Download the connector Download the latest EventBridge connector sink JAR from the [GitHub releases page](https://github.com/awslabs/eventbridge-kafka-connector/releases) for the EventBridge Kafka connector. For example, to download the version v1.4.1, choose the JAR file link, `kafka-eventbridge-sink-with-dependencies.jar`, to download the connector. Then, save the file to a preferred location on your machine. ### Step 2: Create an Amazon S3 bucket 1. To store the JAR file in Amazon S3 for use with MSK Connect, open the AWS Management Console, and then choose Amazon S3. 1. In the Amazon S3 console, choose **Create bucket**, and enter a unique bucket name. For example, **amzn-s3-demo-bucket1-eb-connector**. 1. Choose an appropriate Region for your Amazon S3 bucket. Make sure that it matches the Region where your MSK cluster is deployed. 1. For **Bucket settings**, keep the default selections or adjust as needed. 1. Choose **Create bucket** 1. Upload the JAR file to the Amazon S3 bucket. ### Step 3: Create a plugin in MSK Connect 1. Open the AWS Management Console, and then navigate to **MSK Connect**. 1. In the left navigation pane, choose **Custom plugins**. 1. Choose **Create plugin**, and then enter a **Plugin name**. For example, **eventbridge-sink-plugin**. 1. For **Custom plugin location**, paste the **S3 object URL**. 1. Add an optional description for the plugin. 1. Choose **Create plugin**. After the plugin is created, you can use it to configure and deploy the EventBridge Kafka connector in MSK Connect. ### Step 4: Create the connector Before creating the connector, we recommend to create the required Kafka topic to avoid connector errors. To create the topic, use your client machine. 1. In the left pane of the MSK console, choose **Connectors**, and then choose **Create connector**. 1. In the list of plugins, choose **eventbridge-sink-plugin**, then choose **Next**. 1. For the connector name, enter **EventBridgeSink**. 1. In the list of clusters, choose your MSK cluster. 1. Copy the following configuration for the connector and paste it into the **Connector configuration** field Replace the placeholders in the following configuration, as required. + Remove `aws.eventbridge.endpoint.uri` if your MSK cluster has public internet accesss. + If you use PrivateLink to securely connect from MSK to EventBridge, replace the DNS part after `https://` with the correct private DNS name of the (optional) VPC interface endpoint for EventBridge that you created earlier. + Replace the EventBridge event bus ARN in the following configuration with the ARN of your event bus. + Update any Region-specific values. ``` { "connector.class": "software.amazon.event.kafkaconnector.EventBridgeSinkConnector", "aws.eventbridge.connector.id": "msk-eventbridge-tutorial", "topics": "msk-eventbridge-tutorial", "tasks.max": "1", "aws.eventbridge.endpoint.uri": "https://events.us-east-1.amazonaws.com", "aws.eventbridge.eventbus.arn": "arn:aws:events:us-east-1:123456789012:event-bus/example-event-bus", "value.converter.schemas.enable": "false", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "aws.eventbridge.region": "us-east-1", "auto.offset.reset": "earliest", "key.converter": "org.apache.kafka.connect.storage.StringConverter" } ``` For more information about connector configuration, see [eventbridge-kafka-connector](https://github.com/awslabs/eventbridge-kafka-connector). If needed, change the settings for workers and autoscaling. We also recommend to use the latest available (recommended) Apache Kafka Connect version from the dropdown. Under **Access permissions**, use the role created earlier. We also recommend to enable logging to CloudWatch for observability and troubleshooting. Adjust the other optional settings, such as tags, based on your needs. Then, deploy the connector and wait for the status to enter Running state. ## Send messages to Kafka You can configure message encodings, such as Apache Avro and JSON, by specifying different converters using the `value.converter` and, optionally, `key.converter` settings available in Kafka Connect. The [connector example](#connector-ex) in this topic is configured to work with JSON-encoded messages, as indicated by the use of `org.apache.kafka.connect.json.JsonConverter` for `value converter`. When the connector is in Running state, send records to the `msk-eventbridge-tutorial` Kafka topic from your client machine. # Use Debezium source connector with configuration provider Use Debezium source connector This example shows how to use the Debezium MySQL connector plugin with a MySQL-compatible [Amazon Aurora](https://aws.amazon.com/rds/aurora/) database as the source. In this example, we also set up the open-source [AWS Secrets Manager Config Provider](https://github.com/jcustenborder/kafka-config-provider-aws) to externalize database credentials in AWS Secrets Manager. To learn more about configuration providers, see [Tutorial: Externalizing sensitive information using config providers](msk-connect-config-provider.md). **Important** The Debezium MySQL connector plugin [supports only one task](https://debezium.io/documentation/reference/stable/connectors/mysql.html#mysql-property-tasks-max) and does not work with autoscaled capacity mode for Amazon MSK Connect. You should instead use provisioned capacity mode and set `workerCount` equal to one in your connector configuration. To learn more about the capacity modes for MSK Connect, see [Understand connector capacity](msk-connect-capacity.md). # Complete prerequisites to use Debezium source connector Your connector must be able to access the internet so that it can interact with services such as AWS Secrets Manager that are outside of your Amazon Virtual Private Cloud. The steps in this section help you complete the following tasks to enable internet access. + Set up a public subnet that hosts a NAT gateway and routes traffic to an internet gateway in your VPC. + Create a default route that directs your private subnet traffic to your NAT gateway. For more information, see [Enable internet access for Amazon MSK Connect](msk-connect-internet-access.md). **Prerequisites** Before you can enable internet access, you need the following items: + The ID of the Amazon Virtual Private Cloud (VPC) associated with your cluster. For example, *vpc-123456ab*. + The IDs of the private subnets in your VPC. For example, *subnet-a1b2c3de*, *subnet-f4g5h6ij*, etc. You must configure your connector with private subnets. **To enable internet access for your connector** 1. Open the Amazon Virtual Private Cloud console at [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/). 1. Create a public subnet for your NAT gateway with a descriptive name, and note the subnet ID. For detailed instructions, see [Create a subnet in your VPC](https://docs.aws.amazon.com/vpc/latest/userguide/working-with-vpcs.html#AddaSubnet). 1. Create an internet gateway so that your VPC can communicate with the internet, and note the gateway ID. Attach the internet gateway to your VPC. For instructions, see [Create and attach an internet gateway](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Attach_Gateway). 1. Provision a public NAT gateway so that hosts in your private subnets can reach your public subnet. When you create the NAT gateway, select the public subnet that you created earlier. For instructions, see [Create a NAT gateway](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html#nat-gateway-creating). 1. Configure your route tables. You must have two route tables in total to complete this setup. You should already have a main route table that was automatically created at the same time as your VPC. In this step you create an additional route table for your public subnet. 1. Use the following settings to modify your VPC's main route table so that your private subnets route traffic to your NAT gateway. For instructions, see [Work with route tables](https://docs.aws.amazon.com/vpc/latest/userguide/WorkWithRouteTables.html) in the *Amazon Virtual Private Cloud* *User Guide*. **Private MSKC route table** [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/msk/latest/developerguide/mkc-debeziumsource-connector-example-prereqs.html) 1. Follow the instructions in [Create a custom route table](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Routing) to create a route table for your public subnet. When you create the table, enter a descriptive name in the **Name tag** field to help you identify which subnet the table is associated with. For example, **Public MSKC**. 1. Configure your **Public MSKC** route table using the following settings. **** [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/msk/latest/developerguide/mkc-debeziumsource-connector-example-prereqs.html) Now that you have enabled internet access for Amazon MSK Connect you are ready to create a connector. # Create a Debezium source connector This procedure describes how to create a Debezium source connector. 1. **Create a custom plugin** 1. Download the MySQL connector plugin for the latest stable release from the [Debezium](https://debezium.io/releases/) site. Make a note of the Debezium release version you download (version 2.x, or the older series 1.x). Later in this procedure, you'll create a connector based on your Debezium version. 1. Download and extract the [AWS Secrets Manager Config Provider](https://www.confluent.io/hub/jcustenborder/kafka-config-provider-aws). 1. Place the following archives into the same directory: + The `debezium-connector-mysql` folder + The `jcusten-border-kafka-config-provider-aws-0.1.1` folder 1. Compress the directory that you created in the previous step into a ZIP file and then upload the ZIP file to an S3 bucket. For instructions, see [Uploading objects](https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html) in the *Amazon S3 User Guide.* 1. Copy the following JSON and paste it in a file. For example, `debezium-source-custom-plugin.json`. Replace ** with the name that you want the plugin to have, ** with the ARN of the Amazon S3 bucket where you uploaded the ZIP file, and `` with the file key of the ZIP object that you uploaded to S3. ``` { "name": "", "contentType": "ZIP", "location": { "s3Location": { "bucketArn": "", "fileKey": "" } } } ``` 1. Run the following AWS CLI command from the folder where you saved the JSON file to create a plugin. ``` aws kafkaconnect create-custom-plugin --cli-input-json file:// ``` You should see output similar to the following example. ``` { "CustomPluginArn": "arn:aws:kafkaconnect:us-east-1:012345678901:custom-plugin/example-custom-plugin-name/abcd1234-a0b0-1234-c1-12345678abcd-1", "CustomPluginState": "CREATING", "Name": "example-custom-plugin-name", "Revision": 1 } ``` 1. Run the following command to check the plugin state. The state should change from `CREATING` to `ACTIVE`. Replace the ARN placeholder with the ARN that you got in the output of the previous command. ``` aws kafkaconnect describe-custom-plugin --custom-plugin-arn "" ``` 1. **Configure AWS Secrets Manager and create a secret for your database credentials** 1. Open the Secrets Manager console at [https://console.aws.amazon.com/secretsmanager/](https://console.aws.amazon.com/secretsmanager/). 1. Create a new secret to store your database sign-in credentials. For instructions, see [Create a secret](https://docs.aws.amazon.com/secretsmanager/latest/userguide/manage_create-basic-secret.html) in the *AWS Secrets Manager* User Guide. 1. Copy your secret's ARN. 1. Add the Secrets Manager permissions from the following example policy to your [Understand service execution role](msk-connect-service-execution-role.md). Replace ** with the ARN of your secret. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "secretsmanager:GetResourcePolicy", "secretsmanager:GetSecretValue", "secretsmanager:DescribeSecret", "secretsmanager:ListSecretVersionIds" ], "Resource": [ "arn:aws:secretsmanager:us-east-1:123456789012:secret:MySecret-1234" ] } ] } ``` ------ For instructions on how to add IAM permissions, see [Adding and removing IAM identity permissions](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) in the *IAM User Guide*. 1. **Create a custom worker configuration with information about your configuration provider** 1. Copy the following worker configuration properties into a file, replacing the placeholder strings with values that correspond to your scenario. To learn more about the configuration properties for the AWS Secrets Manager Config Provider, see [SecretsManagerConfigProvider](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-config-provider-aws/configProviders/SecretsManagerConfigProvider.html) in the plugin's documentation. ``` key.converter= value.converter= config.providers.secretManager.class=com.github.jcustenborder.kafka.config.aws.SecretsManagerConfigProvider config.providers=secretManager config.providers.secretManager.param.aws.region= ``` 1. Run the following AWS CLI command to create your custom worker configuration. Replace the following values: + ** - a descriptive name for your custom worker configuration + ** - a base64-encoded version of the plaintext properties that you copied in the previous step ``` aws kafkaconnect create-worker-configuration --name --properties-file-content ``` 1. **Create a connector** 1. Copy the following JSON that corresponds to your Debezium version (2.x or 1.x) and paste it in a new file. Replace the `` strings with values that correspond to your scenario. For information about how to set up a service execution role, see [IAM roles and policies for MSK Connect](msk-connect-iam.md). Note that the configuration uses variables like `${secretManager:MySecret-1234:dbusername}` instead of plaintext to specify database credentials. Replace `MySecret-1234` with the name of your secret and then include the name of the key that you want to retrieve. You must also replace `` with the ARN of your custom worker configuration. ------ #### [ Debezium 2.x ] For Debezium 2.x versions, copy the following JSON and paste it in a new file. Replace the ** strings with values that correspond to your scenario. ``` { "connectorConfiguration": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "", "database.port": "3306", "database.user": "<${secretManager:MySecret-1234:dbusername}>", "database.password": "<${secretManager:MySecret-1234:dbpassword}>", "database.server.id": "123456", "database.include.list": "", "topic.prefix": "", "schema.history.internal.kafka.topic": "", "schema.history.internal.kafka.bootstrap.servers": "", "schema.history.internal.consumer.security.protocol": "SASL_SSL", "schema.history.internal.consumer.sasl.mechanism": "AWS_MSK_IAM", "schema.history.internal.consumer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;", "schema.history.internal.consumer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler", "schema.history.internal.producer.security.protocol": "SASL_SSL", "schema.history.internal.producer.sasl.mechanism": "AWS_MSK_IAM", "schema.history.internal.producer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;", "schema.history.internal.producer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler", "include.schema.changes": "true" }, "connectorName": "example-Debezium-source-connector", "kafkaCluster": { "apacheKafkaCluster": { "bootstrapServers": "", "vpc": { "subnets": [ "", "", "" ], "securityGroups": [""] } } }, "capacity": { "provisionedCapacity": { "mcuCount": 2, "workerCount": 1 } }, "kafkaConnectVersion": "2.7.1", "serviceExecutionRoleArn": "", "plugins": [{ "customPlugin": { "customPluginArn": "", "revision": 1 } }], "kafkaClusterEncryptionInTransit": { "encryptionType": "TLS" }, "kafkaClusterClientAuthentication": { "authenticationType": "IAM" }, "workerConfiguration": { "workerConfigurationArn": "", "revision": 1 } } ``` ------ #### [ Debezium 1.x ] For Debezium 1.x versions, copy the following JSON and paste it in a new file. Replace the ** strings with values that correspond to your scenario. ``` { "connectorConfiguration": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "", "database.port": "3306", "database.user": "<${secretManager:MySecret-1234:dbusername}>", "database.password": "<${secretManager:MySecret-1234:dbpassword}>", "database.server.id": "123456", "database.server.name": "", "database.include.list": "", "database.history.kafka.topic": "", "database.history.kafka.bootstrap.servers": "", "database.history.consumer.security.protocol": "SASL_SSL", "database.history.consumer.sasl.mechanism": "AWS_MSK_IAM", "database.history.consumer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;", "database.history.consumer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler", "database.history.producer.security.protocol": "SASL_SSL", "database.history.producer.sasl.mechanism": "AWS_MSK_IAM", "database.history.producer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;", "database.history.producer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler", "include.schema.changes": "true" }, "connectorName": "example-Debezium-source-connector", "kafkaCluster": { "apacheKafkaCluster": { "bootstrapServers": "", "vpc": { "subnets": [ "", "", "" ], "securityGroups": [""] } } }, "capacity": { "provisionedCapacity": { "mcuCount": 2, "workerCount": 1 } }, "kafkaConnectVersion": "2.7.1", "serviceExecutionRoleArn": "", "plugins": [{ "customPlugin": { "customPluginArn": "", "revision": 1 } }], "kafkaClusterEncryptionInTransit": { "encryptionType": "TLS" }, "kafkaClusterClientAuthentication": { "authenticationType": "IAM" }, "workerConfiguration": { "workerConfigurationArn": "", "revision": 1 } } ``` ------ 1. Run the following AWS CLI command in the folder where you saved the JSON file in the previous step. ``` aws kafkaconnect create-connector --cli-input-json file://connector-info.json ``` The following is an example of the output that you get when you run the command successfully. ``` { "ConnectorArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector/example-Debezium-source-connector/abc12345-abcd-4444-a8b9-123456f513ed-2", "ConnectorState": "CREATING", "ConnectorName": "example-Debezium-source-connector" } ``` # Update a Debezium connector configuration To update the configuration of the Debezium connector, follow these steps: 1. Copy the following JSON and paste it to a new file. Replace the `` strings with values that correspond to your scenario. ``` { "connectorArn": , "connectorConfiguration": , "currentVersion": } ``` 1. Run the following AWS CLI command in the folder where you saved the JSON file in the previous step. ``` aws kafkaconnect update-connector --cli-input-json file://connector-info.json ``` The following is an example of the output when you run the command successfully. ``` { "connectorArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector/example-Debezium-source-connector/abc12345-abcd-4444-a8b9-123456f513ed-2", "connectorOperationArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector-operation/example-Debezium-source-connector/abc12345-abcd-4444-a8b9-123456f513ed-2/41b6ad56-3184-479b-850a-a8bedd5a02f3", "connectorState": "UPDATING" } ``` 1. You can now run the following command to monitor the current state of the operation: ``` aws kafkaconnect describe-connector-operation --connector-operation-arn ``` For a Debezium connector example with detailed steps, see [Introducing Amazon MSK Connect - Stream Data to and from Your Apache Kafka Clusters Using Managed Connectors](https://aws.amazon.com/blogs/aws/introducing-amazon-msk-connect-stream-data-to-and-from-your-apache-kafka-clusters-using-managed-connectors/).