Skip to main content

​
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. The following configuration uses a schema-free JSON value to send records to the specified event bus.
Replace the topic, AWS account, event bus name, and AWS identity settings before applying the configuration. The event detail-type defaults to kafka-connect-<topic>, 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.<connector-id>.

​
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-<topic>. 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.
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.
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.
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, 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.