Есть ли способ получить сообщения из буфера очереди производителя в node-rdkafka?

Question

Представьте, что соединение с кластером Kafka закрыто (например, отключен кабель), и рано или поздно очередь буфера заполнится.Как получить эти сообщения из буфера и, скажем, отправить их в локальное хранилище?Спасибо!

Answer 1

Если node-rdkafka не может отправлять сообщения, находящиеся в буфере, срок их действия истечет после message.timeout.ms, что по умолчанию составляет 300000 (5 минут).По истечении каждого сообщения оно отправляет событие delivery-report и передает сообщение об ошибке и сообщение.

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

// Register delivery report listener
producer.on('delivery-report', function(err, report) {
    if (err) {
        console.error('Delivery report: Failed sending message ' + JSON.stringify(report));
        console.error(err);
        // We could retry sending the message or store it locally
    } else {
        console.log('Message produced, offset: ' + report.offset);
    }
});

Объект report содержит всю информацию (тело, ключ, раздел), необходимую для повторной отправки, если хотите.