S3 sink

Collects records into files and uploads each file as one object to Amazon S3 or an S3-compatible store. Records are buffered in memory per topic and partition, and a file is closed and uploaded when it reaches max_file_size or max_messages_per_file. The object key carries the partition and the first and last offset in the file, so a listing sorts in offset order.

The sink confirms a batch as soon as it is buffered, not when it is uploaded. The consequences for a crash are spelled out under How it works and Replay.

Type	Sink
Library	libpicomq_connector_s3_sink
Ships in	The pico-connectors image
Destination	Object key, templated per topic and partition
Creates destination	No. The bucket must exist
On replay	A second object with an overlapping offset range is possible
Payload	Any schema. Written as JSON lines, a JSON array, or raw bytes

Quick start

type = "sink"
key = "orders_s3"
enabled = true
version = 0
name = "Orders to S3"
path = "libpicomq_connector_s3_sink"

[[topics]]
pattern = 'orders\..*'
schema = "json"
batch_length = 1000
poll_interval = "100ms"

[plugin_config]
bucket = "picomq-archive"
region = "eu-west-1"
prefix = "raw"
path_template = "{topic}/{date}/{hour}"
max_file_size = "64MiB"
output_format = "json_lines"
access_key_id = "AKIA..."
secret_access_key = "..."

Keep the secret out of the file with an environment override. Omit both keys to use the default credential chain instead, which reads the standard AWS environment variables, the shared profile, or the instance metadata service.

PICOMQ_CONNECTORS_SINK_ORDERS_S3_PLUGIN_CONFIG_SECRET_ACCESS_KEY=...

How it works

On open() the sink validates the configuration, builds the client, and checks connectivity by writing an empty object at .picomq-sink-probe in the bucket and deleting it again. A failed probe fails open().

For each batch the runtime hands over, the sink does the following.

1. Finds or creates the in-memory buffer for this topic and partition.
2. Formats each record. json_lines and json_array produce one JSON object per record, raw takes the payload bytes as they are.
3. Appends the record to the buffer and checks rotation. With file_rotation = "size" the file rotates once the buffer holds at least max_file_size bytes, with "messages" once it holds max_messages_per_file records.
4. On rotation, finalises the file, renders the object key from path_template, resets the buffer, and uploads the file.
5. Returns success once every record is buffered. Nothing is uploaded until a rotation happens or the sink is closed.

The upload is attempted up to max_attempts times. The delay before attempt n is retry_delay doubled n - 1 times, with 20 percent jitter and a cap of 60 seconds. Transport errors and HTTP 5xx, 408 and 429 responses are retried. Any other non-2xx status fails the upload at once.

When the upload fails for good, the file is gone from memory. The sink writes a marker object at <key>.lost containing the offset range, the record count and the error, counts the records as lost, and returns the error. The runtime redelivers the current batch, but records from earlier batches that were in the same file were already committed and are not redelivered.

Buffered records are confirmed before they are uploaded

The runtime commits the offset when consume() returns, and consume() returns after buffering. A crash of the runtime loses everything buffered and not yet uploaded, up to one file per topic and partition. A graceful stop flushes every buffer in close(). Size the rotation threshold to the loss you can accept, or read from a source that can be re-run.

Configuration

All keys go under [plugin_config].

Key	Type	Default	Meaning
bucket	string	required	Bucket name. Must already exist
region	string	required	AWS region name. With endpoint set, any label is accepted and passed through
prefix	string	none	Key prefix in front of the rendered template. Surrounding slashes are trimmed
endpoint	string	none	Custom endpoint URL for MinIO, Ceph, R2 and similar. Turns on path-style addressing
access_key_id	string	none	Static credential. Redacted in the API. Set both keys or neither
secret_access_key	string	none	Static credential. Redacted in the API
path_template	template	{topic}/{date}/{hour}	Directory part of the object key, see Object keys
file_rotation	string	size	size or messages
max_file_size	size	8MiB	Rotation threshold for size. Binary and decimal units both parse, 8MiB and 10MB. Must be above 0 and at most 5GiB
max_messages_per_file	int	none	Rotation threshold for messages, required in that mode. Accepted but ignored under size
output_format	string	json_lines	json_lines (also jsonl, jsonlines), json_array or raw
include_metadata	bool	true	Add offset, timestamp, topic, partition and key to each JSON object
include_headers	bool	false	Add headers to each JSON object
max_attempts	int	3	Total PutObject attempts per file. max_retries is accepted as an alias
retry_delay	duration	1s	Base delay between attempts
path_style	bool	true when endpoint is set	Force host/bucket/key instead of bucket.host/key

