Configuration Reference for Amazon S3 Sink Connector for Confluent Platform

To use this connector, specify the name of the connector class in the connector.class configuration property.

connector.class=io.confluent.connect.s3.S3SinkConnector

Connector-specific configuration properties are described below.

Note

These are properties for the self-managed connector. If you are using Confluent Cloud, see Amazon S3 Sink connector for Confluent Cloud.

Connector

format.class

The format class to use when writing data to the store.

  • Type: class

  • Valid Values: Any class that implements the following classes:

    • io.confluent.connect.s3.format.avro.AvroFormat

    • io.confluent.connect.s3.format.bytearray.ByteArrayFormat

    • io.confluent.connect.s3.format.json.JsonFormat

    • io.confluent.connect.s3.format.parquet.ParquetFormat

  • Importance: high

flush.size

Number of records written to store before invoking file commits.

  • Type: int

  • Importance: high

rotate.interval.ms

The time interval in milliseconds to invoke file commits. You can configure this parameter so that the time interval is determined by using a timestamp extractor (for example, Kafka Record Time, Record Field, or Wall Clock extractor). When the first record is processed, a timestamp is set as the base time. This is useful if you require exactly-once-semantics. This configuration ensures that file commits are invoked at every configured interval. The default value -1 indicates that this feature is disabled.

The S3 connector allows one writer at a time for this parameter. When the S3 connector encounters data that belongs to a previously closed output partition, it closes the current partition and creates a new partition in S3. If there is no late data arriving, the S3 connector closes the current partition when it reads a record whose extracted timestamp is greater than the specified amount of time.

  • Type: long

  • Default: -1

  • Importance: high

rotate.schedule.interval.ms

The time interval in milliseconds to periodically invoke file commits. This configuration ensures that file commits are invoked at every configured interval. Time of commit will be adjusted to 00:00 of selected timezone. The commit will be performed at the scheduled time, regardless of the previous commit time or number of messages. This configuration is useful when you have to commit your data based on current server time, for example at the beginning of every hour. The default value -1 means that this feature is disabled.

The S3 Connector supports multiple writers. When the S3 connector encounters late arriving data, it keeps the current partition open and creates a new partition for the late data in S3. It rolls up files based on the time interval specified. For example, if rotate.schedule.interval.ms=180000, it will roll up files 20 times in one hour.

  • Type: long

  • Default: -1

  • Importance: medium

The following Avro converter properties can be used in the connector configuration:

schema.cache.config

The size of the schema cache used in the Avro converter.

  • Type: int

  • Default: 1000

  • Importance: low

enhanced.avro.schema.support

Enable enhanced Avro schema support in the Avro Converter. When set to true, this property preserves Avro schema package information and Enums when going from Avro schema to Connect schema. This information is added back in when going from Connect schema to Avro schema.

  • Type: boolean

  • Default: false

  • Importance: low

connect.meta.data

Allow the Connect converter to add its metadata to the output schema.

  • Type: boolean

  • Default: true

  • Importance: low

The connect.meta.data property preserves the following Connect schema metadata when going from Connect schema to Avro schema. The following metadata is added back in when going from Avro schema to Connect schema.

  • doc

  • version

  • parameters

  • default value

  • name

  • type

For detailed information and configuration examples for Avro converters listed above, see Using Kafka Connect with Schema Registry.

retry.backoff.ms

The retry backoff in milliseconds. This config is used to notify Kafka Connect to retry delivering a message batch or performing recovery in case of transient exceptions.

  • Type: long

  • Default: 5000

  • Importance: low

filename.offset.zero.pad.width

Width to zero pad offsets in store’s filenames if offsets are too short in order to provide fixed width filenames that can be ordered by simple lexicographic sorting.

  • Type: int

  • Default: 10

  • Valid Values: [0,…]

  • Importance: low

avro.codec

The Avro compression codec to be used for output files. Available values: null, deflate, snappy and bzip2 (CodecSource is org.apache.avro.file.CodecFactory)

  • Type: string

  • Default: null

  • Valid Values: [null, deflate, snappy, bzip2]

  • Importance: low

parquet.codec

The Parquet compression codec to be used for output files.

  • Type: string

  • Default: snappy

  • Valid Values: [none, gzip, snappy, lz4, brotli, zstd, lzo]

  • Importance: low

