问题标签 [kafkajs]

我正在观看 Robin Moffatt 的一个视频 ( https://rmoff.net/2020/06/17/loading-csv-data-into-kafka/ )，并相信 Apache Kafka 可能会帮助我自动化我拥有的工作流程。

我有一个要求，我需要从客户那里获取 CSV，将原始信息的子集以各种格式（文本或 csv）发送给 2 个供应商，从这些供应商接收数据，然后合并所有数据。

我有点不喜欢卡夫卡，但我想我会有一个如下的过程：

将客户的数据提取到 kafka 并保存到 SQL Server 或 Postgres 数据库。然后我将发布 2 个“我们有数据”流。每个流基本上都有一行，代表我们从客户那里收到的批次。这些作为主题的流将由 kafkaJS 消费者使用。使用消息中的信息，这些消费者基本上将根据该供应商所需的输出从数据库中选择数据。

在此过程中，我们期待 2 个响应。当每个响应进入 (SFTP) 时，我们会将响应文件（JSON 或 CSV）摄取到数据库中，就像我们对原始客户信息所做的那样。如果我们已收到所有数据，我们将发布另一条消息，该消息将由合并所有数据的消费者使用。

像罗宾这样的卡夫卡忍者有什么建议吗？非常感激。

广东

我@kafkajs/confluent-schema-registry 在打字稿节点应用程序中使用，试图下拉模式。我已经在 J​​ava 中使用类似的融合模式注册表 java 库 w\o 问题完成了此操作。

创建模式注册表客户端时，我看不到指定任何有关严格性的选项的方法。

我的 JSON 模式在块中有一个名为“version”的属性schema {}。我能够检索主题的最新架构 ID，但检索架构本身失败。似乎正在使用一些严格性设置，该设置不允许其不期望的模式中存在属性。从融合模式注册表 Java 库访问模式时，我没有看到这个问题。

有没有办法关闭这种严格模式？

我是 Kafka 的新手，我正在尝试设置一个 localhost 集群。

问题：完成设置后，我无法确认我的消息是否已生成或使用它们。

我所做的步骤：

第 1 步 - 动物园管理员

我确实zookeeper使用本地 telnet 连接和ruok命令（在默认端口 2181 上）安装并验证它已启动并运行。

第 2 步 - 配置

我有 3 个文件夹，每个文件夹都包含我从kafka 官方快速入门下载的 kafka 。在每个文件夹中，我对本地进行了config/server.properties如下修改：

我已经设置了broker.id=1，然后2和3其他人

我已经设置了listeners=PLAINTEXT://:9093，然后9094和 9095其他人

我设置logs.dir=了一个本地日志目录，其中每个代理都有一个子目录

第 3 步 - 启动

我已经使用bin/kafka-server-start.sh config/server.properties3 个 kafka 文件夹中的命令启动了所有 3 台服务器，每台服务器都使用它自己的本地配置

第 4 步 - 主题

我使用以下命令从三个 kafka 文件夹之一创建了一个共享主题：

bin/kafka-topics.sh --create --topic topic-1 --zookeeper localhost:2181 --partitions 3 --replication-factor 3

第 5 步 - 测试

好的，现在要测试一切是否正常运行，我尝试了两种方法第一种是使用 CLI 并手动添加消息，从任何 kafkabin文件夹中，我使用了这两个命令：

kafka-console-producer.sh --broker-list localhost:9093,localhost:9094,localhost:9095 –topic topic-1

kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic topic-1 --from-beginning

但是，当添加消息时，消费者端似乎什么都没有发生。关闭它时，我有Processed a total of 0 messages消息。

我还尝试使用该kafkajs库与消费者和生产者一起制作“hello world”类型脚本，但出现此错误：

{"level":"ERROR","timestamp":"2021-05-10T09:09:57.464Z","logger":"kafkajs","message":"[Connection] Response GroupCoordinator(key: 10, version: 2)","broker":"kevin-UX390UAK:9093","clientId":"my-app","error":"The group coordinator is not available","correlationId":0,"size":55}

任何想法我做错了什么？非常感谢 ！凯夫！

我正在尝试做这样的事情，但 testcafe 不会等到测试运行。一旦它点击 t.navigate(url)，异步 runtest() 函数就会返回 undefined，并且“eachMessage”代码继续以未定义的值。如何确保 runtest 在返回之前成功完成。显然 await 不起作用，我没有从 testcafe 或 nodejs 收到任何错误消息

先感谢您