LibrdKafkaError：经纪人：随机运行约 2 小时后未知成员

Question

现在，我想实施node-rdkafka到我们的服务中，但我多次遇到此错误Broker: Unknown member。github 上的同一问题是https://github.com/confluenceinc/confluence-kafka-dotnet/issues/1464。他们说我们的消费者使用相同的组 ID 来重试或延迟。但我没有发现我的代码有任何重试和延迟。或https://github.com/confluenceinc/confluence-kafka-python/issues/1004，但我重新检查了所有消费者组 ID，它是唯一的。

生产者的配置node-rdkafka如下：

      this.producer = new Producer({
        "client.id": this.cliendID,
        "metadata.broker.list": this.brokerList,
        'compression.codec': "lz4",
        'retry.backoff.ms': 200,
        'socket.keepalive.enable': true,
        'queue.buffering.max.messages': 100000,
        'queue.buffering.max.ms': 1000,
        'batch.num.messages': 1000000,
        "transaction.timeout.ms": 2000,
        "enable.idempotence": false,
        "max.in.flight.requests.per.connection": 1,
        "debug": this.debug,
        'dr_cb': true,
        "retries": 0,
        "log_cb": (_: any) => console.log(`log_cb =>`, _),
        "sasl.username": this.saslUsername,
        "sasl.password": this.saslPassword,
        "sasl.mechanism": this.saslMechanism,
        "security.protocol": this.securityProtocol
      }, {
        "acks": -1
      })

Consumer的配置node-rdkafka如下：

this.consumer = new KafkaConsumer({
        'group.id': this.groupID,
        'metadata.broker.list': this.brokerList,
        "sasl.username": this.saslUsername,
        "sasl.password": this.saslPassword,
        "enable.auto.commit": false,
        "auto.commit.interval.ms": 2000,
        "session.timeout.ms": 45000,
        "max.poll.interval.ms": 300000,
        "heartbeat.interval.ms": 3000,
        "api.version.request.timeout.ms": 10000,
        "max.in.flight.requests.per.connection": 1,
        "debug": this.debug,
        "sasl.mechanism": this.saslMechanism,
        "log.connection.close": true,
        "log.queue": true,
        "log_level": 7,
        "log.thread.name": true,
        "isolation.level": "read_committed",
        "ssl.ca.location": "/etc/ssl/certs/",
        "log_cb": (_: any) => console.log(`log_cb =>`, _),
        "security.protocol": this.securityProtocol
      }, {})

      await new Promise(resolve => {

        this.consumer?.connect()
        this.consumer?.on('ready', () => {
          try {
            this.consumer?.subscribe(subscriptions)
            this.consumer?.consume()
            console.log('[SUCCESS] Subscribe Event => all event')
          } catch (err) {
            console.log('[FAILED] Subscribe => all event')
            console.log(err)
          }

          resolve(this.consumer)
        }).on('data', async (data) => {

          this.topicFunctionMap[data.topic]({
            partition: data.partition,
            topic: data.topic,
            message: {
              key: data.key,
              offset: data.offset.toString(),
              size: data.size,
              value: data.value,
              timestamp: data.timestamp?.toString()
            }
          } as ISubsCallbackParam)

          this.consumer?.commitSync({
            topic: data.topic,
            offset: data.offset,
            partition: data.partition
          })


        })
      })

使用这些配置，消费者能够接收事件，但它不会持续很长时间。大约 2 小时后，它会随机给出这些错误。我不确定是否是因为手动提交或我们的函数花了很长时间，因为我已经尝试了两者async并sync提交，因此 commitSync 它不依赖于我们的函数。

之所以这么说，是因为我们的功能花了很长时间，这使得我们的消费者被踢出群组。也许是我发现其他错误 Broker: Specified group Generation id is not valid 后的嫌疑人

来源：https ://github.com/confluenceinc/confluence-kafka-dotnet/issues/1155

它说我需要增加会话超时，然后我尝试将其增加到"session.timeout.ms": 3000005 分钟，而"heartbeat.interval.ms":3000我在 github 问题中发现的心跳应该小于=（超时/3）。所以我认为3秒就可以了。

使用"session.timeout.ms": 300000和"heartbeat.interval.ms":3000 消费者能够消费并持续很长时间，但问题是：

• 第一次使用这些配置，运行大约 0-2 秒即可接收
• 过了一会儿，就收到了，但是需要 1-10 秒才能收到消息

详细错误：

received event => onCustomerServiceRegister
[COMMIT_ERR]  LibrdKafkaError: Broker: Unknown member
    at Function.createLibrdkafkaError [as create] (/src/app/node_modules/node-rdkafka/lib/error.js:454:10)
    at KafkaConsumer.Client._errorWrap (/src/app/node_modules/node-rdkafka/lib/client.js:481:29)
    at KafkaConsumer.commitSync (/src/app/node_modules/node-rdkafka/lib/kafka-consumer.js:560:8)
    at KafkaRDConnect.<anonymous> (/src/app/dist/events/connectors/kafkaRD.js:240:110)
    at step (/src/app/dist/events/connectors/kafkaRD.js:53:23)
    at Object.next (/src/app/dist/events/connectors/kafkaRD.js:34:53)
    at /src/app/dist/events/connectors/kafkaRD.js:28:71
    at new Promise (<anonymous>)
    at __awaiter (/src/app/dist/events/connectors/kafkaRD.js:24:12)
    at KafkaConsumer.<anonymous> (/src/app/dist/events/connectors/kafkaRD.js:213:72)
    at KafkaConsumer.emit (node:events:376:20)
    at KafkaConsumer.EventEmitter.emit (node:domain:470:12)
    at /src/app/node_modules/node-rdkafka/lib/kafka-consumer.js:488:12 {