0

我正在尝试连接到 kafka 服务器。身份验证基于 GSSAPI。

/opt/app-root/src/server/node_modules/node-rdkafka/lib/error.js:411
  return new LibrdKafkaError(e);
         ^
Error: broker transport failure
    at Function.createLibrdkafkaError (/opt/app-root/src/server/node_modules/node-rdkafka/lib/error.js:411:10)
    at /opt/app-root/src/server/node_modules/node-rdkafka/lib/client.js:350:28

这是我的 test_kafka.js：

const Kafka = require('node-rdkafka');

const kafkaConf = {
  'group.id': 'espdev2',
  'enable.auto.commit': true,
  'metadata.broker.list': 'br01',
  'security.protocol': 'SASL_SSL',
  'sasl.kerberos.service.name': 'kafka',
  'sasl.kerberos.keytab': 'svc_esp_kafka_nonprod.keytab',
  'sasl.kerberos.principal': 'svc_esp_kafka_nonprod@INT.LOCAL',
  'debug': 'all',
  'enable.ssl.certificate.verification': true,
  //'ssl.certificate.location': 'some-root-ca.cer',
  'ssl.ca.location': 'some-root-ca.cer',
  //'ssl.key.location': 'svc_esp_kafka_nonprod.keytab',
};

const topics = 'hello1';

console.log(Kafka.features);

let readStream = new Kafka.KafkaConsumer.createReadStream(kafkaConf, { "auto.offset.reset": "earliest" }, { topics })
readStream.on('data', function (message) {
  const messageString = message.value.toString();
  console.log(`Consumed message on Stream: ${messageString}`);
});
4

1 回答 1

0

您可以查看此问题以了解此错误的解释： https ://github.com/edenhill/librdkafka/issues/1987

取自@edenhill：

作为基于 librdkafka 的客户端的一般规则：假设集群和客户端配置正确，所有错误都可以忽略，因为它们很可能是暂时的，并且 librdkafka 将尝试自动恢复。

在这种特定情况下；如果组协调器请求失败，它将在 500 毫秒内重试（使用状态为 Up 的任何代理）。如果在丢失的心跳超时成员资格（session.timeout.ms）之前找到了新的协调员，则当前分配和组成员资格不会受到影响。自动偏移提交将停止，直到找到新的协调器。

在未来的版本中，我们将扩展错误类型以包含严重性，从而允许应用程序愉快地忽略非终端错误。此时，应用程序应考虑所有错误信息，而不是终端错误。

于 2020-10-05T18:59:07.967 回答