Aiven-Open / Aiven-Open/cloud-storage-connectors-for-apache-kafka
S3 Source Adapter isn't compatible with the S3 Sink Adapter
- 主要語言
- Java
- 星號
- 58
- 分支
- 39
- 平均合併
- 2 天 10 小時
- 30 天內合併 PR
- 5
描述
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!
貢獻指南
評估
這個 Issue 還沒有評估資料。