---
title: "Publish S3 bucket events to a topic"
slug: "dev-event-broker-guides-s3-bucket-events"
description: "Set up S3 bucket notifications to publish events to a VAST Event Broker topic, enabling seamless data consumption with Kafka. Learn how to configure and manage."
updated: 2026-10-28T02:14:59Z
published: 2026-10-28T02:14:59Z
canonical: "docs.vastdata.com/dev-event-broker-guides-s3-bucket-events"
---

> ## Documentation Index
> Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Publish S3 bucket events to a topic

*Applies to VAST Cluster 5.5.*

VAST Cluster can publish an event to a VAST Event Broker topic each time an object in an S3 bucket is created, tagged or deleted. Your application then reacts to new data with an ordinary Kafka consumer, instead of listing the bucket. This guide sets up a bucket notification with the S3 API, consumes the events, and explains their format and delivery guarantees.

Run the code blocks in order in one Python session. Below each block is the output you should see.

## How it works

1. You put a notification configuration on an S3 bucket: which events, which objects (by key prefix or suffix), and which topic.
2. When an S3 request on the bucket succeeds and matches a rule, the cluster queues an event record. Failed requests create no events ([Event Publishing](/docs-5-5/docs/event-publishing)).
3. In the background, the cluster writes the record into the topic, as JSON.
4. Your consumers read it like any other Kafka record. The topic is also a table you can [query](/develop/docs/dev-event-broker-guides-querying-topics).

The S3 request doesn't wait for the event: publishing is asynchronous, so expect a short delay between the request and the event.

