# AWS S3 This Kafka Connect sink connector facilitates the seamless transfer of records from Kafka to AWS S3 Buckets. It offers robust support for various data formats, including AVRO, Parquet, JSON, CSV, and Text, making it a versatile choice for data storage. Additionally, it ensures the reliability of data transfer with built-in support for exactly-once semantics. ## Connector Class ``` io.lenses.streamreactor.connect.aws.s3.sink.S3SinkConnector ``` ## Example {% hint style="success" %} For more examples see the [tutorials](https://docs.lenses.io/latest/connectors/tutorials). {% endhint %} This example writes to a bucket called demo, partitioning by a field called `ts`, store as JSON. {% code fullWidth="true" %} ```properties connector.class=io.lenses.streamreactor.connect.aws.s3.sink.S3SinkConnector connect.s3.kcql=insert into lensesio:demo select * from demo PARTITIONBY _value.ts STOREAS `JSON` PROPERTIES ('flush.size'=1000000, 'flush.interval'=30, 'flush.count'=5000) topics=demo name=demo ``` {% endcode %} ## KCQL Support {% hint style="success" %} You can specify multiple KCQL statements separated by **;** to have a connector sink multiple topics. The connector properties **topics** or **topics.regex** are required to be set to a value that matches the KCQL statements. {% endhint %} The connector uses KCQL to map topics to S3 buckets and paths. The full KCQL syntax is: ```sql INSERT INTO bucketAddress[:pathPrefix] SELECT * FROM kafka-topic [[PARTITIONBY (partition[, partition] ...)] | NOPARTITION] [STOREAS storage_format] [PROPERTIES( 'property.1'=x, 'property.2'=x, )] ``` Please note that you can employ escaping within KCQL for the **INSERT INTO, SELECT \* FROM**, and **PARTITIONBY** clauses when necessary. For example, an incoming Kafka message stored as JSON can use fields containing `.`: ```json { ... "a.b": "value", ... } ``` In this case, you can use the following KCQL statement: ```sql INSERT INTO `bucket-name`:`prefix` SELECT * FROM `kafka-topic` PARTITIONBY `a.b` ``` ## Target Bucket and Path The target bucket and path are specified in the **INSERT INTO** clause. The path is optional and if not specified, the connector will write to the root of the bucket and append the topic name to the path. Here are a few examples: ```sql INSERT INTO testbucket:pathToWriteTo SELECT * FROM topicA; INSERT INTO testbucket SELECT * FROM topicA; INSERT INTO testbucket:path/To/Write/To SELECT * FROM topicA PARTITIONBY fieldA; ``` ## SQL Projection Currently, the connector does not offer support for SQL projection; consequently, anything other than a SELECT \* query is disregarded. The connector will faithfully write all fields from Kafka exactly as they are. ## Source Topic To avoid runtime errors, make sure the *topics* or *topics.regex* setting matches your KCQL statements. If the connector receives data for a topic without matching KCQL, it will throw an error. When using a regex to select topics, follow this KCQL pattern: ``` topics.regex = ^sensor_data_\d+$ connect.s3.kcql= INSERT INTO $target SELECT * FROM `*` .... ``` In this case the topic name will be appended to the $target destination. ## Partitioning & Object Keys The object key serves as the filename used to store data in S3. There are two options for configuring the object key: * **Default**: The object key is automatically generated by the connector and follows the Kafka topic-partition structure. The format is $bucket/\[$prefix]/$topic/$partition/offset.extension. The extension is determined by the chosen storage format. * **Custom**: The object key is driven by the `PARTITIONBY` clause. The format is either `$bucket/[$prefix]/$topic/customKey1=customValue1/customKey2=customValue2/topic(partition_offset).extension` (AWS Athena naming style mimicking Hive-like data partitioning) or `$bucket/[$prefix]/customValue/topic(partition_offset).ext`. The extension is determined by the selected storage format. Custom keys and values can be extracted from the Kafka message key, message value, or message headers, as long as the headers are of types that can be converted to strings. There is no fixed limit to the number of elements that can form the object key, but you should be aware of AWS S3 key length restrictions. {% hint style="warning" %} The Connector automatically adds the topic name to the partition. There is no need to add it to the partition clause. If you want to explicitly add the topic or partition you can do so by using \_topic and \_partition. The partition clause works on header, key and values fields of the Kafka message. {% endhint %} To extract fields from the message values, simply use the field names in the **`PARTITIONBY`** clause. For example: ```sql PARTITIONBY fieldA, fieldB ``` However, note that the message fields must be of primitive types (e.g., string, int, long) to be used for partitioning. You can also use the entire message key as long as it can be coerced into a primitive type: ```sql PARTITIONBY _key ``` In cases where the Kafka message Key is not a primitive but a complex object, you can use individual fields within the message Key to create the S3 object key name: ```sql PARTITIONBY _key.fieldA, _key.fieldB ``` Kafka message headers can also be used in the S3 object key definition, provided the header values are of primitive types easily convertible to strings: ```sql PARTITIONBY _header.[, _header.] ``` Customizing the object key can leverage various components of the Kafka message. For example: ```sql PARTITIONBY fieldA, _key.fieldB, _headers.fieldC ``` This flexibility allows you to tailor the object key to your specific needs, extracting meaningful information from Kafka messages to structure S3 object keys effectively. To enable Athena-like partitioning, use the following syntax: ```sql INSERT INTO $bucket[:$prefix] SELECT * FROM $topic PARTITIONBY fieldA, _key.fieldB, _headers.fieldC STOREAS `AVRO` PROPERTIES ( 'partition.include.keys'=true, ) ``` ## Rolling Windows Storing data in Amazon S3 and partitioning it by time is a common practice in data management. For instance, you may want to organize your S3 data in hourly intervals. This partitioning can be seamlessly achieved using the `PARTITIONBY` clause in combination with specifying the relevant time field. However, it’s worth noting that the time field typically doesn’t adjust automatically. To address this, we offer a Kafka Connect Single Message Transformer (SMT) designed to streamline this process. Let’s consider an example where you need the object key to include the wallclock time (the time when the message was processed) and create an hourly window based on a field called `timestamp`. Here’s the connector configuration to achieve this: {% code fullWidth="true" %} ```properties connector.class=io.lenses.streamreactor.connect.aws.s3.sink.S3SinkConnector connect.s3.kcql=insert into lensesio:demo select * from demo PARTITIONBY _value.metadata_id, _value.customer_id, _header.ts, _header.wallclock STOREAS `JSON` PROPERTIES ('flush.size'=1000000, 'flush.interval'=30, 'flush.count'=5000) topics=demo name=demo value.converter=org.apache.kafka.connect.json.JsonConverter key.converter=org.apache.kafka.connect.storage.StringConverter transforms=insertFormattedTs,insertWallclock transforms.insertFormattedTs.type=io.lenses.connect.smt.header.TimestampConverter transforms.insertFormattedTs.header.name=ts transforms.insertFormattedTs.field=timestamp transforms.insertFormattedTs.target.type=string transforms.insertFormattedTs.format.to.pattern=yyyy-MM-dd-HH transforms.insertWallclock.type=io.lenses.connect.smt.header.InsertWallclock transforms.insertWallclock.header.name=wallclock transforms.insertWallclock.value.type=format transforms.insertWallclock.format=yyyy-MM-dd-HH ``` {% endcode %} In this example, the incoming Kafka message’s Value content includes a field called timestamp, represented as a long value indicating the epoch time in milliseconds. The TimestampConverter SMT will expertly convert this into a string value according to the format specified in the format.to.pattern property. Additionally, the insertWallclock SMT will incorporate the current wallclock time in the format you specify in the format property. The `PARTITIONBY` clause then leverages both the timestamp field and the wallclock header to craft the object key, providing you with precise control over data partitioning. ## Data Storage Format While the **`STOREAS`** clause is optional, it plays a pivotal role in determining the storage format within AWS S3. It’s crucial to understand that this format is entirely independent of the data format stored in Kafka. The connector maintains its neutrality towards the storage format at the topic level and relies on the `key.converter` and `value.converter` settings to interpret the data. Supported storage formats encompass: * AVRO * Parquet * JSON * CSV (including headers) * Text * BYTES Opting for BYTES ensures that each record is stored in its own separate file. This feature proves particularly valuable for scenarios involving the storage of images or other binary data in S3. For cases where you prefer to consolidate multiple records into a single binary file, AVRO or Parquet are the recommended choices. By default, the connector exclusively stores the Kafka message value. However, you can expand storage to encompass the entire message, including the key, headers, and metadata, by configuring the `store.envelope` property as true. This property operates as a boolean switch, with the default value being false. When the envelope is enabled, the data structure follows this format: {% hint style="warning" %} Not supported with a custom partition strategy. {% endhint %} ```json { "key": , "value": , "headers": { "header1": "value1", "header2": "value2" }, "metadata": { "offset": 0, "partition": 0, "timestamp": 0, "topic": "topic" } } ``` Utilizing the envelope is particularly advantageous in scenarios such as backup and restore or replication, where comprehensive storage of the entire message in S3 is desired. ### Examples Storing the message Value Avro data as Parquet in S3: {% code fullWidth="true" %} ```properties ... connect.s3.kcql=INSERT INTO lensesioaws:car_speed SELECT * FROM car_speed_events STOREAS `PARQUET` value.converter=io.confluent.connect.avro.AvroConverter value.converter.schema.registry.url=http://localhost:8081 key.converter=org.apache.kafka.connect.storage.StringConverter ... ``` {% endcode %} The converter also facilitates seamless JSON to AVRO/Parquet conversion, eliminating the need for an additional processing step before the data is stored in S3. {% code fullWidth="true" %} ```properties ... connect.s3.kcql=INSERT INTO lensesioaws:car_speed SELECT * FROM car_speed_events STOREAS `PARQUET` value.converter=org.apache.kafka.connect.json.JsonConverter key.converter=org.apache.kafka.connect.storage.StringConverter ... ``` {% endcode %} Enabling the full message stored as JSON in S3: {% code fullWidth="true" %} ```properties ... connect.s3.kcql=INSERT INTO lensesioaws:car_speed SELECT * FROM car_speed_events STOREAS `JSON` PROPERTIES('store.envelope'=true) value.converter=org.apache.kafka.connect.json.JsonConverter key.converter=org.apache.kafka.connect.storage.StringConverter ... ``` {% endcode %} Enabling the full message stored as AVRO in S3: {% code fullWidth="true" %} ```properties ... connect.s3.kcql=INSERT INTO lensesioaws:car_speed SELECT * FROM car_speed_events STOREAS `AVRO` PROPERTIES('store.envelope'=true) value.converter=io.confluent.connect.avro.AvroConverter value.converter.schema.registry.url=http://localhost:8081 key.converter=org.apache.kafka.connect.storage.StringConverter ... ``` {% endcode %} If the restore (see the S3 Source documentation) happens on the same cluster, then the most performant way is to use the ByteConverter for both Key and Value and store as AVRO or Parquet: {% code fullWidth="true" %} ```properties ... connect.s3.kcql=INSERT INTO lensesioaws:car_speed SELECT * FROM car_speed_events STOREAS `AVRO` PROPERTIES('store.envelope'=true) value.converter=org.apache.kafka.connect.converters.ByteArrayConverter key.converter=org.apache.kafka.connect.converters.ByteArrayConverter ... ``` {% endcode %} ## Flush Options The connector offers three distinct flush options for data management: * Flush by Count - triggers a file flush after a specified number of records have been written to it. * Flush by Size - initiates a file flush once a predetermined size (in bytes) has been attained. * Flush by Interval - enforces a file flush after a defined time interval (in seconds). It’s worth noting that the interval flush is a continuous process that acts as a fail-safe mechanism, ensuring that files are periodically flushed, even if the other flush options are not configured or haven’t reached their thresholds. Consider a scenario where the flush size is set to 10MB, and only 9.8MB of data has been written to the object, with no new Kafka messages arriving for an extended period of 6 hours. To prevent undue delays, the interval flush guarantees that the object is flushed after the specified time interval has elapsed. This ensures the timely management of data even in situations where other flush conditions are not met. The flush options are configured using the **flush.count**, **flush.size**, and **flush.interval** properties. The settings are optional and if not specified the defaults are: * flush.count = 50\_000 * flush.size = 500000000 (500MB) * flush.interval = 3\_600 (1 hour) {% hint style="success" %} A connector instance can simultaneously operate on multiple topic partitions. When one partition triggers a flush, it will initiate a flush operation for all of them, even if the other partitions are not yet ready to flush. {% endhint %} When `connect.s3.latest.schema.optimization.enabled` is set to true, it reduces unnecessary data flushes when writing to Avro or Parquet formats. Specifically, it leverages schema compatibility to avoid flushing data when messages with older but *backward-compatible* schemas are encountered. Consider the following sequence of messages and their associated schemas: ``` pgsqlCopyEditmessage1 -> schema1 message2 -> schema1 (No flush needed – same schema) message3 -> schema2 (Flush occurs – new schema introduced) message4 -> schema2 (No flush needed – same schema) message5 -> schema1 Without optimization: would trigger a flush With optimization: no flush – schema1 is backward-compatible with schema2 message6 -> schema2 message7 -> schema2 (No flush needed – same schema, it would happen based on the flush thresholds) ``` ### Flushing By Interval The next flush time is calculated based on the time the previous flush completed (the last modified time of the object written to S3). Therefore, by design, the sink connector’s behaviour will have a slight drift based on the time it takes to flush records and whether records are present or not. If Kafka Connect makes no calls to put records, the logic for flushing won't be executed. This ensures a more consistent number of records per object. ![sink commit.png](https://1499290846-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FETXWnZknWA8VhF6SwqCx%2Fuploads%2Fgit-blob-9e35cb6cd77a2957e60eac96448f6164084f9443%2Fsink%20commit.png?alt=media) ## Properties The **PROPERTIES** clause is optional and adds a layer of configuration to the connector. It enhances versatility by permitting the application of multiple configurations (delimited by ‘,’). The following properties are supported: | Name | Description | Type | Available Values | Default Value | | ------------------------------ | ----------------------------------------------------------------------------------------------------------------------- | ----------------------- | ----------------------- | -------------------------------------------------------------------- | | padding.type | Specifies the type of padding to be applied. | LeftPad, RightPad, NoOp | LeftPad, RightPad, NoOp | LeftPad | | padding.char | Defines the character used for padding. | Char | | ‘0’ | | padding.length.partition | Sets the padding length for the partition. | Int | | 0 | | padding.length.offset | Sets the padding length for the offset. | Int | | 12 | | partition.include.keys | Specifies whether partition keys are included. | Boolean | |

