Kafka-node: Producer do not reconnect automatically

[fungiboletus]

Questions?

I'm using a very simple code but it looks like if I shutdown kafka and start it again, the client will never reconnect. Is that normal ?

Environment

• Node version: 10.14.0
• Kafka-node version: 4.1.0
• Kafka version: 2.2.0

For specific cases also provide

• Number of Brokers: 1
• Number partitions for topic: 1

Sample

const kafka = require('kafka-node');

const kafkaClient = new kafka.KafkaClient();
const kafkaProducer = new kafka.Producer(kafkaClient);
kafkaProducer.on('ready', () => {
  console.log('ready');
}).on('error', (err) => {
  console.error(err);
});

let i = 0;
setInterval(() => {
  i += 1;
  kafkaProducer.send([{
    topic: 'abcd',
    messages: [`abcd ${i}`],
  }], (error, data) => {
    console.log(error, data);
  });
}, 1000);

If you stop restart kafka while this script is running, it will always fail until you call kafkaClient.connect().

[hyperlink]

Your sends should fail while the broker is down but the client should retry a new socket connection every second until it's able to connect again.

[fungiboletus]

I think it does not try every second with my example :(

[hyperlink]

Your example would crash after the first error it receives since node considers errors emitted on an EventEmitter w/o a listener as an unhandled exception. If you added an error handler and turning on the debugging you can see it retries every second until you start the broker back up again.

[arypurnomoz]

I also got a similar problem, i kept getting "Broker not available (sendRequest)" Error

[hyperlink]

@arypurnomoz makes sense to get that error if you send a message while the broker is down. In that specific case, the connection is active check kicked in before sending a request. Another common one is Broker not available (socket closed) or request timeout if the socket hangs without being closed.

[fungiboletus]

@hyperlink Oh yes sorry I removed the error handler to simplify the example but it crashes immediately without it. I edited my example to put it back.

The broker doesn't seem to reconnect, my example prints the following error every second forever :

message: "Broker not available (sendRequest)"
stack: "BrokerNotAvailableError: Broker not available (sendRequest)
    at new BrokerNotAvailableError (project\node_modules\kafka-node\lib\errors\BrokerNotAvailableError.js:11:9)
    at async.ensureAsync (project\node_modules\kafka-node\lib\kafkaClient.js:1040:16)
    at project\node_modules\kafka-node\node_modules\async\dist\async.js:3269:12
    at project\node_modules\kafka-node\node_modules\async\dist\async.js:74:12
    at project\node_modules\kafka-node\lib\kafkaClient.js:1094:13
    at project\node_modules\kafka-node\node_modules\async\dist\async.js:3880:24
    at replenish (project\node_modules\kafka-node\node_modules\async\dist\async.js:1011:17)
    at iterateeCallback (project\node_modules\kafka-node\node_modules\async\dist\async.js:995:17)
    at project\node_modules\kafka-node\node_modules\async\dist\async.js:969:16
    at project\node_modules\kafka-node\node_modules\async\dist\async.js:3885:13
    at project\node_modules\kafka-node\node_modules\async\dist\async.js:3263:30
    at Immediate.<anonymous> (project\node_modules\kafka-node\node_modules\async\dist\async.js:119:16)
    at runCallback (timers.js:705:18)
    at tryOnImmediate (timers.js:676:5)
    at processImmediate (timers.js:658:5)
    at process.topLevelDomainCallback (domain.js:121:23)"
__proto__: Error {name: "BrokerNotAvailableError", constructor: }
null: undefined
null: NestedError {nested: Error: Unable to find available brokers to try
   …, message: "refreshBrokerMetadata failed", stack: "NestedError: refreshBrokerMetadata failed
    at a…"}

I turned on the debugging logs and I don't see the broker trying to connect back again.

[hyperlink]

unable to reproduce this using your code.

You should see messages like below showing the client attempting to reconnect.

  kafka-node:KafkaClient kafka-node-client reconnecting to 127.0.0.1:9092 +283ms
  kafka-node:KafkaClient kafka-node-client createBroker 127.0.0.1:9092 +0ms
  kafka-node:KafkaClient kafka-node-client sending versions request to 127.0.0.1:9092 +1ms
  kafka-node:KafkaClient kafka-node-client socket closed 127.0.0.1:9092 (hadError: false) +2ms

[fungiboletus]

Oh no. On my side I do not see any "socket closed".

The last debug output is this one :

kafka-node:KafkaClient broker is now ready
kafka-node:KafkaClient kafka-node-client updated internal metadata

As additional informations, I run kafka-node on Windows and my Kafka broker is in a GNU/Linux VM (it's not localhost).

[darky]

I have similar issue, producer not reconnected:
kafka-node-client is not reconnecting to localhost:9092 invalid broker

[darky]

4.0.3 version good for me
Regression since 4.0.4

[hyperlink]

@darky what's the best way to reproduce this?

[darky]

For me it reproduced with long running tests on Jest. Example:

describe('Tests', () => {
  let producer = new HighLevelProducer(new kafka.KafkaClient());

  it('Some long test', () => {
    // long stuff, more 10 min
  }, 20 * 60 * 1000);

  it('Some test with producer', () => {
    producer.send(/* payloads... */);
  });
});

[darky]

Also this error appears: Unable to find available brokers to try

[hyperlink]

@darky looks like the broker disconnected the client after being idle because your test runs longer than the brokers connections.max.idle.ms of 10 minutes.

The producer should still be usable since by default the socket will automatically reconnect after 1 second of being disconnected.

I will look into ways we can avoid emitting this error but for now I would just add an error handler to the producer so it doesn't fail the test with an unhandled exception but the producer.send should be ok.

Let me know if this is what you are seeing.

[darky]

@hyperlink

looks like the broker disconnected the client after being idle because your test runs longer than the brokers connections.max.idle.ms of 10 minutes.

Yes, it actual.

The producer should still be usable since by default the socket will automatically reconnect after 1 second of being disconnected.

No, it not working. This message occured: kafka-node-client is not reconnecting to localhost:9092 invalid broker. Seems it not valid broker anymore.

4.0.3 version works, because isValidBroker check since 4.0.4

[hyperlink]

So did the producer.send fail after seeing this message?

This message might be the discovery socket being disconnected for being idle. The isValidBroker method checks the socket is in the broker metadata before reconnecting it so if your broker reports the host as 127.0.0.1 but you kafkaHost was localhost you can see this message.

[darky]

So did the producer.send fail after seeing this message?

Yep

if your broker reports the host as 127.0.0.1 but you kafkaHost was localhost you can see this message

No, I write localhost for example. Real case with 192.168.246.102:9092

[hyperlink]

Can you output the contents of client.brokerMetadata after ready and after the error happens?

[darky]

After ready:

{ '1': { nodeId: 1, host: '192.168.246.102', port: 9092 } }

After error:

{ '1': { nodeId: 1, host: '192.168.246.102', port: 9092 } }

[ajwootto]

I may be running in to something similar.

I'm testing against a local single-broker Kafka set up with a serverside connection idle timeout of 30 seconds (for testing). Using 127.0.0.1 as the connection address (not localhost)

I set the client idleConnection timeout to 15 seconds (lower than server's timeout)

Then I create a HighLevelProducer and let it sit idle (never producing anything).

After the 15 second client timeout period, the connection closes and I see the following:

[2019-04-29T20:02:42.691Z][level:DEBUG] kafka-node-producer socket closed 127.0.0.1:9092 (hadError: false)
[2019-04-29T20:02:43.690Z][level:DEBUG] kafka-node-producer is not reconnecting to 127.0.0.1:9092 invalid broker

All subsequent produce requests error out and the connection is never reestablished.

I believe the problem is that isValidBroker is checking for a host and port combination to exist in the brokerMetadata, but the port in the metaData is a number while the port that is passed in in this case is a string. I added some logging to the function:

KafkaClient.prototype.isValidBroker = function ({ host, port }) {
  console.log({host, port})
  console.log(_(this.brokerMetadata)
    .values().value())

  return (
    this.connecting ||
    _(this.brokerMetadata)
      .values()
      .some({ host, port })
  );
};

And right before the problem happens, this is the log output:

{ host: '127.0.0.1', port: '9092' }
[ { nodeId: 0, host: '127.0.0.1', port: 9092 } ]

The following modification allows the reconnection to happen succesfully:

KafkaClient.prototype.isValidBroker = function ({ host, port }) {
  return (
    this.connecting ||
    _(this.brokerMetadata)
      .values()
      .some(({host: mHost, port: mPort}) => `${host}:${port}` === `${mHost}:${mPort}`)
  );
};

However, it would probably be better to make sure the port has the same type upstream, but I don't know enough about how this all works to figure out how to ensure that.

[galr52]

I may be running in to something similar.

I'm testing against a local single-broker Kafka set up with a serverside connection idle timeout of 30 seconds (for testing). Using 127.0.0.1 as the connection address (not localhost)

I set the client idleConnection timeout to 15 seconds (lower than server's timeout)

Then I create a HighLevelProducer and let it sit idle (never producing anything).

After the 15 second client timeout period, the connection closes and I see the following:

[2019-04-29T20:02:42.691Z][level:DEBUG] kafka-node-producer socket closed 127.0.0.1:9092 (hadError: false)
[2019-04-29T20:02:43.690Z][level:DEBUG] kafka-node-producer is not reconnecting to 127.0.0.1:9092 invalid broker

All subsequent produce requests error out and the connection is never reestablished.

I believe the problem is that isValidBroker is checking for a host and port combination to exist in the brokerMetadata, but the port in the metaData is a number while the port that is passed in in this case is a string. I added some logging to the function:

KafkaClient.prototype.isValidBroker = function ({ host, port }) {
  console.log({host, port})
  console.log(_(this.brokerMetadata)
    .values().value())

  return (
    this.connecting ||
    _(this.brokerMetadata)
      .values()
      .some({ host, port })
  );
};

And right before the problem happens, this is the log output:

{ host: '127.0.0.1', port: '9092' }
[ { nodeId: 0, host: '127.0.0.1', port: 9092 } ]

The following modification allows the reconnection to happen succesfully:

KafkaClient.prototype.isValidBroker = function ({ host, port }) {
  return (
    this.connecting ||
    _(this.brokerMetadata)
      .values()
      .some(({host: mHost, port: mPort}) => `${host}:${port}` === `${mHost}:${mPort}`)
  );
};

However, it would probably be better to make sure the port has the same type upstream, but I don't know enough about how this all works to figure out how to ensure that.

Well done!! we have encounter the same problem and indeed this solves the problem. Can you please open PR?

[ajwootto]

@hyperlink could we get a release with this fix in it? It's pretty important I think

[hyperlink]

Just published 4.1.3 which should have this.