Kafka to S3 for large messages with kaf-s3

To move large Kafka messages through S3, you can use kaf-s3-connector, a small open source (MIT) Python library at github.com/novatechflow/kaf-s3. Its producer sends small payloads to Kafka as they are and uploads anything above a size limit (900,000 bytes by default) to an S3 bucket, then writes a short JSON reference with the bucket, key and a SHA-256 to the topic. Its consumer spots those references, downloads the object under a size cap, checks the hash and hands you the original bytes. It works with AWS S3 and S3-compatible storage such as MinIO, and ships with a dead letter topic, Prometheus metrics, a Docker image and a Helm chart. This post shows a local quick start, explains how cleanup and integrity checks work, and says plainly when you need a Kafka to S3 sink instead.

---

Sooner or later someone asks to put a PDF, a model output, a big JSON export or a batch of sensor readings on a Kafka topic. Kafka brokers and producers accept about 1 MB per message out of the box. You can raise message.max.bytes, max.request.size and the consumer fetch sizes, but then every broker, every replica and every consumer has to carry those big records, and you find out about the one client you forgot in production.

The pattern I use instead is old and boring, which is why it works. Put the payload in object storage, put a small pointer on the topic, and let the consumer fetch the payload when it needs it. People call it the claim check pattern. I packaged my version as kaf-s3 after building it for a client. The background of that project is in the Kafka to S3 connector case study. This post is the hands-on part.

Code: github.com/novatechflow/kaf-s3. Open source under MIT, Python 3.10 or newer, built on confluent-kafka and boto3. On PyPI as kaf-s3-connector, container image at ghcr.io/novatechflow/kaf-s3-connector.

What kaf-s3 does and what it does not do

This is worth getting right before you install anything, because "Kafka to S3" means two different things.

• kaf-s3 is for large messages. It sits in your producer and consumer code. Big payloads go to S3, Kafka carries a reference, and your consumer gets the original bytes back. Small payloads stay on Kafka as they are.
• kaf-s3 is not a sink connector. It does not copy a whole topic into S3 files for analytics. For that you want a Kafka Connect S3 sink, or a platform that keeps the log in S3 to begin with. I wrote about that second option in Kafka on object storage was inevitable, and KafScale is the open source platform we built for it.
• It is not a general S3 to Kafka importer either. The consumer only resolves references that a kaf-s3 producer wrote. If you need to stream existing files from a bucket into Kafka, that is a source connector job.

How it works

1. You call produce() with bytes. At or under max_inline_bytes (900,000 by default, which leaves headroom under the 1 MB default), the payload goes to Kafka unchanged.
2. Above that, the payload is uploaded to your bucket under a random key, optionally gzipped and encrypted with SSE-KMS. Above 8 MiB it switches to a multipart upload, and the hard cap is 5 GiB.
3. A small JSON reference goes to the topic with s3_bucket, s3_key, the ETag where it is usable, a SHA-256 of the original payload and the compression used. If the Kafka publish fails, the S3 object is removed again.
4. The consumer treats a message as a reference only if it is a JSON object with both s3_bucket and s3_key. Everything else passes through as a normal payload.
5. It checks that the reference points at the bucket and prefix you configured, downloads under a size cap, decompresses under the same cap and verifies the SHA-256 before it returns anything. A reference with no checksum is rejected by default.

A reference on the topic looks like this:

{"s3_bucket": "large-messages", "s3_key": "3f2a...", "etag": "\"9b1c...\"", "sha256": "e3b0...", "compression": null}

Quick start with Kafka and MinIO on your laptop

Start a single Kafka broker and a MinIO server with Docker:

docker run -d --name kafka -p 9092:9092 apache/kafka:latest
docker run -d --name minio -p 9000:9000 \
  -e MINIO_ROOT_USER=minioadmin -e MINIO_ROOT_PASSWORD=minioadmin \
  minio/minio server /data

Install the library. Credentials come from the normal boto3 chain, so for MinIO environment variables are enough:

pip install kaf-s3-connector

export AWS_ACCESS_KEY_ID=minioadmin
export AWS_SECRET_ACCESS_KEY=minioadmin
export AWS_DEFAULT_REGION=us-east-1

python3 -c "import boto3; boto3.client('s3', endpoint_url='http://localhost:9000').create_bucket(Bucket='large-messages')"

Now send a 5 MB payload and read it back:

import os
from s3_connector import S3Producer, S3Consumer

s3 = {"bucket": "large-messages", "endpoint_url": "http://localhost:9000"}

producer_config = {"kafka": {"bootstrap.servers": "localhost:9092"}, "s3": s3}
consumer_config = {
    "kafka": {
        "bootstrap.servers": "localhost:9092",
        "group.id": "large-readers",
        "auto.offset.reset": "earliest",
        "enable.auto.commit": False,
    },
    "s3": s3,
}

payload = os.urandom(5 * 1024 * 1024)

# closing the producer flushes the queue
with S3Producer(config=producer_config) as producer:
    producer.produce(topic="large-messages", payload=payload)

consumer = S3Consumer(config=consumer_config)
consumer.subscribe(["large-messages"])
while True:
    data = consumer.poll(timeout=10.0)
    if data:
        print("received", len(data), "bytes, same as sent:", data == payload)
        consumer.commit()
        break
