摘要

当我和游戏制作人进行流媒体时

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

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

/数据/示例\u主题\u 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

其次，我启动console consumer（也内置于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"}

但我从我的FileStreamLink消费者那里得到了这样的信息：

/数据/示例\u主题\u 1.txt

{id=1337, status=example_topic_1 success}

我的配置

配置/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\主题\ 1.properties

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