Schema

schema.compatibility

The schema compatibility rule to use when the connector is observing schema changes. The supported configurations are NONE, BACKWARD, FORWARD and FULL.

  • Type: string

  • Default: NONE

  • Importance: high

allow.optional.map.keys

An Avro converter configuration parameter you can use in the connector configuration to allow an optional string map key when converting from Connect Schema to Avro Schema.

  • Type: boolean

  • Default: false

  • Importance: low

S3

s3.bucket.name

The S3 Bucket.

  • Type: string

  • Importance: high

s3.object.tagging

Tag S3 objects with start and end offsets, as well as record count.

  • Type: boolean

  • Default: false

  • Importance: low

s3.region

The AWS region to be used the connector.

  • Type: string

  • Default: us-west-2

  • Valid Values: [us-iso-east-1, us-isob-east-1, us-gov-east-1, us-gov-west-1, ap-east-1, ap-northeast-1, ap-northeast-2, ap-south-1, ap-southeast-1, ap-southeast-2, ca-central-1, eu-central-1, eu-north-1, eu-west-1, eu-west-2, eu-west-3, me-south-1, sa-east-1, us-east-1, us-east-2, us-west-1, us-west-2, cn-north-1, cn-northwest-1]

  • Importance: medium

s3.part.size

The Part Size in S3 Multi-part Uploads.

  • Type: int

  • Default: 26214400

  • Valid Values: [5242880,…,2147483647]

  • Importance: high

s3.elastic.buffer.enable

Specifies whether or not to allocate elastic buffer for staging s3-part to save memory. Note that this may cause decreased performance or increased CPU usage.

  • Type: boolean

  • Default: false

  • Importance: low

s3.elastic.buffer.init.capacity

Elastic buffer initial capacity. Only applicable when s3.elastic.buffer.enable is set true.

  • Type: int

  • Default: 131072 (128 KB)

  • Valid Values: [4096,…]

  • Importance: low

s3.credentials.provider.class

Credentials provider or provider chain to use for authentication to AWS. By default the connector uses DefaultAWSCredentialsProviderChain. If you are using an assume role in the credentials provider, proxy configuration will not work. This is a limitation of the underlying AWS SDK v1 client.

  • Type: class

  • Default: com.amazonaws.auth.DefaultAWSCredentialsProviderChain

  • Valid Values: Any class implementing: interface com.amazonaws.auth.AWSCredentialsProvider

  • Importance: low

aws.access.key.id

The AWS access key ID used to authenticate personal AWS credentials such as IAM credentials. Use only if you do not wish to authenticate by using a credentials provider class,``s3.credentials.provider.class``.

  • Type: string

  • Default: “”

  • Importance: high

aws.secret.access.key

The secret access key used to authenticate personal AWS credentials such as IAM credentials. Use this only if you do not want to use a credentials provider class for authentication. You select a credentials provider class using the s3.credentials.provider.class configuration property.

  • Type: password

  • Default: [hidden]

  • Importance: high

s3.ssea.name

The S3 Server Side Encryption Algorithm.

  • Type: string

  • Default: “”

  • Valid Values: [, AES256, aws:kms]

  • Importance: low

s3.sse.customer.key

The S3 Server Side Encryption Customer-Provided Key (SSE-C).

  • Type: password

  • Default: [hidden]

  • Importance: low

s3.sse.kms.key.id

The name of the AWS Key Management Service (AWS-KMS) key to be used for server side encryption of the S3 objects. No encryption is used when no key is provided, but it is enabled when aws:kms is specified as encryption algorithm with a valid key name.

  • Type: string

  • Default: “”

  • Importance: low

s3.acl.canned

An S3 canned ACL header value to apply when writing objects.

  • Type: string

  • Default: null

  • Valid Values: [private, public-read, public-read-write, authenticated-read, log-delivery-write, bucket-owner-read, bucket-owner-full-control, aws-exec-read]

  • Importance: low

s3.wan.mode

Use S3 accelerated endpoint.

  • Type: boolean

  • Default: false

  • Importance: medium

s3.compression.type

