0

我们正在测试分布式模式下的 kafka connect，以将主题记录从 kafka 拉到 HDFS。我们有两个盒子。kafka 和 zookeeper 守护进程正在其中运行的一种。我们在此框中保留了一个 kafka connect 实例。我们有另一个盒子，其中存在 HDFS 名称节点。我们在这里保留了另一个 kafka connect 实例。

我们在第一个盒子中启动了 kafka、zookeeper 和 kafka connect。我们也在第二个盒子里开始了 kafka connect。现在根据 confluent 文档，我们必须使用 REST API 启动 HDFS 连接器（或任何其他连接器）。因此，在这两个框中启动 kafka connect 后，我​​们尝试通过 REST API 启动连接器。我们尝试了以下命令：-

curl -X POST -H "HTTP/1.1 Host: ip-10-16-34-57.ec2.internal:9092 Content-Type: application/json Accept: application/json" --data '{"name": "hdfs-sink", "config": {"connector.class":"io.confluent.connect.hdfs.HdfsSinkConnector", "format.class":"com.qubole.streamx.SourceFormat", "tasks.max":"1", "hdfs.url":"hdfs://ip-10-16-37-124:9000", "topics":"Prd_IN_TripAnalysis,Prd_IN_Alerts,Prd_IN_GeneralEvents", "partitioner.class":"io.confluent.connect.hdfs.partitioner.DailyPartitioner", "locale":"", "timezone":"Asia/Calcutta" }}' http://ip-10-16-34-57.ec2.internal:8083/connectors

只要我们在这里按回车，我们就会得到以下响应：

    <html>
    <head>
    <meta http-equiv="Content-Type" content="text/html;charset=ISO-8859-1"/>
    <title>Error 415 </title>
    </head>
    <body>
    <h2>HTTP ERROR: 415</h2>
    <p>Problem accessing /connectors. Reason:
    <pre>    Unsupported Media Type</pre></p>
    <hr /><i><small>Powered by Jetty://</small></i>
    </body>
    </html>

etc/kafka/ 的 connect-distributed.properties 文件位于两个 kafka 连接节点的下方。我们也创建了上述三个主题（connect-offsets、connect-configs、connect-status）

bootstrap.servers=ip-10-16-34-57.ec2.internal:9092
group.id=connect-cluster
key.converter=com.qubole.streamx.ByteArrayConverter
value.converter=com.qubole.streamx.ByteArrayConverter
enable.auto.commit=true
auto.commit.interval.ms=1000
offset.flush.interval.ms=1000
key.converter.schemas.enable=true
value.converter.schemas.enable=true
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.topic=connect-offsets
rest.port=8083
config.storage.topic=connect-configs
status.storage.topic=connect-status
offset.flush.interval.ms=10000

这里有什么问题？我们是否缺少一些东西来以分布式模式启动 kafka 连接以使用 HDFS 连接器。独立模式下的 kafka connect 工作正常。

4

1 回答 1

0

要上传连接器，这是一个 PUT 命令，而不是 POST：http ://docs.confluent.io/3.1.1/connect/restapi.html#put--connectors-(string-name)-config

附带说明一下，我相信您的 curl 命令可能是错误的：

  • 每个标头需要一个 -H 开关，将所有标头放在一个 -H 参数中不是它的工作方式（我认为）。
  • 我不认为该端口是 Host 标头的一部分。
于 2017-02-02T07:02:04.257 回答