> ## Documentation Index > Fetch the complete documentation index at: https://docs.automq.com/llms.txt > Use this file to discover all available pages before exploring further. # Amazon EventBridge Sink Connector > Configure and operate the Amazon EventBridge Sink Connector in AutoMQ Connect, including event mapping, AWS permissions, S3 offloading, monitoring, and troubleshooting. ## Overview The Amazon EventBridge Sink Connector sends records from one or more Kafka topics to a specified Amazon EventBridge event bus. Each Kafka record is converted into an EventBridge event. The Kafka topic, partition, offset, timestamp, header, key, and value are stored in the event's `detail`, and the event's `source` is generated from the Connector identifier. It is suitable for connecting business events, CDC records, or application messages in Kafka to EventBridge, where EventBridge rules can route them to downstream AWS services or other targets. By default, the event `detail-type` is generated from the Topic, and the event time is set by EventBridge when the event is received. You can customize `detail-type` by Topic or message content, and you can also offload a specified part of the event Value to S3 while retaining reference information in the EventBridge event. ## Prerequisites * The target EventBridge event bus has been created, and the AWS identity used by the Connector has `events:PutEvents` permission for that event bus. If you enable an IAM role, profile, or custom credentials provider, also prepare the corresponding trust relationship and credential source. * If S3 offloading is enabled, the target S3 Bucket has been created, and the AWS identity used by the Connector has `s3:PutObject` permission for that Bucket. * If you use Avro, Protobuf, AWS Glue Schema Registry, or a custom Converter, place the related Converter classes and dependencies in a plugin path loadable by the Kafka Connect Worker, and prepare the permissions required to read the Schema. * If you configure a custom detail-type, time, or credentials provider class, place the related classes and dependencies in the Kafka Connect Worker's plugin path, and satisfy the corresponding interface and no-argument constructor requirements. ## License Uses Apache License 2.0. ## Quick Start Prepare the Connect cluster, Kafka, input topic, and target EventBridge event bus in advance, and confirm network connectivity and access permissions. For creation and management operations, see [Manage Connectors](../manage-connectors). The following configuration uses a schema-free JSON value to send records to the specified event bus. ```properties theme={null} connector.class=software.amazon.event.kafkaconnector.EventBridgeSinkConnector topics= aws.eventbridge.connector.id=eventbridge-sink aws.eventbridge.eventbus.arn=arn:aws:events:us-east-1::event-bus/ aws.eventbridge.region=us-east-1 key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=false ``` Replace the topic, AWS account, event bus name, and AWS identity settings before applying the configuration. The event `detail-type` defaults to `kafka-connect-`, and `detail` contains Kafka record metadata and the decoded key and value. `value.converter.schemas.enable=false` applies to JSON values without a schema envelope. For other message formats, use the corresponding converter and its dependencies. ## Configuration ### Input and Tasks #### `topics` Specifies the list of Kafka topics consumed by the Sink Connector. * **Type**: `list` * **Default**: Empty list `[]` * **Importance**: High * **Valid values / notes**: Use comma-separated Topic names. Configure exactly one of `topics` and `topics.regex`; at least one must be non-empty. #### `topics.regex` Uses a regular expression to select the Kafka topics consumed by the Sink Connector. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: High * **Valid values / notes**: Use a Java regular expression. This setting is mutually exclusive with `topics`; they cannot be configured together. #### `tasks.max` Requests the maximum number of Tasks to create for this Connector. * **Type**: `int` * **Default**: `1` * **Importance**: High * **Valid values / notes**: Must be at least `1`. Actual parallelism is also limited by Kafka partitions, Worker resources, and EventBridge throughput. No global ordering guarantee is provided across Tasks, Topics, or partitions. ### EventBridge Connection and Target #### `aws.eventbridge.connector.id` The unique identifier of the Connector, used to generate the EventBridge event `source` and as the IAM role session name. * **Type**: `string` * **Default**: No fixed default, required * **Importance**: High * **Valid values / notes**: Must not be empty or contain only spaces. There are no additional character or length restrictions. The event `source` has the form `kafka-connect.`. #### `aws.eventbridge.region` The AWS Region where the target EventBridge event bus is located. * **Type**: `string` * **Default**: No fixed default, required * **Importance**: High * **Valid values / notes**: Use a Region name supported by the AWS SDK, such as `us-east-1`. #### `aws.eventbridge.eventbus.arn` The ARN of the target EventBridge event bus. * **Type**: `string` * **Default**: No fixed default, required * **Importance**: High * **Valid values / notes**: Use the full ARN of the target event bus, such as `arn:aws:events:us-east-1:123456789012:event-bus/orders`. #### `aws.eventbridge.endpoint.uri` Overrides the EventBridge service endpoint used by the default AWS SDK. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: Medium * **Valid values / notes**: Leave empty to use the default AWS SDK endpoint. A non-empty value must be a parseable URI; an invalid URI may cause the Task to fail to start. #### `aws.eventbridge.eventbus.global.endpoint.id` Specifies the EventBridge global endpoint ID. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: Medium * **Valid values / notes**: Leave empty to omit the global endpoint ID. A non-empty value uses a two-part format, such as `abcde.veo`. ### AWS Authentication #### `aws.eventbridge.auth.credentials_provider.class` Specifies a custom AWS credentials provider class. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: Medium * **Valid values / notes**: Leave empty to use the default credentials provider. A non-empty class must implement `AwsCredentialsProvider` and have a no-argument constructor. If it implements `Configurable`, the Connector's original configuration is passed to the class. #### `aws.eventbridge.iam.role.arn` Specifies the IAM role ARN assumed through STS by the Connector. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: Medium * **Valid values / notes**: Leave empty to disable this role. A non-empty value must use the IAM role ARN format, such as `arn:aws:iam::123456789012:role/EventBridgePutEventsRole`. When no custom credentials provider is configured, the Connector assumes this role through STS. #### `aws.eventbridge.iam.external.id` Provides an External ID for IAM role-based authentication. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: Medium * **Valid values / notes**: Takes effect only when `aws.eventbridge.iam.role.arn` is used. The External ID is sensitive configuration; do not write it to logs, examples, or public repositories. #### `aws.eventbridge.iam.profile.name` Specifies the credentials Profile in the AWS shared configuration file. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: Medium * **Valid values / notes**: Applies only to the default credentials provider path. When using this setting, do not also set `AWS_PROFILE`, `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, or `AWS_SESSION_TOKEN`, as they can conflict with profile selection. ### Event Mapping #### `aws.eventbridge.detail.types` Sets the `detail-type` for EventBridge events, either by topic mapping or with one expression for all topics. * **Type**: `list` * **Default**: `[kafka-connect-${topic}]` * **Importance**: Medium * **Valid values / notes**: You can use a single expression, such as `orders-${topic}`, or a static value, such as `business-event`. For multiple values, use `topic:detail-type`, such as `orders:order-created,customers:customer-updated`. Topics without a match fall back to `kafka-connect-`. This setting no longer takes effect after a custom detail-type mapper is configured. #### `aws.eventbridge.detail.types.mapper.class` Specifies the class that calculates `detail-type` from the Topic or record content. * **Type**: `string` * **Default**: `software.amazon.event.kafkaconnector.mapping.DefaultDetailTypeMapper` * **Importance**: Medium * **Valid values / notes**: The class must implement `DetailTypeMapper` and have a no-argument constructor. When using the built-in `JsonPathDetailTypeMapper`, you must also configure `aws.eventbridge.detail.types.jsonpathmapper.fieldref`. After this class is configured, `aws.eventbridge.detail.types` is ignored. #### `aws.eventbridge.detail.types.jsonpathmapper.fieldref` Specifies the JSONPath from which `JsonPathDetailTypeMapper` extracts `detail-type` from the Kafka record Value. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: Medium * **Valid values / notes**: Use a parseable and definite JSONPath, such as `$.metadata.event-type`. The extracted result must be a non-empty string. If the path does not exist, the result is empty or `null`, or the result is not a string, the mapper falls back to the record Topic. This setting is required only when using `JsonPathDetailTypeMapper`. #### `aws.eventbridge.time.mapper.class` Specifies the class that calculates the EventBridge `time` field for each event. * **Type**: `string` * **Default**: `software.amazon.event.kafkaconnector.mapping.DefaultTimeMapper` * **Importance**: Medium * **Valid values / notes**: The class must be loadable, implement `TimeMapper`, and have a no-argument constructor. The default value does not provide a custom time; EventBridge sets the time when the `PutEvents` call is made. #### `aws.eventbridge.eventbus.resources` Adds a `resources` list to each EventBridge event. * **Type**: `list` * **Default**: Empty list `[]` * **Importance**: Medium * **Valid values / notes**: Use comma-separated resource values. EventBridge validates the resource format. ### S3 Payload Offloading #### `aws.eventbridge.offloading.default.s3.bucket` Specifies the S3 Bucket used to offload event payloads. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: Medium * **Valid values / notes**: Leave empty to disable offloading. A non-empty value enables claim-check processing, and the Bucket must already exist and allow the current AWS identity to write objects. #### `aws.eventbridge.offloading.default.fieldref` Specifies the JSONPath that selects the part of the event `detail` to offload to S3. * **Type**: `string` * **Default**: `$.detail.value` * **Importance**: Medium * **Valid values / notes**: Takes effect only when an S3 Bucket is configured. The path must be a definite JSONPath beginning with `$.detail.value`; array or wildcard paths are not supported. If no value matches or the value is `null`, the original event is passed through. Empty objects and empty arrays are offloaded. #### `aws.eventbridge.offloading.default.s3.endpoint.uri` Overrides the service endpoint used for S3 offloading. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: Medium * **Valid values / notes**: Used only when S3 offloading is enabled. Leave empty to use the default AWS SDK S3 endpoint. A non-empty value must be a parseable URI. ### Delivery and Error Handling #### `aws.eventbridge.retries.max` Sets the maximum number of Connector-level retries when EventBridge delivery fails. * **Type**: `int` * **Default**: `2` * **Importance**: Medium * **Valid values / notes**: The range is `0` to `10`; `0` means no Connector-level retries. This value is also used by the AWS SDK client, so the actual number of requests may be higher than the Connector-level setting. #### `aws.eventbridge.retries.delay` Sets the wait time between two Connector retries, in milliseconds. * **Type**: `int` * **Default**: `200` * **Importance**: Medium * **Valid values / notes**: The Connector does not reject negative values. Use a non-negative integer because a negative value does not produce an effective wait. This setting does not change whether an error is retryable. #### `errors.tolerance` Sets Kafka Connect's tolerance mode for conversion, converter, and task errors. * **Type**: `string` * **Default**: `none` * **Importance**: Medium * **Valid values / notes**: Valid values are `none` and `all`. `all` should generally be used together with DLQ configuration. It is different from the EventBridge delivery retries controlled by `aws.eventbridge.retries.max`. #### `errors.deadletterqueue.topic.name` Specifies the DLQ Topic that receives failed records handled by the Kafka Connect error reporter. * **Type**: `string` * **Default**: Empty string `""` * **Importance**: Medium * **Valid values / notes**: Leave empty to omit DLQ configuration. After it is configured, the Connector can report record-level non-retryable failures to this Topic. Topic creation and permissions are managed by Kafka Connect and the Kafka cluster. #### `errors.deadletterqueue.topic.replication.factor` Sets the replication factor used when Kafka Connect creates the DLQ Topic. * **Type**: `short` * **Default**: `3` * **Importance**: Medium * **Valid values / notes**: Takes effect only when `errors.deadletterqueue.topic.name` is configured. The value must match the number of Brokers available in the Kafka cluster and the Topic creation policy. ### Message Conversion #### `key.converter` Specifies the Converter that decodes the Kafka message Key into a Kafka Connect value. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid values / notes**: If unset, the Worker's Converter configuration is used. When explicitly set, it must be an instantiable Converter class. The decoded Key is written to `detail.key` in the EventBridge event. #### `value.converter` Specifies the Converter that decodes the Kafka message Value into a Kafka Connect value. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid values / notes**: If unset, the Worker's Converter configuration is used. When explicitly set, it must be an instantiable Converter class. Avro, Protobuf, and Schema Registry dependencies are provided by the Worker environment. Records that fail to decode are not included in an EventBridge send batch. ## Best Practices ### Map Multiple Topics to Routable Event Types **Applicable scenario**: Multiple business Topics need to enter the same EventBridge event bus, and stable `detail-type` values are needed to distinguish order, customer, and other event types so that rules do not have to rely only on Topic names. **Configuration example**: Add the following settings to the Quick Start configuration while retaining the existing event bus and identity settings. ```properties theme={null} topics=orders,customers aws.eventbridge.detail.types=orders:order-created,customers:customer-updated ``` **Key points**: Each Topic uses its corresponding `detail-type`, and a Topic without a match falls back to the default value. Multiple mappings must use the `topic:detail-type` format. If you switch to a custom mapper, remove or ignore `aws.eventbridge.detail.types` to avoid assuming that the two mapping methods are combined. ### Offload Large Payloads to S3 **Applicable scenario**: Kafka records contain relatively large business Values, while EventBridge events need to retain filterable metadata and downstream consumers need to read the complete payload on demand. **Configuration example**: Add the following settings to the Quick Start configuration, using an existing S3 resource that permits writes for the Bucket. ```properties theme={null} aws.eventbridge.offloading.default.s3.bucket= aws.eventbridge.offloading.default.fieldref=$.detail.value ``` **Key points**: The matched value is written to S3. Its original location is removed from the EventBridge event, which receives `dataref` and `datarefJsonPath` reference information. The target AWS identity must have permission to call `s3:PutObject`. The path can only begin with `$.detail.value`; a missing match or a `null` value is not offloaded. S3 offloading does not replace downstream management of object lifecycle, access permissions, and cleanup policies. ### Configure Delivery Retries and Send Unrecoverable Records to a DLQ **Applicable scenario**: Temporary EventBridge or network failures need automatic retries, while conversion failures or non-retryable records must not block subsequent messages and should be retained for investigation and compensation. **Configuration example**: Add the following settings to the Quick Start configuration, and configure separate consumption and retention policies for the DLQ Topic. ```properties theme={null} aws.eventbridge.retries.max=3 aws.eventbridge.retries.delay=1000 errors.tolerance=all errors.deadletterqueue.topic.name= errors.deadletterqueue.topic.replication.factor=3 ``` **Key points**: `aws.eventbridge.retries.max` controls Connector-level delivery retries, while the DLQ is handled by the Kafka Connect error reporter. They address different problems. The AWS SDK also uses this retry value, so the actual number of calls may exceed the direct estimate from the Connector-level setting. Before going live, establish alerting, retention, and replay procedures for the DLQ. Do not treat `errors.tolerance=all` as a guarantee of successful delivery. ## Monitoring ### What to Monitor Monitor whether the Kafka Connect worker, Connector, and task are in the expected state, as well as task record throughput, processing latency, consumer lag, offset commits, failed records, retry counts, and error logs. Also monitor worker JVM heap memory, GC, threads, and CPU. After enabling `errors.tolerance=all` and a DLQ, monitor writes to and backlog in the DLQ topic, and correlate DLQ growth with EventBridge, converter, or permission errors. ### Import the Grafana Dashboard Download the [Kafka Connect Grafana dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), import the JSON into Grafana, and select the data source that collects Kafka Connect metrics. After importing, confirm that the data source name and labels match the metric collection configuration of the current Connect cluster. ## Limitations * A single EventBridge `PutEvents` request can contain at most 10 events, and the event size limit is 256 KiB. A single event that exceeds the limit is not resolved by the Connector-level retry mechanism; S3 offloading or a DLQ should generally be used. * The Connector provides at-least-once delivery semantics for valid Kafka records. If a failure occurs after EventBridge succeeds but before Kafka Connect commits the Offset, the record may be sent again. This cannot be used to claim exactly-once delivery or idempotency on the target side. * S3 offloading supports only definite JSONPath under `$.detail.value`. It does not support the Key, Header, or top-level EventBridge fields, nor array or wildcard paths. * After `aws.eventbridge.detail.types.mapper.class` is configured, `aws.eventbridge.detail.types` does not participate in mapping. The two settings do not take effect in combination. ## FAQ ### The Task reports that the event bus ARN, Region, or Connector identifier is invalid at startup Check whether `aws.eventbridge.connector.id` is empty, whether `aws.eventbridge.region` is an AWS SDK-supported Region name, and whether `aws.eventbridge.eventbus.arn` is a complete event bus ARN. After correcting the values, validate the Connector configuration again and confirm that the Region represented by the ARN matches `aws.eventbridge.region`. ### EventBridge returns a permission error and the record is not delivered Confirm the AWS identity source actually used by the Connector. When using an IAM role, check the trust relationship, role ARN, and External ID. When using a profile, confirm that the worker can read the corresponding AWS configuration file, and remove AWS credential environment variables that override profile selection. The target identity must be allowed to call `events:PutEvents` on the target event bus. ### `detail-type` does not use the event type in the message With the default mapping, `detail-type` comes from the Topic or `aws.eventbridge.detail.types`; it does not automatically read a field from the Value. To extract the event type from a JSON Value, set `aws.eventbridge.detail.types.mapper.class` to `software.amazon.event.kafkaconnector.mapping.JsonPathDetailTypeMapper` and configure a definite `aws.eventbridge.detail.types.jsonpathmapper.fieldref`. The extracted result must be a non-empty string; otherwise, the mapper falls back to the Topic. ### Large messages still cannot be sent to EventBridge Confirm that `aws.eventbridge.offloading.default.s3.bucket` is non-empty, that `aws.eventbridge.offloading.default.fieldref` begins with `$.detail.value` and actually matches a field in the Value, and that the AWS identity has S3 `PutObject` permission. If the event has no matching value or the value is `null`, the Connector passes through the original event and does not generate an S3 reference. ### Failed records do not appear in the DLQ Confirm that both `errors.tolerance=all` and `errors.deadletterqueue.topic.name` are configured, and check DLQ Topic creation, write permissions, and replication factor. EventBridge-retryable errors are retried first according to `aws.eventbridge.retries.max`. Not every error is classified as a reportable record-level failure; use Task logs to determine whether the cause is exhausted retries, a conversion failure, or a Task-level error.