Aiven-Open / Aiven-Open/cloud-storage-connectors-for-apache-kafka

S3 Source Adapter isn't compatible with the S3 Sink Adapter

Offen
#593 1 Kommentar 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen
Vorherrschende Sprache
Java
Sterne
58
Forks
39
Ø Merge
2 T. 10 Std.
Gemergte PRs (30 T.)
5

Beschreibung

After reading the documentation several times and trying out various configurations, I get the impression that the S3 Source Adapter isn't compatible with the S3 Sink Adapter. Or may I have the wrong expectations for these two connectors?

I assumed that it was possible to create a backup of Kafka topics using the Aiven Open S3 Sink/Source connectors.
However, I can't seem to restore the backed-up messages with key, header, value, timestamp, and offset.

Is there a sample configuration that ensures this?

Here is my configuration for the sink.

```
class: io.aiven.kafka.connect.s3.AivenKafkaConnectS3SinkConnector
config:
aws.access.key.id: ${env:AWS_ACCESS_KEY_ID}
aws.secret.access.key: ${env:AWS_SECRET_ACCESS_KEY}
aws.s3.endpoint: https://xxxxx
aws.s3.bucket.name: test-xxxxx

topics: flights

file.name.template: aiven-s3-sink/{{topic}}/{{partition:padding=true}}-{{start_offset:padding=true}}.jsonl
file.compression.type: none

format.output.fields: key,value,headers,offset,timestamp
format.output.fields.value.encoding: none
format.output.type: jsonl
format.output.envelope: true

key.converter: org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable: false
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: false
header.converter: org.apache.kafka.connect.json.JsonConverter
header.converter.schemas.enable: false

timestamp.timezone: Europe/Berlin
timestamp.source: wallclock
```

The messages of my test topic with dummy values creates the following jsonl file.

```json
{"headers":[],"offset":0,"value":{"AIR_TIME":350,"ARR_TIME":12.483334,"DISTANCE":2475,"FL_DATE":"2006-01-01","DEP_TIME":9.083333,"DEP_DELAY":5,"ARR_DELAY":19},"key":null,"timestamp":"2026-04-01T09:40:31.861Z"}
{"headers":[],"offset":1,"value":{"AIR_TIME":343,"ARR_TIME":15.766666,"DISTANCE":2475,"FL_DATE":"2006-01-02","DEP_TIME":11.783334,"DEP_DELAY":167,"ARR_DELAY":216},"key":null,"timestamp":"2026-04-01T09:40:31.861Z"}
{"headers":[],"offset":2,"value":{"AIR_TIME":344,"ARR_TIME":12.133333,"DISTANCE":2475,"FL_DATE":"2006-01-03","DEP_TIME":8.883333,"DEP_DELAY":-7,"ARR_DELAY":-2},"key":null,"timestamp":"2026-04-01T09:40:31.861Z"}
{"headers":[],"offset":3,"value":{"AIR_TIME":331,"ARR_TIME":11.95,"DISTANCE":2475,"FL_DATE":"2006-01-04","DEP_TIME":8.916667,"DEP_DELAY":-5,"ARR_DELAY":-13},"key":null,"timestamp":"2026-04-01T09:40:31.861Z"}
{"headers":[],"offset":4,"value":{"AIR_TIME":321,"ARR_TIME":11.883333,"DISTANCE":2475,"FL_DATE":"2006-01-05","DEP_TIME":8.95,"DEP_DELAY":-3,"ARR_DELAY":-17},"key":null,"timestamp":"2026-04-01T09:40:31.861Z"}
{"headers":[],"offset":5,"value":{"AIR_TIME":320,"ARR_TIME":11.633333,"DISTANCE":2475,"FL_DATE":"2006-01-06","DEP_TIME":8.933333,"DEP_DELAY":-4,"ARR_DELAY":-32},"key":null,"timestamp":"2026-04-01T09:40:31.861Z"}
{"headers":[],"offset":6,"value":{"AIR_TIME":346,"ARR_TIME":12.133333,"DISTANCE":2475,"FL_DATE":"2006-01-08","DEP_TIME":8.95,"DEP_DELAY":-3,"ARR_DELAY":-2},"key":null,"timestamp":"2026-04-01T09:40:31.861Z"}
{"headers":[],"offset":7,"value":{"AIR_TIME":334,"ARR_TIME":12.166667,"DISTANCE":2475,"FL_DATE":"2006-01-09","DEP_TIME":9.05,"DEP_DELAY":3,"ARR_DELAY":0},"key":null,"timestamp":"2026-04-01T09:40:31.861Z"}
{"headers":[],"offset":8,"value":{"AIR_TIME":334,"ARR_TIME":11.816667,"DISTANCE":2475,"FL_DATE":"2006-01-10","DEP_TIME":8.883333,"DEP_DELAY":-7,"ARR_DELAY":-21},"key":null,"timestamp":"2026-04-01T09:40:31.861Z"}
{"headers":[],"offset":9,"value":{"AIR_TIME":321,"ARR_TIME":12,"DISTANCE":2475,"FL_DATE":"2006-01-11","DEP_TIME":9.133333,"DEP_DELAY":8,"ARR_DELAY":-10},"key":null,"timestamp":"2026-04-01T09:40:31.861Z"}
```

The only thing I've managed to do so far is restore messages using an invalid key and the JSON envelope in the value. The offset and timestamp are assigned sequentially at the time of import.

What would a source connector need to look like to create messages containing key, header, value, timestamp, and offset from that jsonl file?

Thanks!

Beitragsleitfaden

Beitragsleitfaden öffnen

Bewertung

Dieses Issue wurde noch nicht bewertet.

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.