Amazon S3 Storage (S3) sink connector for Apache Kafka®

Requirements

The connector requires Java 11 or newer for development and production.

File name format

File name format is tightly related to Record Grouping

The connector uses the following format for output files (blobs): <prefix><filename>.

<prefix>is the optional prefix that can be used, for example, for subdirectories in the bucket. <filename> is the file name. The connector has the configurable template for file names.

Configuration property `file.name.template`. If not set, default template is used: `{{topic}}-{{partition}}-{{start_offset}}`

It supports placeholders with variable names: {{ variable_name }}. Currently, supported variables are:

• topic - the Kafka topic;
• partition:padding=true|false - the Kafka partition, if padding set to true it will set leading zeroes for offset, the default value is false;
• start_offset:padding=true|false - the Kafka offset of the first record in the file, if padding set to true it will set leading zeroes for offset, the default value is false;
• timestamp:unit=yyyy|MM|dd|HH - the timestamp of when the Kafka record has been processed by the connector.
  • unit parameter values:
    • yyyy - year, e.g. 2020 (please note that YYYY is deprecated and is interpreted as yyyy)
    • MM - month, e.g. 03
    • dd - day, e.g. 01
    • HH - hour, e.g. 24
• key - the Kafka key.

To add zero padding to Kafka offsets, you need to add additional parameter padding in the start_offset variable, which value can be true or false (the default). For example: {{topic}}-{{partition}}-{{start_offset:padding=true}}.gz will produce file names like mytopic-1-00000000000000000001.gz.

To add zero padding to partition number, you need to add additional parameter padding in the partition variable, which value can be true or false (the default). For example: {{topic}}-{{partition:padding=true}}-{{start_offset}}.gz will produce file names like mytopic-0000000001-1.gz.

To add formatted timestamps, use timestamp variable.
For example: {{topic}}-{{partition}}-{{start_offset}}-{{timestamp:unit=yyyy}}{{timestamp:unit=MM}}{{timestamp:unit=dd}}.gz will produce file names like mytopic-2-1-20200301.gz.

To configure the time zone for the timestamp variable, use file.name.timestamp.timezone property. Please see the description of properties in the “Configuration” section.

Only the certain combinations of variables and parameters are allowed in the file name template (however, variables in a template can be in any order). Each combination determines the mode of record grouping the connector will use. Currently, supported combinations of variables and the corresponding record grouping modes are:

• topic, partition, start_offset, and timestamp - grouping by the topic, partition, and timestamp;
• key - grouping by the key.
• key, topic, partition - grouping by the topic, partition, and key.

See record grouping in the next section for more details.

If the file name template is not specified, the default value is {{topic}}-{{partition}}-{{start_offset}} (+ .gz when compression is enabled).

Record grouping

Incoming records are being grouped until flushed. The connector flushes grouped records in one file per offset.flush.interval.ms setting for partitions that have received new messages during this period. The setting defaults to 60 seconds.

Record grouping, similar to Kafka topics, has 2 modes:

• Changelog: Connector groups all records in the order received from a Kafka topic, and stores all of them in a file.
• Compact: Connector groups all records by an identity (e.g. key) and only keeps the latest value stored in a file.

Modes are defined implicitly by the fields used of the file name template.

topic:topicB partition:0 offset:0
topic:topicA partition:0 offset:0
topic:topicA partition:0 offset:1
topic:topicB partition:0 offset:1
flush

topic:topicA partition:0 offset:0
topic:topicA partition:0 offset:1
flush
topic:topicA partition:0 offset:2
topic:topicA partition:0 offset:3
flush

key:0 value:0
key:1 value:1
key:0 value:2
key:1 value:3
flush

key:0 value:0
key:1 value:1
key:0 value:2
key:1 value:3
flush
key:0 value:4
flush

topic:topicA partition:0 key:x value:aaa
topic:topicA partition:1 key:x value:bbb
flush

S3 Object Names