path_template accepts more placeholders than the shared destination template syntax, since it names a directory rather than a table. The extra placeholders are listed under Object keys.

What lands in the bucket

With json_lines, each record is one JSON object on its own line.

{"offset":1042,"timestamp":"2026-09-03T21:15:04.118Z","topic":"orders.eu","partition":0,"key":"dXNlci0x","payload":{"id":7,"total":42.5}}

Field	Present when	Content
offset	include_metadata	Record offset
timestamp	include_metadata	Record timestamp, RFC 3339 with milliseconds, UTC
topic	include_metadata	Topic the record came from
partition	include_metadata	Always 0 on PicoMQ
key	include_metadata and the record has a key	Record key, base64
headers	include_headers and the record has headers	Object of header name to value. UTF-8 values are strings, other values base64
payload	always	The record, converted as below

Topic schema	payload in the JSON object
json	The parsed document
text, proto	A string
raw	The parsed document if the bytes are valid JSON, otherwise a base64 string
flatbuffer, avro	A base64 string

json_array wraps the same objects in [ and ] separated by commas, with no newline. raw concatenates the payload bytes of every record with no delimiter, which is only useful for payloads that carry their own framing.

Object keys

The key is <prefix>/<rendered path_template>/<partition>-<first offset>-<last offset>.<ext>. The partition is zero-padded to 5 digits and the offsets to 20, so lexical order equals offset order. The extension is jsonl, json or bin by output_format.

raw/orders.eu/2026-09-03/21/00000-00000000000000001000-00000000000000001999.jsonl

Placeholder	Resolves to
{topic}	The topic name, sanitised
{topic_segment[n]}	Segment n of the topic split on ., negative counts from the end, sanitised
{partition}	The partition number, unpadded
{date}	YYYY-MM-DD in UTC of the first record in the file
{hour}	HH in UTC of the first record in the file
{timestamp}	Millisecond timestamp of the first record in the file

Sanitising keeps ASCII letters, digits, ., _ and - and replaces every other character with _. A topic with fewer segments than the template asks for fails the file when it is about to be uploaded, and that error is returned to the runtime.

Replay

The runtime redelivers a batch after a crash between the write and the offset commit. See Delivery guarantees. For this sink the write is the buffer append, so the window is the same as for any other sink, but the file itself may or may not have been uploaded yet.

Situation	Result
Crash with records still in the buffer	Those records are lost. They were confirmed and are not redelivered
Redelivered batch, the file it went into is still in memory	Records appear twice in the next file
Redelivered batch, the file it went into was already uploaded	A second object is written whose offset range overlaps the first
Upload failed after max_attempts	The file is dropped, a <key>.lost marker is written, the current batch is redelivered

Downstream readers can deduplicate on topic, partition and offset when include_metadata is on. Overlapping offset ranges in the key names make a suspect object easy to find.

Requirements

• An existing bucket. The sink does not create it.
• Credentials with s3:PutObject on the bucket, and s3:DeleteObject for the probe object written at start.
• For non-AWS stores, endpoint set to the service URL. Path-style addressing is turned on automatically.
• Memory for one open file per topic and partition, up to max_file_size each.

Troubleshooting

Symptom	Cause
Partially configured credentials at start	Only one of access_key_id and secret_access_key is set
S3 bucket '...' connectivity check failed at start	Wrong bucket, region or endpoint, or the credentials lack PutObject
Invalid S3 region '...' at start	region is not an AWS region name and no endpoint is set
file_rotation is 'messages' but max_messages_per_file is not configured	Add max_messages_per_file or switch to size
max_file_size (...) exceeds S3 single PutObject limit of 5 GiB	Lower max_file_size. The sink uses a single PutObject per file
S3 PutObject returned non-retriable status 403	Credentials lack write permission on the key prefix, or the bucket policy denies it
.lost objects in the bucket	An upload exhausted its attempts. The marker body has the offset range to recover from a source replay
Nothing appears in the bucket	No file has rotated yet. Lower max_file_size or use messages rotation for low-volume topics