在Docker/ECS运行Kafka Connect S3 Sink无数据写入S3求助

环境与配置 #

我在Elastic Container Service（ECS）上部署Kafka Connect搭配S3 Sink连接器，用于从Kafka读取数据写入S3，相关配置如下：

Dockerfile #

FROM confluentinc/cp-kafka-connect-base:latest
WORKDIR /etc/kafka-connect
COPY entrypoint.sh .
COPY kafka-connect.properties .
COPY s3-sink.properties .
RUN confluent-hub install --no-prompt confluentinc/kafka-connect-s3:latest
ENV CONNECT_PLUGIN_PATH=/usr/share/java,/etc/kafka-connect/jars
CMD /etc/kafka-connect/entrypoint.sh

entrypoint.sh（已知存在变量错误） #

PROPERTIES_FILE="/etc/kafka-connect/kafka-connect.properties"
S3_SINK_PROPERTIES_FILE="/etc/kafka-connect/s3-sink.properties"
sed -i "s|KAFKA_USERNAME|$KAFKA_USERNAME|g" $PROPERTIES_FILE
sed -i "s|KAFKA_PASSWORD|$KAFKA_PASSWORD|g" $PROPERTIES_FILE

# 错误：误用未定义的$CONSOLE_SINK_PROPERTIES_FILE，应为$S3_SINK_PROPERTIES_FILE
sed -i "s|AWS_ACCESS_KEY_ID|$AWS_ACCESS_KEY_ID|g" $CONSOLE_SINK_PROPERTIES_FILE
sed -i "s|AWS_SECRET_ACCESS_KEY|$AWS_SECRET_ACCESS_KEY|g" $CONSOLE_SINK_PROPERTIES_FILE

/usr/bin/connect-distributed $PROPERTIES_FILE $S3_SINK_PROPERTIES_FILE

kafka-connect.properties 配置 #

name=abc
connector.class=io.confluent.connect.s3.S3SinkConnector
tasks.max=1
topics=topicName

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.protobuf.ProtobufConverter
group.id=myGroupId
bootstrap.servers=bootstrapServer:9091,...
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="KAFKA_USERNAME" password="KAFKA_PASSWORD";
sasl.mechanism=PLAIN

key.converter.schemas.enable=false
value.converter.schemas.enable=true

value.converter.basic.auth.credentials.source=USER_INFO
value.converter.value.subject.name.strategy=io.confluent.kafka.serializers.subject.TopicNameStrategy
schema.registry.basic.auth.credentials.source=USER_INFO
key.converter.basic.auth.credentials.source=USER_INFO

offset.storage.topic=test-offsets-topic
config.storage.topic=test-config-topic
status.storage.topic=test-status-topic

注：KAFKA_USERNAME与KAFKA_PASSWORD通过entrypoint.sh中的环境变量替换

s3-sink.properties 配置 #

format.class=io.confluent.connect.s3.format.json.JsonFormat
s3.bucket.name=bucket-name
s3.region=eu-west-2
s3.part.size=5242880
aws.access.key.id=AWS_ACCESS_KEY_ID
aws.secret.access.key=AWS_SECRET_ACCESS_KEY
storage.class=io.confluent.connect.s3.storage.S3Storage

注：AWS_ACCESS_KEY_ID与AWS_SECRET_ACCESS_KEY通过entrypoint.sh中的环境变量替换

问题现象 #

连接器启动后，Kafka主题有大量数据，但没有内容写入S3存储桶；CloudWatch日志无错误信息，仅出现以下提示：

[2024-01-09 23:10:27,086] INFO These configurations '[connector.class, config.storage.topic, tasks.max, topics, group.id, status.storage.topic, value.converter.value.subject.name.strategy, value.converter.basic.auth.credentials.source, key.converter.schemas.enable, schema.registry.basic.auth.credentials.source, name, value.converter.schemas.enable, offset.storage.topic, value.converter, key.converter, key.converter.basic.auth.credentials.source]' were supplied but are not used yet. (org.apache.kafka.clients.admin.AdminClientConfig:378)

问题排查与修复方案 #

1. 修复entrypoint.sh的变量错误 #

先解决已知的变量误用问题：将$CONSOLE_SINK_PROPERTIES_FILE替换为$S3_SINK_PROPERTIES_FILE，否则AWS密钥无法正确注入到S3连接器配置中，导致无法访问S3存储桶。

修改后的entrypoint.sh相关片段：

sed -i "s|AWS_ACCESS_KEY_ID|$AWS_ACCESS_KEY_ID|g" $S3_SINK_PROPERTIES_FILE
sed -i "s|AWS_SECRET_ACCESS_KEY|$AWS_SECRET_ACCESS_KEY|g" $S3_SINK_PROPERTIES_FILE

2. 分离Kafka Connect集群配置与连接器配置 #

connect-distributed命令的参数只能是集群配置文件，不能直接传入连接器配置文件。分布式模式下，连接器配置需要通过Kafka Connect的REST API提交，或存储在config.storage.topic中。

正确的启动命令：

/usr/bin/connect-distributed $PROPERTIES_FILE

同时，将连接器专属配置（connector.class、tasks.max、topics以及S3相关配置）从kafka-connect.properties中移除，合并到s3-sink.properties，之后通过REST API提交：

# 示例：通过REST API创建连接器
curl -X POST -H "Content-Type: application/json" --data @s3-sink.properties http://<connect-host>:8083/connectors

3. 检查存储主题的可用性 #

日志提示配置未被使用，说明Kafka Connect可能无法正常访问offset.storage.topic、config.storage.topic、status.storage.topic：

• 确保这三个主题已提前创建，且配置足够的分区数和副本数（建议每个主题至少3个分区、3个副本）
• 确认Kafka Connect服务账号有权限读写这些主题

4. 补充缺失的Schema Registry配置 #

使用Protobuf转换器时，kafka-connect.properties缺少schema.registry.url配置，会导致转换器无法获取Protobuf Schema，进而无法解析消息。需添加：

schema.registry.url=http://<schema-registry-host>:8081

同时确保Schema Registry的认证信息配置正确（若开启认证）。

5. 验证S3权限 #

确认ECS任务角色或注入的AWS密钥拥有S3存储桶的PutObject、ListBucket权限，避免因权限不足导致数据无法写入。

内容的提问来源于stack exchange，提问作者kk1957