Kafka连接问题

Question

我在CentOS7(confluent)上安装了Apache Kafka，正试图以分布式模式运行filestream Kafka connect，但收到以下错误：

[2017-08-10 05:26:27,355] INFO Added alias 'ValueToKey' to plugin 'org.apache.kafka.connect.transforms.ValueToKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:290)
Exception in thread "main" org.apache.kafka.common.config.ConfigException: Missing required configuration "internal.key.converter" which has no default value.
at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:463)
at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:453)
at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:62)
at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:75)
at org.apache.kafka.connect.runtime.WorkerConfig.<init>(WorkerConfig.java:197)
at org.apache.kafka.connect.runtime.distributed.DistributedConfig.<init>(DistributedConfig.java:289)
at org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:65)

现在可以通过更新workers.properties（如http://docs.confluent.io/current/connect/userguide.html#connect-userguide-distributed-config所述）来解决这个问题

使用的命令:

/home/arun/kafka/confluent-3.3.0/bin/connect-distributed.sh ../../../properties/file-stream-demo-distributed.properties

FILESTREAM属性文件(workers.properties):

name=file-stream-demo-distributed
connector.class=org.apache.kafka.connect.file.FileStreamSourceConnector
tasks.max=1
file=/tmp/demo-file.txt
bootstrap.servers=localhost:9092,localhost:9093,localhost:9094
config.storage.topic=demo-2-distributed
offset.storage.topic=demo-2-distributed
status.storage.topic=demo-2-distributed
key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter.schemas.enable=false
group.id=""

bootstrap.servers=localhost:9092,localhost:9093,localhost:9094
config.storage.topic=demo-2-distributed
offset.storage.topic=demo-2-distributed
status.storage.topic=demo-2-distributed
group.id=""

kafka-console-consumer --zookeeper localhost:2181 --topic demo-2-distributed --from-beginning

我想我错过了一些最基本的东西。谁能帮忙吗？

Answer 1

您需要为Kafka connect框架定义唯一的主题以存储其配置、偏移量和状态。

在workers.properties文件中，将这些参数更改为如下所示：

config.storage.topic=demo-2-distributed-config
offset.storage.topic=demo-2-distributed-offset
status.storage.topic=demo-2-distributed-status

这些主题用于存储connect的状态和配置元数据，而不用于存储connect上运行的任何连接器的消息。不要在这三个主题中的任何一个上使用控制台消费者，并期望看到消息。

{
  "name": "MyFileSink",
  "config": {
      "topics": "mytopic",
      "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
      "tasks.max": 1,
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "value.converter": "org.apache.kafka.connect.storage.StringConverter",
      "file": "/tmp/demo-file.txt"
    }
}

分布式工作程序运行后，您需要使用curl将配置文件应用到它，如下所示：

curl -X POST -H "Content-Type: application/json" --data @file-sink-config.json http://localhost:8083/connectors

之后，配置将安全地存储在您为所有分布式工作者创建的配置主题中。确保config主题（以及status和offset主题）不会使消息过期，否则会使连接器配置松散。