0

我有一个 HTTP2 应用程序,它创建一个新的 Kafka 流生产者,并在接收来自传入流的消息时向其发送消息。

第一次连接并正常运行。在随后的流中,它会失败而不会引发任何类型的错误。

这就是我的做法:

"use strict";

const {KafkaStreams} = require('kafka-streams');
const {nativeConfig: config} = require("./kafka-config.js");

const kafkaStreams = new KafkaStreams(config);
const stream = kafkaStreams.getKStream(null);

...stream stuff...
stream.to("test-topic");

await stream.start()

await stream.writeToStream("my message");

await setTimeout(kafkaStreams.closeAll.bind(kafkaStreams), 100););

有没有人知道什么可能是错的?

4

0 回答 0