AWS MSK Kafka Connect S3接收器连接器

Question

我们使用S3接收器连接器从MSK自动气象站的S3桶中接收数据。

我们已经在AWS EKS(Kubernetes)上部署了KafkaS3水槽连接器

当我们启动连接器时，当 S3 存储桶上发生分段上传时出现以下错误。

我们对S3存储桶有策略限制，因为启用了服务器端加密（AWS-KMS），即如果没有KMS密钥，我们无法上传。

下面是我们用于连接器的配置，下面是错误详细信息，供您参考。

好心帮忙

{"name"："kc-s3-nuoAcCountIDs-Sink"，"connector.class"："io.confluent.connect.s3. S3Sink连接器"，"主题"："测试主题"，"tasks.max"："1"，"s3.bucket.name"："s3-But-name"，"value.converter.schemas.enable"："false"，"storage.class"："io.confluent.connect.s3.storage.S3存储"，"format.class"："io.confluent.connect.s3.format.avro.AvroFormat"，"partitioner.class"："io.confluent.connect.storage.partitioner.默认分区器"，"behavior.on.null.values"："忽略"，"schema.compatibility"："NONE"，"partition.duration.ms"： 3600000，"path.format"："'年'=YYYY/'月'=MM/'日'=dd/'小时'=HH"，"语言环境"："en-US"，"时区"："UTC"，"partition.field.name"："帐户平台"，"s3.region"："eu-West-2"，"flush.size"： 100000}

ktask . execute(worker sink task . Java:201)\ n \ tat org . Apache . Kafka . connect . runtime . worker task . dorun(worker task . Java:185)\ n \ tat org . Apache . Kafka . connect . runtime . worker task . run(worker task . Java:235)\ n \ tat Java . base/Java . util . concurrent . executors＄runnable adapter . call(executors . Java:515)\ \ tat Java . base/Java .

Answer 1

在我们的例子中，我们需要传递 S3 连接器的 KMS 密钥设置。

根据官方文档以及上述设置，我们在 S3 连接器配置中使用了以下两个附加设置：

"s3.sse.kms.key.id": "<kms-key-id-here>",
"s3.ssea.name": "aws:kms"

我们现在能够在 s3 存储桶中获取数据。