问题标签 [node-kafka]

我正在尝试在打字稿项目中使用 consumerGroup 而不是 HighLevelConsumer，但我无法为 consumerGroup 配置 consumerOptions。

分配时出现以下错误

1) fromOffset: '最早的'

'{ autoCommit: boolean; 类型的参数 fetchMaxBytes：数字；fetchMaxWaitMs：数字；从偏移：字符串；groupId...' 不可分配给“ConsumerGroupOptions”类型的参数。属性“fromOffset”的类型不兼容。类型 'string' 不可分配给类型 '"earliest" | “最新” | “没有任何”'。

2）协议：['roundrobin']

'{ autoCommit: boolean; 类型的参数 fetchMaxBytes：数字；fetchMaxWaitMs：数字；协议：字符串[]；groupId...' 不可分配给“ConsumerGroupOptions”类型的参数。属性“协议”的类型不兼容。类型 'string[]' 不能分配给类型 '("roundrobin" | "range" | CustomPartitionAssignmentProtocol)[]'。类型 'string' 不可分配给类型 '"roundrobin" | “范围” | CustomPartitionAssignmentProtocol'。

3) outOfRangeOffset: '最早的'

'{ autoCommit: boolean; 类型的参数 fetchMaxBytes：数字；fetchMaxWaitMs：数字；outOfRangeOffset：字符串；g...' 不可分配给“ConsumerGroupOptions”类型的参数。属性“outOfRangeOffset”的类型不兼容。类型 'string' 不可分配给类型 '"earliest" | “最新” | “没有任何”'。

环境细节：

• 节点版本：v8.11.1
• 卡夫卡节点版本：v2.6.1
• 类型/kafka 节点：v2.0.6

谁能帮我理解我哪里出错了？提前致谢。

我正在使用 Kafka Node 库，并测试高级生产者。

我创建了一个包含 10 个分区的主题“HLPTestInput”，并编写了一个函数以每秒生成一次。

生产者写入分区 0、2、4、6 和 8，但不写入奇数分区。

奇怪的是，当我从这个主题消费并生产到第二个主题“HLPTestInputFromConsumer”时，它有 5 个分区，消息被写入所有分区。

有没有我缺少的配置？

有一个 nodejs Kafka 生产者将文件内容发送到 Kafka。当 Kafka 消费者使用来自 Kafka 的消息时，它看起来像 -

当我m.message.value.toString('utf8')在 nodejs Kafka 消费者中使用时，它会打印实际消息。但我需要在 java Kafka 消费者中消费。我尝试过 property.put("value.serializer.encoding", "utf8")但new String(consumerRecord.value())仍然打印 { "type":"Buffer","data":[91,13 .....]。我的问题是如何在 java 中使用由 nodejs Kafka 生产者生成的字符串格式的消息。

我将 Kafka 用于队列，Node 服务使用Kafka-Node生成和消费到 Kafka 主题的消息。

我一直在使用自制的分布式跟踪解决方案，但现在我们正在转向 Elastic APM。

这似乎是为 HTTP 服务器量身定做的，但是如何配置它以与 Kafka 一起使用？

我希望能够跟踪如下事务：服务 A 向服务 B 发送 HTTP 请求，服务 B 将其生成到 Kafka 主题 C，服务 D 从中使用它，服务 D 将一些数据放入 Kafka 主题 E，从中它由服务 B 使用。

我正在使用 npm 包 kafka-node 版本 3.0.1。

但是，我在消息中收到垃圾字符 -

任何帮助表示赞赏。

我正在使用 kafka-node 模块向 kafka 发送消息。在集群环境中，我有一个具有 3 个分区和复制因子为 3 的主题。

主题描述是 -

生产者配置 -

当我发送数据时，它遵循分区类型为循环（2），如循环方式

当我按照以下步骤操作时

• 获取连接到kafka:9092,kafka:9093的HighLevelProducer实例
• 发送一个消息
• 手动停止 kafka-server:9092
• 尝试使用 HighLevelProducer 发送另一条消息，并且 send() 将触发回调并出现错误：TimeoutError: Request timed out after 30000ms

我期望的是，如果一个分区不可访问（因为代理关闭），生产者应该自动将数据发送到下一个可用分区，但由于异常我丢失了消息

例外情况如下 -

我有一个在由 node-kafka 提供支持的 node.js 上运行的 kafka 消费者组。当这个消费者组处于活动或非活动状态时，我希望看到 kafa-consumer-groups CLI 报告它。

kafka-consumer-groups CLI 确实显示了控制台消费者，而不仅仅是节点消费者。

我可以在 Kafka Tool 中看到节点消费者组。它没有出现在 Kafa-consumer-groups CLI 输出中

kafka-consumer-groups --bootstrap-server localhost:9092 --list kafka-consumer-groups --bootstrap-server localhost:9092 --group node-kafka-consumer --describe

kafka-consumer-groups CLI 应该显示所有消费者 - 控制台和编程（在我的例子中是 node-kafka 消费者）

请在下面找到操作系统详细信息

我参考了以下参考并尝试了，但它不起作用

1. https://github.com/Blizzard/node-rdkafka/issues/263
2. https://github.com/ibm-messaging/event-streams-samples/blob/master/kafka-nodejs-console-sample/Dockerfile
3. https://github.com/Blizzard/node-rdkafka/blob/master/examples/docker-alpine.md

我跑了下面的命令

yum install gcc-c++ python-devel.x86_64 cyrus-sasl-devel.x86_64 ca-certificates

npm install node-rdkafka-- 在输出下方查找

找到我的测试代码

找到以下错误，我得到

注意： 相同的代码适用于 Ubuntu 16。我们遵循以下步骤

我创建了一个 HighLevelProducer 来将消息发布到主题流，该主题流将由 ConsumerGroupStream 使用 kafka-node 使用。当我从同一个 ConsumerGroup 创建多个消费者以从同一个主题消费时，只创建一个分区并且只有一个消费者正在消费。我还尝试定义该主题的分区数，尽管我不确定在创建主题时是否需要定义它，如果需要，我需要提前多少个分区。此外，是否可以将对象推送到 Transform 流而不是字符串（我目前使用 JSON.stringify 因为否则我在消费者中得到 [Object object]。

消费者：

我正在尝试为具有 2 个分区的单个主题生成一些消息。所有消息都将仅发送到 2 号分区。我希望生产者流将消息分发到所有分区。

消费者组：