The Amazon EventBridge sink connector pulls data from Pulsar topics and persists data to Amazon EventBridge.

Quick start

Prerequisites

The prerequisites for connecting an AWS EventBridge sink connector to external systems include:

1. Create EventBridge and EventBus in AWS.
2. Create the AWS User and create AccessKey(Please record AccessKey and SecretAccessKey).
3. Assign permissions to AWS User, and ensure they have the PutEvents permissions to the AWS EventBus. For details, see permissions for event buses

  {
    "Version": "2012-10-17",
    "Statement": [
      {
        "Sid": "AllowAccountToPutEvents",
        "Effect": "Allow",
        "Principal": {
          "AWS": "<ACCOUNT_ID>"
        },
        "Action": "events:PutEvents",
        "Resource": "{EventBusArn}"
      }
    ]
}

• You can set permissions directly for this user. With this method, when you create a connector, you only need to configure accessKey and secretAccessKey.
• Or you can use Security Token Service, this video explains how to use STS on AWS. With this method, when you create a connector, in addition to configuring accessKey and secretAccessKey, you also need to configure role and roleSessionName.

1. Create a Rule in EventBridge.

• The data structure sent to Event Bridge is described [here](## Metadata mapping), and you can create event pattern based on this structure.
• Set the target according to your needs. If you're testing this connector, you can set the target to Cloud Watch.

1. Create a connector

The following command shows how to use pulsarctl to create a builtin connector. If you want to create a non-builtin connector, you need to replace --sink-type aws-eventbridge with --archive /path/to/pulsar-io-aws-eventbridge.nar. You can find the button to download the nar package at the beginning of the document.

pulsarctl sinks create \
  --sink-type aws-eventbridge \
  --name aws-eventbridge-sink \
  --tenant public \
  --namespace default \
  --inputs "Your topic name" \
  --parallelism 1 \
  --sink-config \
  '{
    "accessKeyId": "Your AWS access key", 
    "secretAccessKey": "Your AWS secret access key",
    "region": "Your event bridge region",
    "eventBusName": "Your eventbus name"
  }'

The --sink-config is the minimum necessary configuration for starting this connector, and it is a JSON string. You need to substitute the relevant parameters with your own. If you want to configure more parameters, see Configuration Properties for reference.

2. Send messages to the topic

        PulsarClient client = PulsarClient.builder()
            .serviceUrl("{{Your Pulsar URL}}")
            .build();

        Producer<String> producer = client.newProducer(Schema.STRING)
            .topic("{{Your topic name}}")
            .create();

        String message = "{\"msg\": \"msg-data\"}";
        MessageId msgID = producer.send(message);
        System.out.println("Publish " + message + " and message ID " + msgID);
        
        producer.flush();
        producer.close();
        client.close();

3. Show data on AWS EventBridge

The connector will send the following format of JSON event to EventBridge.

{
  "version": "0",
  "id": "6a7e8feb-b491-4cf7-a9f1-bf3703467718",
  "detail-type": "{{Your topic name}}",
  "source": "{{Your connector name}}",
  "account": "111122223333",
  "time": "2017-12-22T18:43:48Z",
  "region": "us-west-1",
  "resources": [
    "arn:aws:ec2:us-west-1:123456789012:instance/i-1234567890abcdef0"
  ],
  "detail": {
    "data": {
      "msg": "msg-data"
    },
    "message_id": "124:191:0"
  }
}

Configuration Properties

Before using the AWS EventBridge sink connector, you need to configure it. This table outlines the properties and the descriptions.

For details about this connector's advanced features and configurations, see Advanced features.

Advanced features

Delivery guarantees

The AWS EventBridge sink connector provides two delivery guarantees: at-most-once and at-least-once.

Data convert

In AWS EventBridge, all events is JSON format.

Pulsar supports multiple schema types. When receiving the data from Pulsar, the AWS EventBridge sink connectors recognize it and convert it to a JSON string according to the following table:

In EventBridge, the user data is in the detail$data field.

{
  "version": "0",
  "id": "6a7e8feb-b491-4cf7-a9f1-bf3703467718",
  "detail": {
    "data": {
      "instance-id": " i-1234567890abcdef0",
      "state": "terminated"
    }
  }
}

Metadata mapping

In EventBridge, a complete event contains many system fields. These system fields can help you to configure the rule.

An Event containing event data:

{
  "version": "0",
  "id": "6a7e8feb-b491-4cf7-a9f1-bf3703467718",
  "source-type": "test-aws-event-bridge-sink-connector",
  "detail-type": "topic_name_test_1",
  "source": "aws.ec2",
  "account": "111122223333",
  "time": "2017-12-22T18:43:48Z",
  "region": "us-west-1",
  "resources": [
    "arn:aws:ec2:us-west-1:123456789012:instance/i-1234567890abcdef0"
  ],
  "detail": {
    "data": {
      "instance-id": " i-1234567890abcdef0",
      "state": "terminated"
    }
  }
}

This connector maps the following fields:

• sourceType: The default value is ${{Connector Name}}.
• detailType: The default value is ${{Topic Name}}.

And, this connector supports setting the metadata of Pulsar to every Event (set in the detail field).

You can select the desired metadata through the following configuration:

#
optional: schema_version | partition | event_time | publish_time
#
message_id | sequence_id | producer_name | key | properties
metaDataField = event_time, message_id

An Event containing metadata :

{
  "version": "0",
  "id": "6a7e8feb-b491-4cf7-a9f1-bf3703467718",
  "source-type": "test-aws-event-bridge-sink-connector",
  "detail-type": "topic_name_test_1",
  "source": "aws.ec2",
  "account": "111122223333",
  "time": "2017-12-22T18:43:48Z",
  "region": "us-west-1",
  "resources": [
    "arn:aws:ec2:us-west-1:123456789012:instance/i-1234567890abcdef0"
  ],
  "detail": {
    "data": {
      "instance-id": " i-1234567890abcdef0",
      "state": "terminated"
    },
    "event_time": 789894645625,
    "message_id": "1,1,1"
  }
}

Parallelism

You can configure the parallelism of Sink execution by using the scheduling mechanism of the Function, and multiple sink instances will be scheduled to run on different worker nodes. Multiple sinks will consume messages together according to the configured subscription mode.

Since EventBus doesn't need to guarantee sequentiality, the connectors support the shared subscription model.

To increase the write throughput, you can configure the following:

parallelism = 4

When retainOrdering is set to false, the Shared subscription mode is used.

Batch Put

AWS EventBridge connectors support batch put events, which are mainly controlled by the following three parameters:

• batchSize: When the buffered message is larger than batchSize, it will trigger flush (put) events. 0 means no trigger.
• maxBatchBytes: When the buffered message data size is larger than maxBatchBytes, it will trigger flush pending events. This value should be less than 256000 and greater than 0, The default value is 640.
• batchTimeMs: When the interval from the last flush exceeds batchTimeMs, it will trigger flush pending events. 0 means no trigger.

In addition to these three parameters that control flush behavior, in AWS EventBridge, batches larger than 256KB per write are not allowed. So, when the buffered message is larger than 256KB, it will trigger a flush.

Retry Put

In AWS Event Bridge, about Handling failures with PutEvents, It suggests retrying each error message until it succeeds.

This connector will provide two flow configs for the controller's retry strategy:

maxRetryCount: 100 // Maximum retry send event count, when event send failed.
intervalRetryTimeMs: 1000 //The interval time(milliseconds) for each retry, when event send failed.

More Links

• GitHub Repo
• Announcing the Amazon EventBridge Sink Connector for Apache Pulsar
• Amazon EventBridge connector is now integrated with StreamNative Cloud