问题标签 [apache-kafka-connect]

以 JSON 格式从 Kafka 生产/消费。使用以下属性以 JSON 格式保存到 HDFS：

制片人：

消费者 ：

问题一：

获得例外：

问题 2：

启用以上两个属性不会引发任何问题，但不会通过 hdfs 写入数据。

任何建议将不胜感激。

谢谢

我有一个生产者正在为某个主题生成 protobuf 消息。我有一个反序列化 protobuf 消息的消费者应用程序。但是 hdfs sink 连接器直接从 Kafka 主题中获取消息。键和值转换器etc/schema-registry/connect-avro-standalone.properties将设置为什么？最好的方法是什么？提前致谢！

poll()对于SourceTask 实现中的方法，我有一个带有以下代码的 Kafka 连接器。

如果我将消费者附加到数据主题，我会收到从连接器发送的以下消息：

如果我使用以下命令直接向主题发布消息：

然后附加相同的消费者，我收到以下消息：

这是我期望看到的连接器实现的输出，而不是{"schema":...我收到的消息。

如何更改实现，poll()以便在消息的实际键和值中不出现架构元数据的情况下发送消息？

社区，请您帮我理解为什么我的大约 3%的消息没有以HDFS. 我写了一个简单的生产者JAVA来生成 1000 万条消息。

我使用Confluent HDFS Connector独立模式将数据写入HDFS. 配置如下：

连接器将数据写入 HDFS，但在等待 20000 毫秒（由于rotate.interval.ms）后，并未收到所有消息。

知道这种行为的原因是什么吗？我的错误在哪里？我正在使用Confluent 3.0.1/Kafka 10.0.0.1。

来自 kafka 主题的消息未写入 hdfs。

原因 ：

谁能解释一下。

我正在努力从 Camus 迁移到 Kafka Connect。我正在研究 Kafka Connect 的实现，特别关注分布式模式。我能够在我假设与我的 Kafka 集群通信的本地机器上成功启动一个工作程序。我还能够运行两个 GET 命令，例如/返回/connector-plugins正确的 JSON。但是，当我尝试发布创建连接器的命令时，我收到 500 错误和超时。具体来说，我使用此命令 POST 进行测试：

curl -X POST -H "Content-Type: application/json" --data '{"name": "local-file-sink", "config": {"connector.class":"FileStreamSinkConnector", "tasks.max":"1", "file":"test.sink.txt", "topics":"myTopic" }}' localhost:8083/connectors

最终我得到了这个回应：

{"error_code": 500, "message": "Request timed out"}

我不知道发生了什么。我的 Kafka Connect 分布式工作人员的日志显示了这一点：

您可以在其中查看错误代码和命令。

我想我的主要问题和问题是：

1. 我怎样才能更好地调试 Kafka Connect，以便我可以尝试解决这个问题？
2. 我正在做的事情有什么明显错误的吗？
3. 是否有关于让 Kafka Connect 分布式工作程序和连接器运行的分步文档或博客文章？我还没有真正看到任何甚至最佳实践类型的文档？也许我只是太早的采用者。

我期待收到社区的回复，并感谢您的帮助！

我在本地“启动”Docker 容器（独立于 Kafka 节点容器）中以分布式模式启动 Kafka 连接器。连接器按预期工作，但是当我终止启动容器时，连接器停止工作。我希望它能够继续工作，因为我相信它是在另一个容器中的 Kafka 节点上的工作人员上注册和运行的。我更详细的设置如下：

目前，我正在本地通过 Docker 容器运行所有内容。我有：

1. Zookeeper 节点 (3.4.9)
2. 一个 Kafka 节点（Apache，0.10.1.0）
3. 一个“启动”节点。

启动节点下载适当的 Kafka 版本并解压缩其内容。然后它构建连接器源，设置类路径以包含必要的 JAR，然后执行连接器，如下所示：

分布式属性文件设置组 ID、各种主题名称、模式和转换器以及引导服务器（指向上面的 Kafka 节点 (2)）。

此命令似乎执行正确，restful 连接器 http 服务已成功启动。然后我可以向http://example:8083/connectors发出 POST 请求，为连接器任务提供配置。命令完成且没有错误，连接器已成功启动。我可以从 Kafka 节点 (2) 中的主题消费，我看到的输出表明连接器正在工作并通过它发送数据。

当我终止启动节点 (3) 时，我希望连接器能够继续运行，因为我在 Kafka 集群中注册了它，尽管它是一个集群。连接器不会继续运行，并且似乎与启动节点一起死亡。连接器现在不应该由集群中的工作人员管理吗？我是否需要更改启动连接器的方式或者我误解了什么？

通常出于调试目的，我想分析/检查 Kafka 主题中的消息。

我相信我们可以编写连接器来将数据从 Kafka Topics 导出到 MySql DB。

在深入研究之前，我想确保它是否是一种可行的方法？如果是这样，你能给我举个例子吗？

我是 kafka 的新手，我已经在本地安装了带有默认配置的 kafka 10。现在，我面临一个问题。我正在从控制台生产者那里生成消息。如果消息大约是 4096 字节。然后，它被控制台消费者很好地消费了。但是当我从 4096 字节增加消息大小时。然后，在使用它时，消息被截断到大约 4096 字节。我没有得到问题。

怎么了 ？a) 消息是否发布不完整。b) 消息是否被不完全消费。

注意：我没有对默认设置进行任何更改，我正在使用控制台生产者和消费者。

请任何人帮助