![An S3 client writes, deletes or tags an object in a bucket that has a notification configuration: which events, which objects by key prefix or suffix, and which topic. The cluster publishes a JSON record with a Records list to the topic, asynchronously and at least once. Kafka consumers read the records, and the topic can also be queried as a table.](https://cdn.us.document360.io/89b91113-5dc6-49db-86ec-57b16dc8ce56/Images/Documentation/s3-events-flow.svg?sv=2026-02-06&spr=https&st=2026-11-08T08%3A56%3A13Z&se=2026-11-08T09%3A56%3A13Z&sr=c&sp=r&sig=CnFBB5V39ZC1HZD%2B4REhtxzggJM5yoLAihZgkHtwN%2BU%3D)

*A notification rule on the bucket publishes S3 events into a topic, asynchronously and at least once.*

## Before you start

You need:

- A Kafka view and a topic on it. The topic must exist before events are published to it: the notification configuration doesn't create it, and the cluster doesn't check that it exists when you save the configuration.  This guide creates its topic with a Kafka client.
- An S3 bucket to watch, on a view with the S3 Bucket protocol and the *S3 Native* security flavor, in the same tenant as the Kafka view ([Configuring S3 Event Notifications for a View](/docs-5-5/docs/publishing-s3-bucket-events-to-third-party-event-brokers#configuring-s3-event-notifications-for-a-view)).
- S3 credentials for a user that can change the bucket's notification configuration and write objects to it.
- The environment variables and clients from [Query topics as tables](/develop/docs/dev-event-broker-guides-querying-topics#before-you-start), and `boto3`. Set `SOURCE_BUCKET` to the bucket to watch:

```
export SOURCE_BUCKET="<bucket to watch>"
pip install boto3
```

Your platform team can also set up the notification for you, in the VAST Web UI (the view's **Event Notifications** tab) or with the VAST CLI. **Admin task: send to your platform team.** See [Publishing S3 Bucket Events to VAST Event Broker](/docs-5-5/docs/publishing-s3-bucket-events-to-vast-event-broker). S3 event notifications aren't supported on Global Access destination buckets ([5.5.0 release notes](/docs-5-5/docs/vast-cluster-5-5-0-release-notes)).

## Set up

Create the topic, and an S3 client:

```
import json
import os
import time
from urllib.parse import unquote_plus

import boto3
from confluent_kafka import Consumer, KafkaException
from confluent_kafka.admin import AdminClient, NewTopic

EVENTS_BUCKET = os.environ["EVENTS_BUCKET"]
SOURCE_BUCKET = os.environ["SOURCE_BUCKET"]
TOPIC = "docs-s3-bucket-events-objects"
GROUP = "docs-s3-bucket-events-reader"
PREFIX = "docs-s3-bucket-events/"

def kafka_config():
    cfg = {"bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"]}
    if os.environ.get("KAFKA_USERNAME"):
        cfg.update({
            "security.protocol": "SASL_PLAINTEXT",
            "sasl.mechanism": "PLAIN",
            "sasl.username": os.environ["KAFKA_USERNAME"],
            "sasl.password": os.environ["KAFKA_PASSWORD"],
        })
    return cfg

admin = AdminClient(kafka_config())
admin.create_topics([NewTopic(TOPIC, num_partitions=3, replication_factor=1)])[TOPIC].result()

s3 = boto3.client("s3", endpoint_url=os.environ["AWS_S3_ENDPOINT_URL"])
print("ready")
```

```
ready
```

## Configure the notification

Use the standard S3 `PutBucketNotificationConfiguration` call, with a topic configuration. Write the topic as:

```
vast-event-broker:<bucket of the Kafka view>:<topic>
```

`vast-event-broker` is a fixed word that means the cluster's own broker. For a third-party Kafka broker that your platform team has added to the cluster, the format is `&lt;broker name&gt;:&lt;topic&gt;` instead.

```
s3.put_bucket_notification_configuration(
    Bucket=SOURCE_BUCKET,
    NotificationConfiguration={
        "TopicConfigurations": [{
            "Id": "docs-s3-bucket-events",
            "TopicArn": f"vast-event-broker:{EVENTS_BUCKET}:{TOPIC}",
            "Events": ["ObjectCreated:*", "ObjectRemoved:*"],
            "Filter": {"Key": {"FilterRules": [{"Name": "prefix", "Value": PREFIX}]}},
        }],
    },
)
print("notification configured")
```

```
notification configured
```

The rules:

- **Event names** don't have the `s3:` prefix that AWS uses:

| Events | Names |
| --- | --- |
| Object creation | `ObjectCreated:Put`, `ObjectCreated:Post`, `ObjectCreated:Copy`, `ObjectCreated:CompleteMultipartUpload`, or `ObjectCreated:*` for all |
| Object tagging | `ObjectTagging:Put`, `ObjectTagging:Delete`, or `ObjectTagging:*` |
| Object deletion | `ObjectRemoved:Delete`, `ObjectRemoved:DeleteMarkerCreated`, or `ObjectRemoved:*` |

Other names, including the AWS form `s3:ObjectCreated:*`, are rejected with `InvalidArgument`. Don't list a wildcard and one of its own events in the same rule, such as `ObjectCreated:*` and `ObjectCreated:Put`.
- **Filters** take a key prefix, a suffix or both. Write the rule names in lowercase, `prefix` and `suffix`.
- **Rules can't overlap.** Two rules can't both match the same request, for example one for keys starting `abc` and one for keys starting `abcdef` ([Configuring S3 Event Notifications for a View](/docs-5-5/docs/publishing-s3-bucket-events-to-third-party-event-brokers#configuring-s3-event-notifications-for-a-view)).
- **The Kafka view's bucket** must be in the same tenant as the source bucket. Otherwise the call fails with `kafka bucket name not found`.
- **The call replaces** the bucket's whole notification configuration, as in AWS S3. To change one rule, send all of them.

## Make some events

Upload two objects and delete one:

```
s3.put_object(Bucket=SOURCE_BUCKET, Key=PREFIX + "a.json", Body=b'{"n": 1}')
s3.put_object(Bucket=SOURCE_BUCKET, Key=PREFIX + "b.json", Body=b'{"n": 2}')
s3.delete_object(Bucket=SOURCE_BUCKET, Key=PREFIX + "a.json")
print("done")
```

```
done
```

## Consume the events

Each Kafka record's value is a JSON document with a `Records` list. Poll until the three events have arrived, or a timeout passes:

```
consumer = Consumer({**kafka_config(), "group.id": GROUP,
                     "auto.offset.reset": "earliest", "enable.auto.commit": False})
consumer.subscribe([TOPIC])

events = []
deadline = time.time() + 120
while len(events) < 3 and time.time() < deadline:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        raise KafkaException(msg.error())
    body = json.loads(msg.value())
    for record in body.get("Records", []):      # skip anything else, such as test events
        events.append((msg.key(), record))
    consumer.commit(message=msg, asynchronous=False)

for kafka_key, record in sorted(events, key=lambda e: e[1]["s3"]["object"]["key"]):
    name = record["eventName"].removeprefix("s3:")
    obj = record["s3"]["object"]
    print(name, record["s3"]["bucket"]["name"], unquote_plus(obj["key"]), obj.get("size"), kafka_key)
```

```
ObjectCreated:Put docs-s3-source docs-s3-bucket-events/a.json 8 b'docs-s3-source/docs-s3-bucket-events/a.json'
ObjectRemoved:Delete docs-s3-source docs-s3-bucket-events/a.json 8 b'docs-s3-source/docs-s3-bucket-events/a.json'
ObjectCreated:Put docs-s3-source docs-s3-bucket-events/b.json 8 b'docs-s3-source/docs-s3-bucket-events/b.json'
```

The sort is stable, so the events for each object stay in the order they arrived.

## Event format

The record follows the AWS S3 event structure ([S3 Event Record Format](/docs-5-5/docs/publishing-s3-bucket-events-to-third-party-event-brokers#s3-event-record-format)):

```
{"Records": [{
  "eventVersion": "2.2",
  "eventSource": "vast:s3",
  "awsRegion": "<region>",
  "eventTime": "<timestamp>",
  "eventName": "<event type>",
  "userIdentity": {"principalId": "<principal ID>"},
  "requestParameters": {"sourceIPAddress": "<source IP>"},
  "responseElements": {"x-amz-request-id": "<request ID>", "x-amz-id-2": "<request ID>"},
  "s3": {
    "s3SchemaVersion": "1.0",
    "configurationId": "<rule Id>",
    "bucket": {"name": "<bucket name>", "ownerIdentity": {"principalId": "<principal ID>"}, "arn": "<bucket ARN>"},
    "object": {"key": "<key>", "size": <size>, "eTag": "<eTag>", "versionId": "<version>", "sequencer": "<sequencer>"}
  }
}]}
```

What to rely on:

| Field | Notes |
| --- | --- |
| `eventName` | The event, such as `ObjectCreated:Put`. Strip an `s3:` prefix if it's there, as the example does. |
| `eventTime` | When the cluster finished processing the request, in ISO-8601 |
| `userIdentity.principalId` | The user name for local users, `username@domainname` for users from an authentication provider |
| `s3.configurationId` | The `Id` of the rule that matched |
| `s3.bucket.name` | The source bucket. Use this rather than `arn`. |
| `s3.object.key` | The object key, URL-encoded as in AWS. Decode it with `unquote_plus()`. |
| `s3.object.size` | Bytes. Not present for `ObjectRemoved:DeleteMarkerCreated`. |
| `s3.object.versionId` | Only when the bucket has versioning |
| `s3.object.sequencer` | Only for PUT and DELETE requests |
| `awsRegion` | The VAST cluster's name |

The Kafka record's key is `&lt;bucket&gt;/&lt;object key&gt;`, and all events for one object go to the same partition, so they arrive in order for that object. Events for different objects can arrive in any order.

The cluster sends a test event (`&quot;Event&quot;: &quot;s3:TestEvent&quot;`) when you save a notification definition. It has no `Records` list, so skip records without one, as the example does.

## Delivery: at least once

The cluster retries publishing after a failure, so an event can be delivered more than once. Make your consumer idempotent:

- Deduplicate on the request ID (`responseElements[&quot;x-amz-request-id&quot;]`) plus `eventName` and the object key, or
- Make the processing safe to repeat, for example by writing results keyed by bucket, object key and `eTag`.

Don't use the topic as an audit log of every request. In rare failures, such as the cluster being unable to read the notification configuration, the S3 request still succeeds, and the event is dropped with an alert to your platform team.

The consumer side works like any other topic: commit offsets after you've processed a record, as the example does, so a restart resumes where you left off. See [Consuming](/develop/docs/dev-event-broker-guides-consuming).

## Clean up

Remove the notification rules, delete the test object, and delete the consumer group and topic:

```
s3.put_bucket_notification_configuration(Bucket=SOURCE_BUCKET, NotificationConfiguration={})
s3.delete_object(Bucket=SOURCE_BUCKET, Key=PREFIX + "b.json")
consumer.close()
admin.delete_consumer_groups([GROUP])[GROUP].result()
admin.delete_topics([TOPIC])[TOPIC].result()
print("cleaned up")
```

```
cleaned up
```

Remove the notification rules before you delete their topic, so the cluster doesn't try to publish to a topic that no longer exists.

## See also

- [Query topics as tables](/develop/docs/dev-event-broker-guides-querying-topics): read bucket events with SQL, such as all objects created in the last hour.
- [Consuming](/develop/docs/dev-event-broker-guides-consuming): consumer groups and offset commits.
- [Topics](/develop/docs/dev-event-broker-guides-topics): create topics and set their retention.
- [Event Publishing](/docs-5-5/docs/event-publishing) in the KB, including third-party brokers.