NodeJS：KafkaJSProtocolError：组成员支持的协议与现有成员的协议不兼容

Question

我正在尝试使用 MongoDB debezium 连接器从 Kafka 捕获数据，但是当我尝试使用 KafkaJS 读取数据时出现错误：

KafkaJSProtocolError: The group member's supported protocols are incompatible with those of existing members

我正在使用 docker 图像来捕获数据。

以下是步骤，我正在关注：

1. 启动 Zookeeper

   docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:latest
   

2. 启动卡夫卡

   docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper debezium/kafka:latest
   

3. 我已经在复制模式下运行了 MongoDB

4. 启动 debezium Kafka 连接

   docker run -it --rm --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link zookeeper:zookeeper --link kafka:kafka  debezium/connect:latest
   

5. 然后发布 MongoDB 连接器配置

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "mongodb-connector", "config": { "connector.class": "io.debezium.connector.mongodb.MongoDbConnector", "mongodb.hosts": "rs0/abc.com:27017", "mongodb.name": "fullfillment", "collection.whitelist": "mongodev.test", "mongodb.user": "kafka", "mongodb.password": "kafka01" } }'
   

6. 如果我运行一个观察者 docker 容器，我可以在控制台中以 Json 格式数据

   docker run -it --name watchermongo --rm --link zookeeper:zookeeper --link kafka:kafka debezium/kafka:0.9 watch-topic -a -k fullfillment.mongodev.test
   

但我想在应用程序中捕获这些数据，以便我可以对其进行操作、处理并推送到 ElasticSearch。为此，我正在使用

https://github.com/tulios/kafkajs 

但是当我运行消费者代码时，我得到了错误。这是代码示例

//'use strict';






// clientId=connect-1, groupId=1

const { Kafka } = require('kafkajs')



const kafka = new Kafka({

  clientId: 'connect-1',

  brokers: ['localhost:9092', 'localhost:9093']

})


// Consuming

const consumer = kafka.consumer({ groupId: '1' })



var consumeMessage = async () => {



await consumer.connect()

await consumer.subscribe({ topic: 'fullfillment.mongodev.test' })



await consumer.run({

  eachMessage: async ({ topic, partition, message }) => {

    console.log({

      value: message.value.toString(),

    })

  },

})



}



consumeMessage();


KafkaJSProtocolError: The group member's supported protocols are incompatible with those of existing members

Answer 1

您不应该在 Connect 和您的 KafkaJS 使用者中使用相同的 groupId。如果这样做，它们将属于同一个消费者组，这意味着消息只会被其中一个或另一个消费，即使它甚至可以工作。

如果您将 KafkaJS 使用者的 groupId 更改为独特的，它应该可以工作。

请注意，默认情况下，新的 KafkaJS 消费者组将从最新的偏移量开始消费，因此它不会消费已经产生的消息。您可以在 consumer.subscribe 调用中使用 fromBeginning 标志覆盖此行为。见https://kafka.js.org/docs/consuming#from-beginning