【问题标题】：Kafka Connect FileStreamSink connector removes quotation marks and changes colon to equal sign for JSON messageKafka Connect FileStreamSink 连接器删除引号并将 JSON 消息的冒号更改为等号
【发布时间】：2019-06-28 16:29:06
【问题描述】：

总结

当我与控制台制作人进行流式传输时

{"id":1337,"status":"example_topic_1 success"}

我从我的文件流消费者那里得到这个

/data/example_topic_1.txt

{id=1337, status=example_topic_1 success}

这对我来说是一个主要问题，因为如果不对引号过去的位置做出假设，就无法恢复原始 JSON 消息。如何在保留引号的同时将消息输出到文件？

详情

首先，我启动我的文件接收器连接器。

# sh bin/connect-standalone.sh \
>   config/worker.properties \
>   config/connect-file-sink-example_topic_1.properties

其次，我启动控制台使用者（也内置于 Kafka），以便我可以轻松地直观地确认消息是否正确通过。
```
# sh bin/kafka-console-consumer.sh \
>   --bootstrap-server kafka_broker:9092 \
>   --topic example_topic_1
```
最后，我启动一个用于发送消息的控制台生产者，并输入一条消息。
```
# sh bin/kafka-console-producer.sh \
>   --broker-list kafka_broker:9092 \
>   --topic example_topic_1
```
来自控制台消费者的消息正确弹出，带有引号。
```
{"id":1337,"status":"example_topic_1 success"}
```
但我是从我的 FileStreamSink 消费者那里得到的：

/data/example_topic_1.txt
```
{id=1337, status=example_topic_1 success}
```

我的配置

config/worker.properties

offset.storage.file.filename=/tmp/example.offsets

bootstrap.servers=kafka_broker:9092
offset.flush.interval.ms=10000

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

config/connect-file-sink-example_topic_1.properties

name=file-sink-example_topic_1
connector.class=FileStreamSink
tasks.max=1
file=/data/example_topic_1.txt
topics=example_topic_1

【问题讨论】：

使用 StringConverter，而不是 JsonConverter

标签： apache-kafka apache-kafka-connect

【解决方案1】：

由于您实际上并不想解析 JSON 数据，而只是将其作为文本块直接传递，因此您需要使用 StringConverter：

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter

本文详细解释了转换器的细微差别：https://rmoff.net/2019/05/08/when-a-kafka-connect-converter-is-not-a-converter/。尽管使用kafkacat 代替控制台生产者/消费者，这显示了您尝试执行的操作的示例。

【讨论】：