Setting the file. One moment.
Skill 111 · Amazon Eventbridge Event Bus
Subchapter 111.3
references/code-samples.mdMarkdown13 KBView on GitHub
Load this when writing the first CLI or SDK call: creating a bus and a subscriber, publishing with either API, replaying from history, granting cross-account access, or checking per-entry results. Each sample uses only fields the service requires plus the ones the skill body tells you to always set. The CloudFormation form of the same setup is in .
A bus, the waiter (chapter 15), and one SQS-targeted subscriber with a filter and a dead-letter queue (chapters 7, 9, and 12):
BUS_ARN=$(aws eventbridgev2 create-event-bus --name my-bus --query EventBusArn --output text)
aws eventbridgev2 wait event-bus-active --event-bus-arn "$BUS_ARN"
aws eventbridgev2 create-subscriber --name orders-to-queue \
--event-bus-arn "$BUS_ARN" \
--filter-configuration '{"Filters":[{"Scope":"DATA","Pattern":"{\"detail\":{\"orderId\":[{\"exists\":true}]}}"}]}' \
--invoke-configuration '{"TargetArn":"arn:aws:sqs:us-east-1:111122223333:my-queue","RoleArn":"arn:aws:iam::111122223333:role/my-delivery-role"}' \
--on-failure-configuration '{"Arn":"arn:aws:sqs:us-east-1:111122223333:my-dlq"}'You SHOULD allow a short delay after the subscriber is created before publishing, because otherwise
the first
event can be missed (see provisioning-and-state.md). The delivery role needs
sqs:SendMessage on both the queue and the dead-letter queue (chapter 13). No SqsParameters block
appears above because a standard queue needs none of its members; only a universal target requires a
parameter block of its own (chapter 9).
StartingPosition is a plain enum, LATEST or POINT_IN_TIME; the point itself goes in the separate
PointInTimeConfiguration member (chapter 17). So --starting-position takes the bare enum value, with
no JSON wrapper:
aws eventbridgev2 create-subscriber --name orders-replay \
--event-bus-arn "$BUS_ARN" \
--starting-position POINT_IN_TIME \
--point-in-time-configuration '{"PointType":"HORIZON"}' \
--invoke-configuration '{"TargetArn":"arn:aws:sqs:us-east-1:111122223333:my-queue","RoleArn":"arn:aws:iam::111122223333:role/my-delivery-role"}'{"PointType":"TIMESTAMP","StartingPoint":1756605600} starts from a timestamp instead; StartingPoint
is epoch seconds (chapter 17). No --filter-configuration appears above because it is optional on
create; an unfiltered subscriber receives every event on the bus (see
filters-and-expressions.md).
Each shape below is the --invoke-configuration for one target type, with the delivery-role action it
needs (chapter 13). At most one parameter block, and only a universal target requires its own
(chapter 9). Per-target delivery behaviour and traps are in
target-configuration.md.
SNS topic. The role needs sns:Publish; no parameter block; a RAW transformer delivers the payload
alone to the topic’s subscribers:
'{"TargetArn":"arn:aws:sns:us-east-1:111122223333:my-topic","RoleArn":"'"$ROLE_ARN"'"}'Lambda function. The role needs lambda:InvokeFunction. The function receives a JSON array of
events, a batch, not a single event object, so you MUST write the handler for a list:
'{"TargetArn":"arn:aws:lambda:us-east-1:111122223333:function:my-fn","RoleArn":"'"$ROLE_ARN"'"}'Kinesis stream. The role needs kinesis:PutRecords; KinesisParameters may be omitted, and
PartitionKey accepts a JSONata expression, which it should, being per-message (chapter 9, rule 2):
'{"TargetArn":"arn:aws:kinesis:us-east-1:111122223333:stream/my-stream","RoleArn":"'"$ROLE_ARN"'","KinesisParameters":{"PartitionKey":"{% $events.Data.detail.orderId %}"}}'Step Functions state machine. The role needs states:StartExecution:
'{"TargetArn":"arn:aws:states:us-east-1:111122223333:stateMachine:my-machine","RoleArn":"'"$ROLE_ARN"'"}'Another new custom event bus. The role needs both events:PutEvents and events:PutRawEvents,
because the action is chosen per forwarded event (see authorization.md):
'{"TargetArn":"arn:aws:events:us-east-1:111122223333:event-busv2/dest-bus/exampleid0123456789abcdef","RoleArn":"'"$ROLE_ARN"'"}'Universal target, here Step Functions’ startExecution called directly. The parameter block and
BatchConfiguration are required, and the field names inside Input are PascalCase Smithy member
names, StateMachineArn not stateMachineArn (chapter 10):
'{"TargetArn":"arn:aws:events:::aws-sdk:sfn:startExecution","RoleArn":"'"$ROLE_ARN"'","UniversalTargetParameters":{"Input":"{\"StateMachineArn\":\"arn:aws:states:us-east-1:111122223333:stateMachine:my-machine\",\"Input\":\"{}\"}"},"BatchConfiguration":{"MaxBatchSize":1,"MaxBatchWindowInSeconds":0}}'A worked universal dynamodb:putItem with a JSONata Input is in
target-configuration.md. HTTP targets take either an API Gateway
execute-api ARN or an EventBridge API destination ARN, with HttpParameters; see
target-configuration.md.
Share the bus through AWS RAM where possible; the managed permission names are in
security-and-sharing.md. For a hand-written grant, the operation is
put-resource-policy, and the request names the bus through --resource-arn and the policy through
--policy-document:
aws eventbridgev2 put-resource-policy \
--resource-arn "$BUS_ARN" \
--policy-document '{
"Version": "2012-10-17",
"Statement": [{
"Sid": "AllowConsumerSubscribe",
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::444455556666:root"},
"Action": ["events:CreateSubscriber", "events:DescribeEventBus"],
"Resource": "'"$BUS_ARN"'"
}]
}'Omitting --policy-name writes the customer-managed default policy; the AWS_RAM policy is written
only by RAM. The principal must be a real account, and the actions use the events: namespace, never
eventbridgev2: (see authorization.md).
The DATA filter above matches PutEvents traffic, whose payload sits under detail (chapter 6). The
raw publish carries its payload at the top of Data, so it does not match that filter. The samples below
leave that mismatch in place, so both payload shapes stay visible. Data is base64, and the
base64 below decodes to {"orderId":"123"}:
aws eventbridgev2 put-events --event-bus-arn "$BUS_ARN" --entries '[{
"Source": "com.example.orders",
"DetailType": "OrderPlaced",
"Detail": "{\"orderId\":\"123\"}"
}]'
aws eventbridgev2 put-raw-events --event-bus-arn "$BUS_ARN" --entries '[{
"Data": "eyJvcmRlcklkIjoiMTIzIn0=",
"SystemMetadata": { "ContentType": "application/json" }
}]'The client name is eventbridgev2 in every language, and the class name is EventBridgeV2Client. The
publish call can succeed while individual entries fail,
so you MUST check every entry (chapter 21). Results align with the request entries by index, and
FailedEntryCount can be absent on a fully successful publish, so you SHOULD null-check it rather than
assume zero. SuccessCode: DEDUPLICATED is a success, not an error (chapter 5), so you MUST NOT count
successes as new events, because a deduplicated entry was not newly ingested.
Python:
import boto3, json
events = boto3.client("eventbridgev2")
response = events.put_events(
EventBusArn=bus_arn,
Entries=[{
"Source": "com.example.orders",
"DetailType": "OrderPlaced",
"Detail": json.dumps({"orderId": "123"}),
}],
)
failed = [entry for entry in response["Entries"] if entry.get("ErrorCode")]Java (AWS SDK for Java v2; the client is EventBridgeV2Client in
software.amazon.awssdk.services.eventbridgev2):
import java.util.List;
import software.amazon.awssdk.services.eventbridgev2.EventBridgeV2Client;
import software.amazon.awssdk.services.eventbridgev2.model.PutEventsRequest;
import software.amazon.awssdk.services.eventbridgev2.model.PutEventsRequestEntry;
import software.amazon.awssdk.services.eventbridgev2.model.PutEventsResponse;
import software.amazon.awssdk.services.eventbridgev2.model.PutEventsResultEntry;
EventBridgeV2Client events = EventBridgeV2Client.create();
PutEventsResponse response = events.putEvents(PutEventsRequest.builder()
.eventBusArn(busArn)
.entries(PutEventsRequestEntry.builder()
.source("com.example.orders")
.detailType("OrderPlaced")
.detail("{\"orderId\":\"123\"}")
.build())
.build());
List<PutEventsResultEntry> failed = response.entries().stream()
.filter(entry -> entry.errorCode() != null)
.toList();TypeScript (AWS SDK for JavaScript v3; the package name follows the service id, so the client ships as
@aws-sdk/client-eventbridgev2):
import { EventBridgeV2Client, PutEventsCommand } from "@aws-sdk/client-eventbridgev2";
const events = new EventBridgeV2Client({ region: "us-east-1" });
const response = await events.send(new PutEventsCommand({
EventBusArn: busArn,
Entries: [{
Source: "com.example.orders",
DetailType: "OrderPlaced",
Detail: JSON.stringify({ orderId: "123" }),
}],
}));
const failed = (response.Entries ?? []).filter((entry) => entry.ErrorCode);Data is a blob. An SDK takes the raw bytes and does the wire encoding itself; base64 is only the
CLI’s text form of a blob. You MUST NOT base64-encode bytes before handing them to an SDK, because they
are then delivered as base64 text rather than your payload, with no error anywhere. An Avro publish also
carries the schema-registry configuration at the request level (see
schema-registries.md).
Python:
response = events.put_raw_events(
EventBusArn=bus_arn,
Entries=[{
"Data": avro_bytes, # raw bytes; boto3 encodes the wire form
"SystemMetadata": {"ContentType": "application/avro"},
}],
SchemaRegistryConfiguration={"RegistryUri": glue_registry_arn},
)Java, where the raw-bytes form is SdkBytes.fromByteArray:
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.services.eventbridgev2.model.PutRawEventsRequest;
import software.amazon.awssdk.services.eventbridgev2.model.PutRawEventsRequestEntry;
import software.amazon.awssdk.services.eventbridgev2.model.PutRawEventsSystemMetadata;
import software.amazon.awssdk.services.eventbridgev2.model.SchemaRegistryConfiguration;
events.putRawEvents(PutRawEventsRequest.builder()
.eventBusArn(busArn)
.entries(PutRawEventsRequestEntry.builder()
.data(SdkBytes.fromByteArray(avroBytes))
.systemMetadata(PutRawEventsSystemMetadata.builder()
.contentType("application/avro")
.build())
.build())
.schemaRegistryConfiguration(SchemaRegistryConfiguration.builder()
.registryUri(glueRegistryArn)
.build())
.build());The same create as the CLI form above, plus the waiter. EventBusActive and EventBusDeleted are
modelled waiters, so every SDK generates them (chapter 15). In Python:
bus_arn = events.create_event_bus(Name="my-bus")["EventBusArn"]
events.get_waiter("event_bus_active").wait(EventBusArn=bus_arn)
subscriber_arn = events.create_subscriber(
Name="orders-to-queue",
EventBusArn=bus_arn,
FilterConfiguration={"Filters": [{
"Scope": "DATA",
"Pattern": '{"detail":{"orderId":[{"exists":true}]}}',
}]},
InvokeConfiguration={"TargetArn": queue_arn, "RoleArn": role_arn},
OnFailureConfiguration={"Arn": dlq_arn},
)["SubscriberArn"]You SHOULD read the subscriber back with describe_subscriber to confirm OnFailureConfiguration and
LogConfiguration stored what you sent, because the create response does not prove it (chapter 16).
Publishing is throttled per account across both publish APIs (chapter 23), and ThrottlingException
is retryable (chapter 21). The SDK’s built-in retries absorb short bursts; when they are exhausted, you
SHOULD retry with exponential backoff and jitter rather than immediately, because retrying immediately
keeps hitting the same limit:
import random, time
from botocore.exceptions import ClientError
def publish_with_backoff(entries, attempts=5):
delay = 0.2
for attempt in range(attempts):
try:
return events.put_events(EventBusArn=bus_arn, Entries=entries)
except ClientError as error:
if error.response["Error"]["Code"] != "ThrottlingException" or attempt == attempts - 1:
raise
time.sleep(delay + random.uniform(0, delay))
delay *= 2If you are steadily at the ceiling, raise the account quota instead of retrying harder (see cost-and-quotas.md).
You SHOULD read the queue and delete what you read, because otherwise counts lie to you. The full test-rig sequence is in delivery-troubleshooting.md.