kafkajs -断开连接后优雅地停止kafkajs实例

Question

我在生产测试和集成测试中都使用kafkajs。

在进行所有测试之前，我将使用生产者&消费者连接/订阅/运行(EachMeassage)创建一个kafkajs实例.经过所有的测试，我想优雅地停止所有节点进程，包括kafkajs组件。

我实际上是这么做的

export function stopHelper(): Promise<void> {
    return new Promise<void>((resolve, reject) => {
        if (kafkaHelperState === kafkaHelperStateStatus.running) {
            kafkaHelperState = kafkaHelperStateStatus.stopping
            log.debug("stopHelper", kafkaHelperState);
            Promise.all([producer.disconnect, consumer.disconnect])
                .then(() => {
                    kafkaHelperState = kafkaHelperStateStatus.stopped
                    log.info("stopHelper", kafkaHelperState);
                    resolve()
                })
                .catch(error => reject(error))
        } else {
            log.warn("stopHelper", "kafkaHelper is not " + kafkaHelperStateStatus.running)
        }
    })
}

承诺似乎起作用了。我可以看到，我的集成测试套件已经完成，生产者和消费者都断开了连接。但是我的节点进程仍然在运行，什么也不做。

在那之前我用的是卡夫卡节点。当我停止使用者时，我的节点进程就结束了，而不必指定任何process.exit(0)。

是否有一种方法可以优雅地销毁节点进程中的kafkajs实例？

Answer 1

Promise.all([producer.disconnect(), consumer.disconnect()])

而不是

Promise.all([producer.disconnect, consumer.disconnect])