false
Default (Custom Partitioning): true

| | store.envelope | Indicates whether to store the entire Kafka message | Boolean | | | | store.envelope.fields.key | Indicates whether to store the envelope’s key. | Boolean | | | | store.envelope.fields.headers | Indicates whether to store the envelope’s headers. | Boolean | | | | store.envelope.fields.value | Indicates whether to store the envelope’s value. | Boolean | | | | store.envelope.fields.metadata | Indicates whether to store the envelope’s metadata. | Boolean | | | | flush.size | Specifies the size (in bytes) for the flush operation. | Long | | 500000000 (500MB) | | flush.count | Specifies the number of records for the flush operation. | Int | | 50000 | | flush.interval | Specifies the interval (in seconds) for the flush operation | Long | | 3600(1h) | | key.suffix | When specified it appends the given value to the resulting object key before the "extension" (avro, json, etc) is added | String | | \ | The sink connector optimizes performance by padding the output objects. This proves beneficial when using the S3 Source connector to restore data. This object name padding ensures that objects are ordered lexicographically, allowing the S3 Source connector to skip the need for reading, sorting, and processing all objects, thereby enhancing efficiency. ## Compression AVRO and Parquet offer the capability to compress files as they are written. The S3 Sink connector provides advanced users with the flexibility to configure compression options. Here are the available options for the `connect.s3.compression.codec`, along with indications of their support by Avro, Parquet and JSON writers:
CompressionAvro SupportAvro (requires Level)Parquet SupportJSON
UNCOMPRESSED✅✅✅
SNAPPY✅✅
GZIP✅✅
LZ0✅
LZ4✅
BROTLI✅
BZIP2✅
ZSTD✅⚙️✅
DEFLATE✅⚙️
XZ✅⚙️
Please note that not all compression libraries are bundled with the S3 connector. Therefore, you may need to manually add certain libraries to the classpath to ensure they function correctly. ## Authentication The connector offers two distinct authentication modes: * Default: This mode relies on the default AWS authentication chain, simplifying the authentication process. * Credentials: In this mode, explicit configuration of AWS Access Key and Secret Key is required for authentication. Here’s an example configuration for the **Credentials** mode: ```properties ... connect.s3.aws.auth.mode=Credentials connect.s3.aws.region=eu-west-2 connect.s3.aws.access.key=$AWS_ACCESS_KEY connect.s3.aws.secret.key=$AWS_SECRET_KEY ... ``` For enhanced security and flexibility when using the **Credentials** mode, it is highly advisable to utilize Connect Secret Providers. ## Error policies The connector supports [Error policies](https://docs.lenses.io/latest/connectors/tutorials/using-error-policies). ### Retry behaviour The connector applies retries at **two independent layers**. They are complementary, not duplicates: each one targets a different category of failure, and both are active at the same time. #### Layer 1 — HTTP / AWS SDK retries Every individual call the connector makes to S3 (object upload, copy, delete, list, get, head) is routed through the official AWS SDK for Java, which transparently retries transient failures using exponential backoff. These retries are **invisible to Kafka Connect**: they happen entirely inside a single `put()` invocation, and the connector only sees the failure if every HTTP attempt has been exhausted. Typical failures absorbed at this layer: * TCP / TLS handshake errors, connection resets, DNS hiccups * HTTP 5xx responses from S3 * Throttling errors (`SlowDown`, `RequestLimitExceeded`, `ProvisionedThroughputExceededException`) * Short-lived endpoint blips Properties: | Name | Description | Type | Default | | -------------------------------- | ---------------------------------------------------------------------------------------------------------------------------- | ---- | ------- | | `connect.s3.http.max.retries` | Maximum number of attempts per individual HTTP request. | int | 5 | | `connect.s3.http.retry.interval` | Initial backoff delay (in milliseconds) before the first HTTP retry. The AWS SDK applies its own exponential backoff on top. | long | 50 | The AWS SDK handles the per-attempt backoff internally; the connector does not expose a separate multiplier knob. {% hint style="info" %} The AWS SDK's classification of "retryable" errors is broad and covers the vast majority of transient S3 issues. Increase `connect.s3.http.max.retries` if you operate against a heavily throttled bucket or over a noisy network. {% endhint %} #### Layer 2 — Connector / Kafka Connect retries When **all** HTTP retries above have been exhausted, or when an error happens **outside** an HTTP call (serialisation, schema, file-system, etc.), control returns to the connector's error policy. If the policy is `RETRY`, the connector throws a `RetriableException`, which causes Kafka Connect to **redeliver the same batch of records** to `put()` after a delay. This is repeated until the batch eventually succeeds or the configured retry budget is exhausted. Properties: | Name | Description | Type | Default | | --------------------------- | ------------------------------------------------------------------------------------------------ | ------ | ------- | | `connect.s3.error.policy` | `THROW` (fail immediately), `NOOP` (swallow and continue), or `RETRY` (re-deliver the batch). | string | `THROW` | | `connect.s3.max.retries` | Maximum number of batch redeliveries before the task fails. Only used when `error.policy=RETRY`. | int | 20 | | `connect.s3.retry.interval` | Delay (in milliseconds) between batch redeliveries. Only used when `error.policy=RETRY`. | int | 60000 | {% hint style="warning" %} If `connect.s3.error.policy` is left at its default `THROW`, the `max.retries` and `retry.interval` settings are **not** used — any error escaping the HTTP layer will fail the task immediately. {% endhint %} #### Which layer handles what | Failure category | Handled by | Properties to tune | | ------------------------------------------------------------------------------------------------------------------------ | -------------------------------------- | ---------------------------------------------------------------------------------------- | | Transient cloud noise: 5xx, throttling, network resets, DNS / TLS blips | AWS SDK retries (silent) | `connect.s3.http.max.retries`, `connect.s3.http.retry.interval` | | Sustained S3 unavailability, IAM / auth failures, schema or format errors, or anything that escapes the SDK retry budget | Connector-level retry policy (`RETRY`) | `connect.s3.error.policy=RETRY` + `connect.s3.max.retries` + `connect.s3.retry.interval` | #### Choosing values * **Tune the `http.*` settings to absorb cloud noise.** The defaults (5 attempts, starting at 50 ms) are sensible for most workloads. Increase `connect.s3.http.max.retries` if you operate over a noisy network or against a heavily-throttled bucket. * **Use `error.policy=RETRY` as a backstop** for longer outages. The total ride-through window is approximately `max.retries x retry.interval`. With the defaults (20 x 60 s) the task survives roughly 20 minutes of continuous failure before giving up. * **Combine with Kafka Connect's framework-level error handling** (`errors.tolerance=all`, `errors.deadletterqueue.topic.name`, etc.) for **per-record** poison pills (converter / SMT failures). The framework's `errors.tolerance` is **not** a substitute for `error.policy=RETRY`: it handles record-level errors, not batch-level S3 infrastructure failures. The two settings address different failure modes and are intended to be used together. #### Example A robust production configuration that combines both layers and adds Kafka Connect's poison-pill protection: ```properties # Layer 1 - leave at defaults, or relax for noisy networks # connect.s3.http.max.retries=5 # connect.s3.http.retry.interval=50 # Layer 2 - ride through up to ~30 minutes of S3 unavailability connect.s3.error.policy=RETRY connect.s3.max.retries=30 connect.s3.retry.interval=60000 # Kafka Connect framework - per-record DLQ for converter / SMT errors errors.tolerance=all errors.log.enable=true errors.log.include.messages=true errors.deadletterqueue.topic.name=my-connector-dlq errors.deadletterqueue.context.headers.enable=true errors.deadletterqueue.topic.replication.factor=3 ``` ### Offset commit semantics A frequent question is **when** the S3 sink connector advances Kafka consumer offsets, and what role the various "temporary" locations play in that process. {% hint style="success" %} Offsets are only advanced after the data has been durably written to its final object key in S3 (with exactly-once enabled, the connector's .indexes/ entry must also be updated first). Neither the local staging file nor the transient .temp-upload/... S3 object causes offsets to advance. {% endhint %} #### End-to-end flow for one batch
1. **Local staging.** Incoming records are serialised and appended to a file on the **Connect worker's local disk**, inside the directory pointed to by `connect.s3.local.tmp.directory` (or an OS temp directory if not set). Nothing is written to S3 and no Kafka offsets advance. If the task crashes here, the records are simply re-consumed from Kafka on restart. 2. **Flush.** When a flush threshold is reached (`flush.count`, `flush.size`, `flush.interval`, schema change, etc.), the connector uploads the local staging file to S3. The pipeline depends on `connect.s3.exactly.once.enable`: * **Exactly-once (default, `exactly.once.enable=true`).** A 3-step pipeline that goes via a transient object, fenced by S3 ETags: 1. **Upload** the local staging file to a transient object at `s3:///.temp-upload////`. 2. **Copy** that transient object to the final destination key, using an `If-Match`/ETag precondition so a concurrent writer (e.g. during a rebalance) cannot overwrite or duplicate the final object. 3. **Delete** the transient object under `.temp-upload/`. e `.indexes/` entry is updated after each step so that a restarted task can pick up the pipeline mid-flight. At-least-once (`exactly.once.enable=false`).\*\* The local staging file is uploaded **directly to the final destination key**. There is no `.temp-upload/` indirection, no ETag fencing, and no `.indexes/` checkpoint. The connector falls back to Kafka Connect's native at-least-once offset management. 3. **Committed offset advances.** Only once the final object is in place at its proper destination key — and, with exactly-once enabled, only once the corresponding `.indexes/` entry has been updated — does the writer's committed offset advance. 4. **`preCommit`.** The next time Kafka Connect calls `preCommit`, the connector returns the latest safely committed offset for each partition. If any records are still buffered locally and not yet flushed, the connector returns the offset of the **first still-buffered record** — Kafka Connect will not advance past anything that is not durably in S3. #### The two "temporary" locations | Location | Where it physically lives | Purpose | Affects committed offset? | | ------------------------------------------------------- | --------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ---------------------------- | | Local staging file | Disk on the Connect worker (`connect.s3.local.tmp.directory` or OS tmp) | Buffers Kafka records into the chosen file format (AVRO / Parquet / JSON / CSV / Text / BYTES) before flush. Bounded by `flush.size` / `flush.count` / `flush.interval`. | No | | `.temp-upload////...` S3 object | The **same S3 bucket** as the destination, under the `.temp-upload/` prefix | Atomic, fenced staging step used only when exactly-once is enabled. The local file is uploaded here, then copied to the final key with an ETag precondition, then deleted. Exists only for one commit. | No | | Final destination key | The configured S3 bucket / path | The actual data that downstream consumers read. | **Yes** (after index update) | {% hint style="info" %} Nothing is held in memory only. The local staging file is real on-disk storage on the worker, and .temp-upload/... is a real S3 object. This is what allows the connector to recover cleanly from crashes, rebalances and worker restarts. {% endhint %} #### Restart behavior The connector is designed so that no Kafka offset ever moves ahead of data that has been durably written to its final S3 key. The exact failure modes vary by mode: **Exactly-once (default)** * **Crash during local staging** — nothing in S3, offsets unchanged. Records are re-consumed from Kafka. No duplicates, no data loss. * **Crash mid-pipeline** (between upload, copy and delete in `.temp-upload/`) — the `.indexes/` entry records exactly which step was reached. On restart the connector resumes the pipeline from that point. The ETag precondition on the copy step prevents two writers from racing the final key during a rebalance. * **Crash after the final write but before the index entry advances** — the final object is in S3 but the offset has not advanced. On restart the records are re-uploaded; the ETag fence keeps the existing final object unchanged. **At-least-once (`exactly.once.enable=false`)** * **Crash during local staging** behaves identically to the exactly-once case (records re-consumed, no data loss). * **Crash during or after the upload to the final key** may produce duplicate or partially-overwritten objects on restart, because there is no fencing and no `.indexes/` checkpoint. Use this mode only when downstream consumers can tolerate duplicates. #### Operational notes * Tuning the flush thresholds (`flush.count` / `flush.size` / `flush.interval`) controls how often offsets advance, and therefore how much data is replayed after a worker crash. Smaller flush windows = smaller replays on restart, at the cost of more, smaller objects in S3. * The `.temp-upload/` prefix is internal connector machinery. It is safe to ignore in S3 lifecycle policies, but **do not exclude it from the connector's IAM permissions** — the connector needs `s3:PutObject`, `s3:GetObject`, `s3:DeleteObject` and `s3:CopyObject`-equivalent privileges on objects under that prefix. * If your bucket has Object Lock or strict bucket policies, ensure they allow the short-lived `.temp-upload/...` objects to be written **and deleted** during the commit pipeline. ## API Compatible systems The connector can also be used against API compatible systems provided they implement the following: ``` listObjectsV2 listObjectsV2Pagbinator putObject getObject headObject deleteObjects deleteObject ``` ## Indexes Directory The connector uses the concept of index objects that it writes to in order to store information about the latest offsets for Kafka topics and partitions as they are being processed. This allows the connector to quickly resume from the correct position when restarting and provides flexibility in naming the index objects. By default, the index objects are grouped within a prefix named `.indexes` for all connectors. However, each connector will create and store its index objects within its own nested prefix inside this `.indexes` prefix. You can configure the prefix for these index objects using the property connect.s3.indexes.name. This property specifies the path from the root of the S3 bucket. Note that even if you configure this property, the connector will still place the indexes within a nested prefix of the specified prefix. ### Examples | Index Name (`connect.s3.indexes.name`) | Resulting Indexes Prefix Structure | Description | | -------------------------------------- | --------------------------------------------- | -------------------------------------------------------------------------------------------------------- | | `.indexes` (default) | `.indexes//` | The default setup, where each connector uses its own subdirectory within `.indexes`. | | `custom-indexes` | `custom-indexes//` | Custom root directory `custom-indexes`, with a subdirectory for each connector. | | `indexes/s3-connector-logs` | `indexes/s3-connector-logs//` | Uses a custom subdirectory `s3-connector-logs` within `indexes`, with a subdirectory for each connector. | | `logs/indexes` | `logs/indexes//` | Indexes are stored under `logs/indexes`, with a subdirectory for each connector. | ## Option Reference | Name | Description | Type | Available Values | Default Value | | -------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------- | ------------------------------------------------------------------------------------------ | -------------- | | connect.s3.aws.auth.mode | Specifies the AWS authentication mode for connecting to S3. | string | “Credentials,” “Default” | “Default” | | connect.s3.aws.access.key | The AWS Access Key used for authentication. | string | | (Empty) | | connect.s3.aws.secret.key | The AWS Secret Key used for authentication. | string | | (Empty) | | connect.s3.aws.region | The AWS Region where the S3 bucket is located. | string | | (Empty) | | connect.s3.pool.max.connections | Specifies the maximum number of connections allowed in the AWS Client’s HTTP connection pool when interacting with S3. | int | -1 (undefined) | 50 | | connect.s3.custom.endpoint | Allows for the specification of a custom S3 endpoint URL if needed. | string | | (Empty) | |

