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

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

未關閉
#593 1 則留言 0 個 reaction 已指派 0 人 在 GitHub 檢視
主要語言
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 還沒有評估資料。

把新 issue 寄到你的電子郵件信箱

精選適合新手參與的 GitHub issue 摘要。