Блог

Я пытаюсь подключиться к серверу 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}`);
});
  

Вы можете посмотреть на эту проблему для объяснения этой ошибки: https://github.com/edenhill/librdkafka/issues/1987

Взято из @edenhill:

Как общее правило для клиентов на основе librdkafka: учитывая, что кластер и клиент правильно настроены, все ошибки можно игнорировать, поскольку они, скорее всего, временные, и librdkafka попытается восстановиться автоматически.

В этом конкретном случае; если запрос координатора группы завершается неудачно, он будет повторен (с использованием любого брокера в состоянии Up) в течение 500 мс. Текущее назначение и членство в группе не будут затронуты, если новый координатор будет найден до истечения времени ожидания пропущенных ударов сердца (session.timeout.ms ). Фиксация автоматического смещения будет остановлена до тех пор, пока не будет найден новый координатор.

В будущей версии мы расширим тип ошибки, включив серьезность, что позволит приложениям с радостью игнорировать нетерминальные ошибки. В это время приложение должно считать все ошибки информационными, а не терминальными.