S3 to Kafka: How to Turn New S3 Files into Kafka Messages

Files that land in Amazon S3 reach Kafka topics in one of four ways: a Kafka Connect S3 source connector, S3 Event Notifications that trigger an AWS Lambda function, a Spark or Flink streaming job, or a managed pipeline. Each finds new objects under a prefix, parses every file into records and produces one message per record.

Kafka Connect suits teams already running Connect or Confluent Cloud, Lambda a modest number of uploads needing a quick reaction, Spark or Flink files that need transforming first, and a managed pipeline teams that don't want to run connectors. What matters most: how new files are found (key order, modification time or events), which formats are parsed, and how consumers handle duplicates, since most options deliver at least once. Archiving topics to S3 is the reverse job, for an S3 sink connector.

Estuary is one managed option for this route: its S3 capture lists a bucket prefix on an interval and parses new files into a collection, and its Kafka materialization produces each record to a topic in your cluster.

Choosing an Amazon S3 to Kafka method

Four of the five approaches find files by listing the bucket, so most of this guide is about what listing-based tools get right and wrong.

How S3 files become Kafka messages

Every method runs five stages and differs mostly in the first two.

• Discovery. Call ListObjectsV2 on a schedule, or receive one S3 Event Notification per new object. A general purpose bucket lists up to 1,000 keys per call in lexicographic key order; directory buckets don't sort.
• Tracking. Record what was read so a restart doesn't start over: the last key (cheap, but a late file with an earlier-sorting key is never seen), each file's modification time, or every processed path (Flink's file source, whose state grows with the file count).
• Parsing. Split each file into records: JSON lines, CSV rows, text lines, Avro or Parquet records.
• Producing. Write each record as a message with a key and a serializer.
• Consuming. Consumers must tolerate the duplicates that restarts and retried notifications create.

Amazon S3 requirements

The reading process needs s3:ListBucket on the bucket, which you can limit to the prefix with an s3:prefix condition, and s3:GetObject on prefix/*. Objects encrypted with an AWS KMS key also need kms:Decrypt on that key.

Where the files live changes what a listing returns:

Three layout rules prevent most problems:

1. One file type per prefix. Most tools take one format per connector.
2. Keys that sort in arrival order, such as logs/2026/10/21/T101500Z-host1.json.
3. Write each file once, under a new key, rather than overwriting.

Event-driven pickup sends s3:ObjectCreated:* notifications to SQS, SNS, Lambda or EventBridge; SQS FIFO queues aren't supported.

Kafka requirements

• Topics created with the partitions and retention you need, or permission for the producer to create them.
• Producer credentials that can write to those topics, usually SASL/SCRAM or SASL/PLAIN over TLS. Amazon MSK serves IAM authentication on ports 9098 and 9198 and SASL/SCRAM on 9096 and 9196.
• A Schema Registry for Avro, Protobuf or JSON Schema. Confluent's AvroConverter fails without schema.registry.url.
• A message size budget. The broker's message.max.bytes defaults to 1,048,588 bytes per record batch, and the producer's max.request.size to 1,048,576 bytes.

Setting up an Amazon S3 to Kafka pipeline with a Kafka Connect source connector

1. Install the connector plugin on your Connect workers, upload it to MSK Connect as a custom plugin, or create the fully managed connector in Confluent Cloud.
2. Create the target topics with the partitions and retention you need.
3. Create an IAM role or user with the S3 permissions above.
4. Write the connector config. Confluent's connector defaults to backup-and-restore mode, which reads only <topic>+<partition>+<offset>.<extension> files from its own S3 sink, so set mode to GENERIC. It still ignores objects outside topics.dir (default topics); point that at your folder and map paths to topics with topic.regex.list. Aiven's connector needs file.name.template.
5. Choose the discovery order. Lenses can order by LastModified instead of key.
6. Decide what a bad record does.errors.tolerance=all skips records that fail conversion and errors.log.enable=true logs them. Kafka Connect's dead letter queue is for sink connectors only.
7. Validate. Drop a test file and consume it. Restart the connector and confirm old files aren't re-sent. Add a file whose key sorts before the last one read, and see whether it arrives.

A generalized-mode config, submitted to the Connect REST API:

json
{
  "name": "s3-orders-source",
  "config": {
    "connector.class": "io.confluent.connect.s3.source.S3SourceConnector",
    "mode": "GENERIC",
    "s3.bucket.name": "example-landing-bucket",
    "s3.region": "us-east-1",
    "topics.dir": "orders",
    "topic.regex.list": "orders-raw:.*\\.json",
    "format.class": "io.confluent.connect.s3.format.json.JsonFormat",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "tasks.max": "2"
  }
}

What happens after the initial load?

• Existing files. A listing tool's first pass sees every file under the prefix, so a large history becomes a large Kafka backlog.
• New files arrive on the next poll or notification.
• Late keys. A tool that remembers only the last key skips a file whose key sorts before it.
• Overwritten files. Confluent's generalized mode doesn't reload a renamed file or a newer upload of one it has read. Tools that track modification time re-read the whole file and produce every record again.
• Deleted files. Deleting an object doesn't remove its messages from Kafka.
• Duplicates. Restarts and retried notifications re-send records; deduplicate on file path plus record position.
• Schema changes. A new field must pass the registry subject's compatibility rule.

Amazon S3 file formats and Kafka message formats

Each tool parses a different set of formats:

• One file, many messages: one per line or CSV row, or one per Avro or Parquet record.
• Parquet is often read whole. Redpanda Connect's decoder reads each file into memory, so file size becomes a memory limit.
• Compression. Confluent's backup-and-restore mode can't read compressed files; Aiven reads gzip, snappy and zstd.

Choosing the message key

The producer's default partitioner hashes the key, so messages with the same key stay in one partition, in order. A file-and-offset key is unique per record: it spreads a file across partitions and gives consumers an identity for deduplication. A device or customer ID keeps each entity's records in order instead.

Performance, latency and S3 request cost

Freshness is discovery delay plus parse and produce time: up to one poll interval for listing tools, usually seconds for notifications. On the S3 side you pay for GETs and for LIST requests, which AWS bills at the PUT, COPY and POST rate.

• Listing grows with the prefix. At 1,000 keys per call, re-listing a million-object prefix takes 1,000 requests per pass. Narrow date-partitioned prefixes, listing from the last key, and expiring files already in Kafka keep passes short.
• Ordering by modification time costs a full listing. Lenses lists and sorts every object in that mode.
• Parallelism. Lenses reads each data partition with one task; Confluent's generalized mode spreads files over tasks.max tasks.

Production considerations

• Retries and recovery. Know where the tool keeps its position. Confluent's connector stores offsets under the connector name, so recreating it under the same name doesn't re-read the bucket.
• Schema changes. Set the registry's compatibility rule before later files add fields.
• Backfills. Re-reading a prefix re-produces every message in it.
• Monitoring. Alert on time since the last message per topic and on consumer lag; watch the object count under the prefix.
• Security. Least-privilege IAM, KMS grants only where needed, TLS and SASL to the brokers.

Common problems and troubleshooting

The connector runs but never produces a message

Confluent's generalized connector shows No new files ready after scan task in the task's stack trace. Confluent's troubleshooting page points to topics.dir: unset, the connector expects files under a topics folder. Also check that mode is GENERIC, since backup-and-restore mode ignores files not named like S3 sink output, and that topic.regex.list matches your keys. Aiven's connector processes nothing without file.name.template.

Some new files never reach Kafka

Compare the missing file's key with its neighbours. Aiven and Lenses (by default) read in key order, so a late file under an earlier-sorting key is behind the connector's position. Directory buckets don't sort listings at all.

• Name files so keys sort in arrival order, for example with a timestamp prefix.
• Switch Lenses to LastModified ordering, which still assumes objects don't arrive late.
• Set Aiven's aws.s3.fetch.buffer.size, which exists so slow uploads aren't missed.

RecordTooLargeException on large records

The producer rejects a record over max.request.size, and the broker rejects a batch over message.max.bytes. Split oversized records, or raise the topic's max.message.bytes and the producer's max.request.size together; raising one moves the error to the other.

Using Estuary for Amazon S3 to Kafka

On this route the Amazon S3 capture reads files and the Apache Kafka materialization produces messages. The capture writes every record to a collection keyed by _meta/file and _meta/offset (object path and record position); the materialization produces each document to your topic as it commits, with no sync schedule.

• Discovery and tracking. Each run lists the whole prefix and reads files modified after the newest file it already read, so a late key is still picked up. Files modified in the 30 seconds before a run wait for the next one. Runs start five minutes apart by default (the capture's interval).
• Ascending Keys lists from the last key read instead of the whole prefix. A file whose key sorts earlier is then skipped, so use it only for keys written in time order.
• Overwrites. A re-uploaded file is read again in full, and its records are produced again under the same keys. Write new keys instead.
• Formats.Auto detects the format from the extension or content type, or pick Avro, CSV, JSON, Protobuf, W3C Extended Log or Parquet, compressed with gzip, zip or zstd. Parser settings apply to the whole capture. Parquet files are staged whole: files over 8 GB are skipped and row groups over 1 GB fail, so split large files.
• Skipped objects: zero-byte objects, and the Glacier Flexible Retrieval and Deep Archive storage classes.
• Messages. The key is the collection key. The value holds the key fields and the selected fields, by default the record's top-level fields and flow_published_at.
• Avro subjects are <topic>-<hash of the schema>, not <topic>-value, and the value record is always named root. Consumers that decode by the schema ID in each message work; tools that look up <topic>-value, or generated classes by record name, don't. Use JSON for those.
• Delivery is at least once: the producer is idempotent, but a failed delivery restarts the task, which can produce some messages again.
• Size limits. Strings over 65,536 bytes, and arrays or objects over 1,000 entries, are truncated. No setting raises this.
• Deletes. S3 collections carry no deletion events, so no tombstones are sent.
• Auth is SASL (PLAIN, SCRAM-SHA-256 or SCRAM-SHA-512) over TLS with publicly trusted certificates. There's no IAM option, so use MSK's SCRAM listener.

Implementing Amazon S3 to Kafka with Estuary

These steps adapt Estuary's general data flow guide to this route.

1. Complete the prerequisites in Amazon S3 requirements and Kafka requirements. Estuary also needs:
   • An IAM policy from the connector's setup templates on an IAM user's access key, or on a role Estuary assumes through AWS IAM authentication.
   • Brokers reachable from Estuary's IP addresses, and a SASL user that can write the topics (and create them, if Estuary should).
2. Create the Amazon S3 capture. On the Sources page, click New Capture, choose Amazon S3, name the capture, and fill in the endpoint properties:
   • AWS Region, Bucket and Prefix. Leave Prefix empty to capture the whole bucket.
   • Match Keys: an optional regular expression matched against bucket/key, such as .*\.json.
   • Authentication: Access Key (AWS Access Key ID and AWS Secret Access key), AWS IAM (AWS Role ARN and AWS Region), or Anonymous for a public bucket.
   • Parser Configuration: Format and Compression default to Auto. For CSV, Error Threshold drops up to that percentage of malformed rows instead of failing.
   • Advanced: Ascending Keys (off by default), and AWS Endpoint for an S3-compatible store.

A test failure reading bucket "…" does not exist can also mean access was denied: check the policy.

3. Review the binding. Click Next. Discovery returns one binding for the bucket and prefix, with an inferred schema. For more prefixes, add more bindings. Click Save and publish; the first run reads every file already under the prefix.
4. Create the Apache Kafka materialization. On the capture's details page, click Materialize, choose Apache Kafka, and fill in the endpoint properties:
   • Bootstrap Servers: host:port list.
   • Credentials: SASL (User & Password) with SASL Mechanism, Username and Password. Confluent Cloud uses PLAIN with an API key and secret.
   • Message Format: Avro or JSON (see the subject rules above). Compression defaults to lz4.
   • Schema Registry: Schema Registry Endpoint, Schema Registry Username and Schema Registry Password, required for Avro.
   • Topic Partitions (default 6) and Topic Replication Factor (default 3), used only for topics Estuary creates.
   • Avro Logical Types: on by default, so date-time fields register as timestamp-micros rather than strings.

5. Map the collection to a topic. Click Next. Each binding's Topic defaults to the last segment of the collection name; set it to your topic. Field selection shapes the value, and can add the whole document as a string field.
6. Publish. Click Save and publish. Estuary creates missing topics, leaves existing ones as they are, and starts producing the backlog.
7. Verify that messages are arriving.
   • On the capture details page, the Shard Information card reads PRIMARY. Open the collection from the Collections list to see its Data Preview.
   • The materialization details page charts Data Read.
   • Consume the topic, then repeat the checks from Setting up an Amazon S3 to Kafka pipeline: a new file appears after the next run, and with Ascending Keys off, a file whose key sorts earlier still arrives.

How Estuary handles restarts, re-reads and extra destinations

The capture and the materialization are separate tasks joined by the collection, a durable log of every record read from the bucket.

Restarts resume inside a file

The capture commits each batch of records with its position: the file path and the count of records done. After a restart it re-opens that file and skips records already in the collection. The Kafka side is at least once, but because each message key is a file path and offset, consumers can deduplicate on the key. An overwritten file is the exception: it comes back under the same keys, possibly with different content.

One bucket read for several destinations

To send the same files to Kafka and a warehouse, add a second materialization on the collection; S3 is listed and read once. Rebuilding a topic is a materialization backfill from the collection: no bucket re-read, but every stored record is produced again. Replay reaches back as far as the collection keeps data.

Estuary is SOC 2 Type II certified and HIPAA compliant, with a GDPR badge on its security page.

When another approach may make more sense

• Kafka Connect already runs well: an Apache 2.0 connector from Lenses or Aiven adds no platform. Both read in key order by default, so late keys need the fixes above.
• A few uploads that each need a fast reaction: a Lambda function on s3:ObjectCreated events. Notifications are at least once, a function runs at most 15 minutes, and you own the producer code.
• Heavy transforms before Kafka: Spark Structured Streaming reads files in modification-time order, needs them placed atomically, and writes Kafka at least once. Flink's Kafka sink can write exactly once through transactions, for consumers reading with isolation.level=read_committed.
• Restoring topics from S3 sink backups: Confluent's backup-and-restore mode returns records to their original topics but doesn't restore keys or headers. Lenses' envelope format stores the key, headers and metadata with each record.
• Kafka consumers, but no Kafka cluster: Estuary's Dekaf serves a collection as a read-only Kafka topic, with Estuary as broker and schema registry. Consumers connect to dekaf.estuary-data.com:9092 over SASL_SSL with PLAIN and read Avro. Nothing lands in your own cluster, so use the Kafka materialization when other systems need the topic.

Related resources

• Estuary Amazon S3 capture, multiple prefixes in one capture, capture interval
• Estuary Apache Kafka materialization and Dekaf
• The Amazon S3 to Apache Kafka integration
• AWS: ListObjectsV2, S3 Event Notifications, MSK Connect plugins
• Kafka Connect S3 sources: Confluent generalized, Confluent Cloud, Lenses, Aiven
• Apache Kafka: broker, producer and Connect configs

To try the route, create an Amazon S3 capture in the Estuary dashboard. Pricing: estuary.dev/pricing.