From f15ecdcc13458590a27f5b23423e5cf6871c5511 Mon Sep 17 00:00:00 2001 From: Alexandre Rieux Date: Tue, 3 Dec 2019 09:58:27 +0000 Subject: [PATCH 1/3] Fix first send missing metadata --- lib/baseProducer.js | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/lib/baseProducer.js b/lib/baseProducer.js index 12cb9f7d..1b587196 100644 --- a/lib/baseProducer.js +++ b/lib/baseProducer.js @@ -122,8 +122,19 @@ BaseProducer.prototype.send = function (payloads, cb) { var client = this.client; var requireAcks = this.requireAcks; var ackTimeoutMs = this.ackTimeoutMs; + const isMissingTopicMetadata = !Object.keys(client.topicMetadata).length - client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks, ackTimeoutMs, cb); + if (isMissingTopicMetadata) { + client.refreshMetadata(_.map(payloads, 'topic'), (err) => { + if (err) { + cb(err) + } + + client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks, ackTimeoutMs, cb); + }); + } else { + client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks, ackTimeoutMs, cb); + } }; BaseProducer.prototype.buildPayloads = function (payloads, topicMetadata) { From b11f82e630f484938d104caa52910b20505ab530 Mon Sep 17 00:00:00 2001 From: Alexandre Rieux Date: Tue, 3 Dec 2019 09:59:53 +0000 Subject: [PATCH 2/3] Remove info from doc --- README.md | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/README.md b/README.md index 38041d03..8b971b15 100644 --- a/README.md +++ b/README.md @@ -1092,18 +1092,6 @@ Result: # Troubleshooting / FAQ -## HighLevelProducer with KeyedPartitioner errors on first send - -Error: - -``` -BrokerNotAvailableError: Could not find the leader -``` - -Call `client.refreshMetadata()` before sending the first message. Reference issue [#354](https://github.com/SOHU-Co/kafka-node/issues/354) - - - ## How do I debug an issue? This module uses the [debug module](https://github.com/visionmedia/debug) so you can just run below before starting your app. From e29e1aa907e1251fb3f22a782610104ae2cee412 Mon Sep 17 00:00:00 2001 From: Alexandre Rieux Date: Tue, 3 Dec 2019 10:20:17 +0000 Subject: [PATCH 3/3] Add missing semi --- lib/baseProducer.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/baseProducer.js b/lib/baseProducer.js index 1b587196..476a011d 100644 --- a/lib/baseProducer.js +++ b/lib/baseProducer.js @@ -122,12 +122,12 @@ BaseProducer.prototype.send = function (payloads, cb) { var client = this.client; var requireAcks = this.requireAcks; var ackTimeoutMs = this.ackTimeoutMs; - const isMissingTopicMetadata = !Object.keys(client.topicMetadata).length + const isMissingTopicMetadata = !Object.keys(client.topicMetadata).length; if (isMissingTopicMetadata) { client.refreshMetadata(_.map(payloads, 'topic'), (err) => { if (err) { - cb(err) + cb(err); } client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks, ackTimeoutMs, cb);