Compression type for files written to S3. Applied when using JsonFormat or ByteArrayFormat. Available values: none, gzip.

  • Type: string

  • Default: none

  • Valid Values: [none, gzip]

  • Importance: low

s3.compression.level

Compression level for files written to S3. Applied when using JsonFormat or ByteArrayFormat.

  • Type: int

  • Default: -1

  • Valid Values: -1 for system default, or [-1,…,9] for levels between no compression and best compression

  • Importance: low

s3.part.retries

Maximum number of retry attempts for failed requests. Zero means no retries. The actual number of attempts is determined by the S3 client based on multiple factors including, but not limited to: the value of this parameter, type of exception occurred, and throttling settings of the underlying S3 client.

  • Type: int

  • Default: 3

  • Valid Values: [0,…]

  • Importance: medium

s3.retry.backoff.ms

How long to wait in milliseconds before attempting the first retry of a failed S3 request. Upon a failure, this connector may wait up to twice as long as the previous wait, up to the maximum number of retries. This avoids retrying in a tight loop under failure scenarios.

  • Type: long

  • Default: 200

  • Valid Values: [0,…]

  • Importance: low

format.bytearray.extension

Output file extension for ByteArrayFormat. Defaults to .bin.

  • Type: string

  • Default: .bin

  • Importance: low

format.bytearray.separator

String inserted between records for ByteArrayFormat. Defaults to System.lineSeparator() and may contain escape sequences like \n. An input record that contains the line separator will look like multiple records in the output S3 object.

  • Type: string

  • Default: null

  • Importance: low

s3.proxy.url

S3 Proxy settings encoded in URL syntax. This property is meant to be used only if you need to access S3 through a proxy.

  • Type: string

  • Default: “”

  • Importance: low

s3.proxy.user

S3 Proxy User. This property is meant to be used only if you need to access S3 through a proxy. Using s3.proxy.user instead of embedding the username and password in s3.proxy.url allows the password to be hidden in the logs.

  • Type: string

  • Default: null

  • Importance: low

s3.proxy.password

S3 Proxy Password. This property is meant to be used only if you need to access S3 through a proxy. Using s3.proxy.password instead of embedding the username and password in s3.proxy.url allows the password to be hidden in the logs.

  • Type: password

  • Default: [hidden]

  • Importance: low

s3.http.send.expect.continue

Enable or disable use of the HTTP/1.1 handshake using EXPECT: 100-CONTINUE during multi-part upload. If true, the client will wait for a 100 (CONTINUE) response before sending the request body. Else, the client uploads the entire request body without checking if the server is willing to accept the request.

  • Type: boolean

  • Default: true

  • Importance: low

behavior.on.null.values

How to handle records with a null value (for example, Kafka tombstone records). Valid options are ignore, fail and write. Note that if you set behavior.on.null.values to write, store.kafka.keys must be set to true. Additionally, if you set behavior.on.null.values to ignore, and errors.tolerance to none, you must disable the reporting of null records to the Dead Letter Queue (DLQ) . To achieve this, set errors.log.enable to false and do not configure a DLQ topic.

  • Type: string

  • Default: fail

  • Valid Values: [ignore, fail, write]

  • Importance: low

sts.role.arn

Role ARN to use when starting a session.

  • Type: string

  • Default: null

  • Importance: high

sts.role.external.id

The role external ID used when retrieving session credentials under an assumed role.

  • Type: string

  • Default: null

  • Importance: medium

sts.role.session.name

Role session name to use when starting a session.

  • Type: string

  • Default: null

  • Importance: high

tombstone.encoded.partition

The Amazon S3 output folder to write the tombstone records to. The configured partitioner would map tombstone records to this output folder.

  • Type: string

  • Default: "tombstone"

  • Importance: low

Storage

storage.class

The underlying storage layer.

  • Type: class

  • Importance: high

topics.dir

Top level directory to store the data ingested from Kafka.

  • Type: string

  • Default: topics

  • Importance: high

store.url

Store’s connection URL, if applicable. For example: http://bucket.s3-aws-region.amazonaws.com.

  • Type: string

  • Default: null

  • Importance: high

directory.delim

Directory delimiter pattern

  • Type: string

  • Default: /

  • Importance: medium

file.delim