\[Deprecated]

connect.s3.vhost.bucket

|

Enables the use of Vhost Buckets for S3 connections. Always set to true when custom endpoints are used.

Deprecation:
This setting maps directly to the AWS SDK’s pathStyleAccessEnabled() method. However, these two concepts are semantically opposite.

Use
available since version 11.3.0.
connect.s3.path.style.access

| boolean | true, false | false | | connect.s3.path.style.access |

When set to

true it enables path-style access (matches AWS SDK semantics). When set to

false it enables virtual-hosted style access.

| boolean | true, false | | | connect.s3.error.policy | connect.s3.path.style.access Defines the error handling policy when errors occur during data transfer to or from S3. | string | “NOOP,” “THROW,” “RETRY” | “THROW” | | connect.s3.max.retries | Sets the maximum number of retries the connector will attempt before reporting an error to the Connect Framework. | int | | 20 | | connect.s3.retry.interval | Specifies the interval (in milliseconds) between retry attempts by the connector. | int | | 60000 | | connect.s3.http.max.retries | Sets the maximum number of retries for the underlying HTTP client when interacting with S3. | long | | 5 | | connect.s3.http.retry.interval | Specifies the retry interval (in milliseconds) for the underlying HTTP client. An exponential backoff strategy is employed. | long | | 50 | | connect.s3.local.tmp.directory | Enables the use of a local folder as a staging area for data transfer operations. | string | | (Empty) | | connect.s3.kcql | A SQL-like configuration that defines the behavior of the connector. Refer to the KCQL section below for details. | string | | (Empty) | | connect.s3.compression.codec | Sets the Parquet compression codec to be used when writing data to S3. | string | “UNCOMPRESSED,” “SNAPPY,” “GZIP,” “LZ0,” “LZ4,” “BROTLI,” “BZIP2,” “ZSTD,” “DEFLATE,” “XZ” | “UNCOMPRESSED” | | connect.s3.compression.level | Sets the compression level when compression is enabled for data transfer to S3. | int | 1-9 | (Empty) | | connect.s3.seek.max.files | Specifies the maximum threshold for the number of files the connector uses to ensure exactly-once processing of data. | int | | 5 | | connect.s3.indexes.name | Configure the indexes prefix for this connector. | string | | ".indexes" | | connect.s3.exactly.once.enable | By setting to 'false', disable exactly-once semantics, opting instead for Kafka Connect’s native at-least-once offset management | boolean | true, false | true | | connect.s3.schema.change.detector | Configure how the file will roll over upon receiving a record with a schema different from the accumulated ones. This property configures schema change detection with `default` (object equality), `version` (version field comparison), or `compatibility` (Avro compatibility checking). | string | `default`, `version`, `compatibility` | `default` | | connect.s3.skip.null.values | Skip records with null values (a.k.a. tombstone records). | boolean | true, false | false | | connect.s3.latest.schema.optimization.enabled | When set to true, reduces unnecessary data flushes when writing to Avro or Parquet formats. Specifically, it leverages schema compatibility to avoid flushing data when messages with older but *backward-compatible* schemas are encountered. | boolean | true,false | false | --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://docs.lenses.io/latest/connectors/kafka-connectors/sinks/aws-s3.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.