S3 connector stores series of files in the specified bucket. Each object is named using pattern [<aws.s3.prefix>]<topic>-<partition>-<start_offset>[.gz] (see [#file-name-format](File name format section) for more patterns). The .gz extension is used if gzip compression is used, see file.compression.type below.

Data File Format

Output files are text files that contain one record per line (i.e., they're separated by \n) except PARQUET format.

There are four types of data format available:

• [Default] Flat structure, where field values are separated by comma (csv)

  Configuration: format.output.type=csv. Also, this is the default if the property is not present in the configuration.

• Complex structure, where file is in format of JSON lines. It contains one record per line and each line is a valid JSON object(jsonl)

  Configuration: format.output.type=jsonl.

• Complex structure, where file is a valid JSON array of record objects.

  Configuration: format.output.type=json.

• Complex structure, where file is in Apache Parquet file format.

  Configuration: format.output.type=parquet.

The connector can output the following fields from records into the output: the key, the value, the timestamp, the offset and headers. (The set and the order of output: the key, the value, the timestamp, the offset and headers. The set of these output fields is configurable.) The field values are separated by comma.

a2V5,TG9yZW0gaXBzdW0gZG9sb3Igc2l0IGFtZXQ=,1232155,1554210895

,,,1554210895

 { "key": "k1", "value": "v0", "offset": 1232155, "timestamp":"2020-01-01T00:00:01Z" }

  { "key": "user1", "value": {"name": "John", "address": {"city": "London"}}, "offset": 1232155, "timestamp":"2020-01-01T00:00:01Z" }

[
  { "key": "k1", "value": "v0", "offset": 1232155, "timestamp":"2020-01-01T00:00:01Z" },
  { "key": "k2", "value": "v1", "offset": 1232156, "timestamp":"2020-01-01T00:00:05Z" }
]

[
  { "key": "user1", "value": {"name": "John", "address": {"city": "London"}}, "offset": 1232155, "timestamp":"2020-01-01T00:00:01Z" }
]

{ "value": "v0" }

{ "value": {"name": "John", "address": {"city": "London"}} }

"v0"

{"name": "John", "address": {"city": "London"}}

{
    "type": "record", "fields": [
      {"name": "key", "type": "RecordKeySchema"},
      {"name": "offset", "type": "long"},
      {"name": "timestamp", "type": "long"},
      {"name": "headers", "type": "map"},
      {"name": "value", "type": "RecordValueSchema"}
  ]
}

{
  "type": "string"
}

{
    "type": "record", "fields": [
      {"name": "foo", "type": "string"},
      {"name": "bar", "type": "long"}
  ]
}

{
    "type": "record", "fields": [
      {"name": "key", "type": "string"},
      {"name": "offset", "type": "long"},
      {"name": "timestamp", "type": "long"},
      {"name": "headers", "type": "map", "values": "long"},
      { "name": "value",
        "type": "record",
        "fields": [
          {"name": "foo", "type": "string"},
          {"name": "bar", "type": "long"}
        ]
      }
  ]
}

{ "value": {"name": "John", "address": {"city": "London"}} }

{"name": "John", "address": {"city": "London"}}

Connector Configuration

Important Note Since version 2.6 all existing configuration is deprecated and will be replaced with new one during a certain transition period (within 2-3 releases)

List of deprecated configuration parameters:

• aws_access_key_id - AWS Access Key ID for accessing S3 bucket. Mandatory.
• aws_secret_access_key - AWS S3 Secret Access Key. Mandatory.
• aws_s3_endpoint - The endpoint configuration (service endpoint & signing region) to be used for requests.
• aws_s3_region - Name of the region for the bucket used for storing the records. Defaults to us-east-1.
• aws_s3_bucket - Name of an existing bucket for storing the records. Mandatory.
• aws_s3_prefix - The prefix that will be added to the file name in the bucket. Can be used for putting output files into a subdirectory.
• output_compression- Compression type for output files. Supported algorithms are gzip, snappy, zstd and none. Defaults to gzip.
• output_fields - A comma separated list of fields to include in output. Supported values are: key, offset, timestamp and value. Defaults to value.

List of new configuration parameters:

• aws.access.key.id - AWS Access Key ID for accessing S3 bucket.
• aws.secret.access.key - AWS S3 Secret Access Key.
• aws.s3.bucket.name - - Name of an existing bucket for storing the records. Mandatory. See bucket name rules: https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html
• aws.s3.endpoint - The endpoint configuration (service endpoint & signing region) to be used for requests.
• aws.s3.prefix - [Deprecated] Use file.name.prefix and file.name.template instead. The prefix that will be added to the file name in the bucket. Can be used for putting output files into a subdirectory.
• aws.s3.region - Name of the region for the bucket used for storing the records. Defaults to us-east-1.
• aws.sts.role.arn - AWS role ARN, for cross-account access role instead of aws.access.key.id and aws.secret.access.key
• aws.sts.role.external.id - AWS ExternalId for cross-account access role
• aws.sts.role.session.name - AWS session name for cross-account access role
• aws.sts.role.session.duration - Session duration for cross-account access role in Seconds. Minimum value - 900.
• aws.sts.config.endpoint - AWS STS endpoint for cross-account access role.
• file.name.template - The file name. The connector has the configurable template for file names. Constant string prefix could be added to the file name to put output files into a subdirectory.
• file.compression.type - Compression type for output files. Supported algorithms are gzip, snappy, zstd and none. Defaults to gzip.
• format.output.fields - A comma separated list of fields to include in output. Supported values are: key, offset, timestamp, value and headers. Defaults to value.
• format.output.fields.value.encoding - Controls encoding of value field. Possible values are: base64 and none. Defaults: base64
• timestamp.timezone - The time zone in which timestamps are represented. Accepts short and long standard names like: UTC, PST, ECT, Europe/Berlin, Europe/Helsinki, or America/New_York. For more information please refer to https://docs.oracle.com/javase/tutorial/datetime/iso/timezones.html. The default is UTC.
• timestamp.source - The source of timestamps. Supports only wallclock which is the default value.

S3 multi-part uploads

To configure S3 multi-part uploads buffer size change:

• aws.s3.part.size.bytes - The Part Size in S3 Multi-part Uploads in bytes. Maximum is 2GB and default is 5MB.

Retry strategy configuration

There are four configuration properties to configure retry strategy exists.

AWS S3 server side encryption properties

• aws.s3.sse.algorithm - The name of the Server-side encryption algorithm to use for uploads. If unset the default SSE-S3 is used.
• To use SSE-S3 set to AES256 or leave empty
• To use SSE-KMS set to aws:kms
• To use DSSE-KMS set to aws:kms:dsse

Cleaning temporary files from failed multipart uploads

The S3 Sink Connector uploads files using the S3 multipart upload API for improved performance and handling large files. Occasionally the API can throw an exception or the connector can fail to complete a multipart upload. This can leave orphaned “parts” of a failed multipart upload taking up unnecessary space. To handle these incomplete parts AWS recommends setting up a Lifecycle rule to delete old parts that weren't completed as described in this excellent (blog post)[https://aws.amazon.com/blogs/aws-cloud-financial-management/discovering-and-deleting-incomplete-multipart-uploads-to-lower-amazon-s3-costs/]. Alternatively, if you would prefer to work through the official documentation it is available (here)[https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpu-abort-incomplete-mpu-lifecycle-config.html]

Developing together with Commons library

This project depends on Common Module for Apache Kafka Connect.

Integration testing

Integration tests are implemented using JUnit, Gradle and Docker.

To run them, you need:

• Docker installed.

Integration testing doesn't require valid AWS credentials.

To simulate AWS S3 behaviour, tests use LocalStack.

In order to run the integration tests, execute from the project root directory:

./gradlew clean integrationTest