File delimiter pattern

  • Type: string

  • Default: +

  • Importance: medium

Keys and headers

store.kafka.keys

Enable or disable writing record keys to storage.

  • Type: boolean

  • Default: false

  • Importance: low

keys.format.class

The format class to use when writing keys to the store. Possible formats are Avro, Json, ByteArray and Parquet.

  • Type: class

  • Default: io.confluent.connect.s3.format.avro.AvroFormat

  • Importance: low

store.kafka.headers

Enable or disable writing record headers to storage.

  • Type: boolean

  • Default: false

  • Importance: low

headers.format.class

The format class to use when writing keys to the store. Possible formats are Avro, Json and Parquet.

  • Type: class

  • Default: io.confluent.connect.s3.format.avro.AvroFormat

  • Importance: low

s3.path.style.access.enabled

Specifies whether or not to enable path-style access to the bucket used by the connector. Setting this property to false enables the S3 connector to work in environments where path-style access is disabled.

  • Type: boolean

  • Default: true

  • Importance: low

Partitioner

partitioner.class

The partitioner to use when writing data to the store. You can use DefaultPartitioner, which preserves the Kafka partitions; FieldPartitioner, which partitions the data to different directories according to the value of the partitioning field specified in partition.field.name; TimeBasedPartitioner, which partitions data according to ingestion time.

  • Type: class

  • Default: io.confluent.connect.storage.partitioner.DefaultPartitioner

  • Importance: high

  • Dependents: partition.field.name, partition.duration.ms, path.format, locale, timezone

partition.field.name

The name of the partitioning field when FieldPartitioner is used.

Note

This configuration only supports fields that are directly available in the JSON record structure. Nested fields (for example, user.address.city) are not supported. Only top-level fields (for example, user_id, timestamp) can be used for partitioning.

  • Type: list

  • Default: “”

  • Importance: medium

partition.duration.ms

The duration of a partition milliseconds used by TimeBasedPartitioner. The default value -1 means that we are not using TimeBasedPartitioner.

  • Type: long

  • Default: -1

  • Importance: medium

path.format

This configuration is used to set the format of the data directories when partitioning with TimeBasedPartitioner. The format set in this configuration converts the Unix timestamp to proper directories strings. For example, if you set path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH, the data directories will have the format /year=2015/month=12/day=07/hour=15/.

  • Type: string

  • Default: “”

  • Importance: medium

locale

The locale to use when partitioning with TimeBasedPartitioner. Used to format dates and times. For example, use en-US for US English, en-GB for UK English, or fr-FR for French (in France). These may vary by Java version. See the available locales.

  • Type: string

  • Default: “”

  • Importance: medium

timezone

The timezone to use when partitioning with TimeBasedPartitioner. Used to format and compute dates and times. All timezone IDs must be specified in the long format, such as America/Los_Angeles, America/New_York, and Europe/Paris, or UTC. Alternatively a locale independent, fixed offset, datetime zone can be specified in form [+-]hh:mm. Support for these timezones may vary by Java version. See the available timezones within each locale, such as those within the US English locale.

  • Type: string

  • Default: “”

  • Importance: medium

timestamp.extractor

The extractor that gets the timestamp for records when partitioning with TimeBasedPartitioner. It can be set to Wallclock, Record or RecordField in order to use one of the built-in timestamp extractors or be given the fully-qualified class name of a user-defined class that extends the TimestampExtractor interface.

  • Type: string

  • Default: Wallclock

  • Importance: medium

timestamp.field

The record field to be used as timestamp by the timestamp extractor.

  • Type: string

  • Default: timestamp

  • Importance: medium

CSFLE configuration

csfle.enabled

Accepts a boolean value. CSFLE is enabled for the connector if csfle.enabled is set to True.

  • Type: boolean

  • Default: False

auto.register.schemas

Specifies if the Serializer should attempt to register the Schema with Schema Registry.

  • Type: boolean

  • Default: true

  • Importance: medium

use.latest.version

Only applies when auto.register.schemas is set to false. If auto.register.schemas is set to false and use.latest.version is set to true, then instead of deriving a schema for the object passed to the client for serialization, Schema Registry uses the latest version of the schema in the subject for serialization.

  • Type: boolean

  • Default: true

  • Importance: medium