consumer.close()

On AWS you drop endpoint_url and the environment variables and give the producer and consumer an IAM role instead. Do not put AWS keys in the config. The library is built around the default credential chain on purpose.

Things that bite in production

Cleaning up S3 objects

The objects outlive the Kafka records that point to them, so something has to delete them. Use an S3 lifecycle rule. Set ttl_seconds on the producer, every object gets a kaf-s3-ttl-seconds tag, and the rule in config/s3-lifecycle.json expires it. Make the expiry longer than the topic retention. If you do not, a consumer that falls behind or replays from an old offset finds its objects gone and the message is skipped as object_missing.

There is also delete_after_consume. I keep it in the library, but I tell people not to use it. Kafka is a fan-out log. The moment a second consumer group reads the topic, the first group deletes objects the second one still needs, and no consumer can see that other groups exist. It only works with one consumer group and manual commits, and the connector refuses to start with it under auto-commit.

Where this came from. kaf-s3 grew out of a client project where big payloads kept landing on Kafka.

• The payloads. Documents, JSON blobs, binary exports and telemetry batches.
• The size. Far past the 1 MB default. kaf-s3 caps a single payload at 5 GB by default.
• Why not raise the broker limits. Kafka is built to stream lots of small messages fast. Pushing large messages through the brokers slows everything down and makes the cluster unstable.
• The fix. Large payloads go straight to an S3 bucket or any S3 compatible store.
• The Kafka message. Kafka only carries a small JSON reference with the bucket, the key and a checksum.
• What got better. The brokers never see the big payloads anymore, so the cluster stays stable.
• A bonus. The data already sits in S3, so lakehouse tools like Apache Iceberg, Spark or Flink can pick it up right away.

Delivery and commits

• Turn off auto-commit and commit after processing. That gives you at-least-once delivery. With auto-commit an offset can move before you have handled the payload.
• produce() is asynchronous. Call flush() or use the context manager, or queued messages are lost when the process exits.
• poll() returns None in two cases. No message arrived, or a message was skipped as unusable. Register the skipped hook if you need to tell them apart.
• One instance per thread. Producers and consumers are not safe to share across threads.

Bad references and the dead letter topic

Set dlq_topic and failures go there instead of disappearing. Integrity problems get fixed reason codes such as sha256_mismatch, prefix_violation, unexpected_bucket or object_too_large. The bucket and prefix checks matter more than they look. Without them, anyone who can write to the topic can make your consumer fetch any object its IAM role can read.

Duplicates and deterministic keys

With deterministic_keys the S3 key is a hash of the payload, so a retried send writes the same object instead of a second one. Kafka can still carry two references to it. If duplicates matter, deduplicate on a business ID in the consumer. My post on finding duplicate Kafka messages shows how to measure that.

Run it as a container or on Kubernetes

The image runs as a producer that reads stdin or as a consumer, and it exposes Prometheus metrics on port 8000:

docker run --rm -p 8000:8000 \
  -e MODE=consumer \
  -e TOPIC=large-messages \
  -e KAFKA_BOOTSTRAP_SERVERS=broker:9092 \
  -e KAFKA_GROUP_ID=large-readers \
  -e S3_BUCKET=large-messages \
  ghcr.io/novatechflow/kaf-s3-connector:latest

The Helm chart is in charts/kaf-s3-connector. Put Kafka SASL passwords in a Kubernetes Secret and reference it with envFrom. Values in env end up in the Deployment in plain text. The metrics endpoint has no auth, so keep it on an internal network.

When to use what

You need	Use
Some messages are bigger than 1 MB	kaf-s3 in producer and consumer
A copy of every topic as files in S3	A Kafka Connect S3 sink
Topics feeding Iceberg tables	A sink or Flink job, see my lakehouse ingest guide
All Kafka data in S3, no broker disks	KafScale
Topics copied between clusters	kaf-mirror

Frequently asked questions

How do I send messages larger than 1 MB through Kafka?

Either raise the size limits on brokers, producers and consumers, or store the payload in object storage and send a reference. kaf-s3 does the second one for you in Python.

Does kaf-s3 work with MinIO and other S3-compatible storage?

Yes. Set endpoint_url in the S3 config, or S3_ENDPOINT_URL for the container.

Is kaf-s3 a Kafka Connect connector?

No. It is a Python library and a small runtime. It does not run inside Kafka Connect and it does not write topics to S3 as files.

How does the consumer know a message is a reference?

A message is a reference only if it is a JSON object with both s3_bucket and s3_key. Everything else is returned as it is.

What happens if the S3 object was deleted?

The consumer skips the message with the reason object_missing and sends it to the dead letter topic if you set one. Other S3 errors, like access denied, are raised to your code.

Source code

The library is on GitHub at github.com/novatechflow/kaf-s3, with unit tests that need nothing running and an integration suite that runs against a real Kafka broker and real S3 in throwaway containers. Issues and pull requests are welcome.

If you are planning a Kafka to S3 pipeline, for large messages, a data lake or both, and want someone who has built it before, that is the kind of work I do in my Kafka consulting.