How do I deploy EventBridge Pipes with Amazon MSK?

Resolution

Note: For more information about Amazon MSK and how to create a cluster, see Get started using Amazon MSK.

Prerequisites

• MSK cluster authentication: EventBridge must have permission to access the MSK cluster, retrieve records, and perform other tasks.
• Network configuration: EventBridge must have access to the Amazon Virtual Private Cloud (Amazon VPC) resources associated with your MSK cluster.
• Amazon MSK execution role permissions: EventBridge requires permissions to manage resources that are related to your MSK topic.

Deploy EventBridge Pipes with Amazon MSK

Complete the following steps:

1. Navigate to the EventBridge console in your respective AWS Region.
2. In the navigation panel, choose Pipes.
3. Choose Create pipe.
4. Enter a name for the pipe.
5. (Optional) Enter a description for the pipe.

Select Amazon MSK as the source for the pipe

Complete the following steps:

1. For Select source, choose Amazon MSK.
2. Choose the Amazon MSK cluster that you created previously.
3. Enter the name of the Apache Kafka topic used to store records in your Apache Kafka cluster.
4. (Optional) Provide the ID of an Apache Kafka consumer group to join. For more information, see Amazon Managed Streaming for Apache Kafka topic as a source in EventBridge Pipes.
5. (Optional) Choose the authentication method and secret key.

Apply additional settings

Complete the following steps:

1. (Optional) Configure Batch size and batch window.
2. Configure the starting position.
   Note: For starting position, latest is the default setting. If required, change to Trim horizon. Trim horizon is the same as Earliest for Apache Kafka.
3. (Optional) Apply settings for Filtering and Enrichment.
4. Choose your target from the Target service list.
5. (Optional) Define an input transformer.
6. Confirm that the pipe configuration is correct. Then, choose Create pipe.

When the pipe transitions to Running status, EventBridge internally polls for new messages from the source. Then, it synchronously invokes the target.

Note: The event payload contains an array of messages. Each item contains details of the MSK topic and MSK partition identifier, together with a timestamp and a base64-encoded message.

The following is an MSK event example:

[  {
    "eventSource": "aws:kafka",
    "eventSourceArn": "arn:aws:kafka:sa-east-1:123456789012:cluster/vpc-2priv-2pub/751d2973-a626-431c-9d4e-d7975eb44dd7-2",
    "eventSourceKey": "mytopic-0",
    "topic": "mytopic",
    "partition": "0",
    "offset": 15,
    "timestamp": 1545084650987,
    "timestampType": "CREATE_TIME",
    "key": "abcDEFghiJKLmnoPQRstuVWXyz1234==",
    "value": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
    "headers": [
      {
        "headerKey": [
          104,
          101,
          97,
          100,
          101,
          114,
          86,
          97,
          108,
          117,
          101
        ]
      }
    ]
  }
]

Troubleshoot invocation-related errors

Use Amazon CloudWatch metrics through the CloudWatch console in your respective Region for the pipe to isolate activation, invocation, or throttling-related issues.

When you create, delete, and update pipes, the operation might result in a failure state. For more information, see Pipe failure states.

When you invoke a pipe, two main types of errors can occur: pipe internal errors or customer invocation errors.

A pipe internal error might occur if an HTTP connection fails when it attempts to invoke the customer target service. Or, an error might occur if there is a transient drop in availability on the pipe service itself.

A customer invocation error might occur if there are insufficient permissions on the pipe to invoke the target. Or, an error might occur if there are logic errors in your synchronously invoked customer AWS Lambda, AWS Step Functions, API destination, or Amazon API Gateway endpoint. For more information, see permissions.

EventBridge Pipes doesn't support cross-account processing of an Amazon MQ Broker from a different account. For more information, see Amazon EventBridge Pipes error handling and troubleshooting.

Related information

Creating an Amazon EventBridge pipe