Aiven-Open / Aiven-Open/cloud-storage-connectors-for-apache-kafka
S3 Source Adapter isn't compatible with the S3 Sink Adapter
- 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
Bewertung
Dieses Issue wurde noch nicht bewertet.