Kafkajs: When using Promise.all to start connections, Kafka Broker times out

[ThisIsMissEm]

I had some code that essentially started two Kafka consumers (each for different topics), when doing:

await Promise.all([
  consumer1.start(),
  consumer2.start()
])

I'd receive errors (via logs) about the Kafka broker being unable to be connected to, but only for one of the consumers.

Finally, I changed it to:

await consumer1.start()
await consumer2.start()

And everything worked as expected.

The consumer start() code is essentially just:

  async start() {
    await this.consumer.connect();
    await this.consumer.subscribe({
      topic: this.topic,
      fromBeginning: true
    });

    await this.consumer.run({
       // ....
    })
  }

My gut feeling on this is that there's some sort of race condition triggered in the Broker connection code when starting multiple consumers (or potentially multiple consumers and producers) in parallel.

[ThisIsMissEm]

Perhaps further on this, what's the difference between await consumer.run({ and consumer.run({ — the docs use both, but aren't clear as to why you wouldn't want to await the promise from consumer.run()

[JaapRood]

My gut feeling on this is that there's some sort of race condition triggered in the Broker connection code when starting multiple consumers (or potentially multiple consumers and producers) in parallel.

Sounds like it, right? The first thing I wondered after this suggestion is whether the consumers are identical (subscribe to the same topics, etc.). That might be some useful info for whomever is going to dig into this.

[ThisIsMissEm]

Nope, totally different consumers (different topics & group-id's). I could probably try digging into it further, initially I thought it was an issue with my local kafka only having one broker (docker) so I added another, still didn't fix the issue: one consumer still failed repeatedly.

I also thought maybe it was due to a topic not existing, so I added admin code to ensure the topics existed, still no change.

I can probably add a breakpoint in and try to debug, but I'm not sure whether I'll get far.

On 26. Aug 2019, at 09:38, Jaap van Hardeveld [e-mail] wrote:

My gut feeling on this is that there's some sort of race condition triggered in the Broker connection code when starting multiple consumers (or potentially multiple consumers and producers) in parallel.

Sounds like it, right? The first thing I wondered after this suggestion is whether the consumers are identical (subscribe to the same topics, etc.). That might be some useful info for whomever is going to dig into this.

—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub, or mute the thread.

[tulios]

@ThisIsMissEm we have a similar code running, and so far we haven't seen any issues. KafkaJS establishes a connection with each broker it has to interact, and due to the way the authentication mechanisms work it has to syncronize the connection.

So following your code:
1) connect will lock the connection to the specific broker, establish the TCP connection, fetch the available versions and execute the flow of the selected authentication mechanism.
2) subscribe will fetch metadata for the given topics
3) run will do a lot of things: join and syncronize the consumer group being the main tasks.

Can you post the error message you are receiving? I suspect that all consumers are trying to connect to the same broker, this will depend on the size of the cluster, and the connection is taking longer than the current timeout. You can try to increase the authenticationTimeout, e.g:

new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  authenticationTimeout: 5000,
})

what's the difference between await consumer.run({ and consumer.run({

Sorry about that, we should document this better. consumer.run never ends, it chains one promise after the other. Awaiting the first promise blocks your code until the consumer has joined the group and the group is in sync. We await sometimes to make the tests more stable, to only execute the test bits after the group is stable but in reality, you don't have to await for consumer.run.

[ThisIsMissEm]

Here's what the full log looks like (I've just changed a few names/verbs to redact internal information, the log is still consistent with what I see)

mock-api $ yarn start
yarn run v1.17.3
$ nodemon
[nodemon] 1.19.1
[nodemon] to restart at any time, enter `rs`
[nodemon] watching: /REDACTED/src/**/* .env
[nodemon] starting `ts-node -r tsconfig-paths/register src/index.ts | pino-pretty`
[1566941229752] INFO  (mock-api/68093 on QL000000870): Before Start
[1566941229753] INFO  (mock-api/68093 on QL000000870): Start
[1566941229753] INFO  (mock-api/68093 on QL000000870): starting consumer: products
    module: "ProductsConsumer"
[1566941229755] INFO  (mock-api/68093 on QL000000870): starting consumer: stores
    module: "StoresConsumer"
[1566941229795] INFO  (mock-api/68093 on QL000000870): kafkajs: Starting
    module: "KafkaConnector"
    namespace: "Consumer"
    groupId: "products-11c947b08e778a786280cf16c51ceb28f0a5095a0a79860724202d98977df3d2"
[1566941230762] INFO  (mock-api/68093 on QL000000870): kafkajs: Connection timeout
    module: "KafkaConnector"
    namespace: "Connection"
    broker: "localhost:9092"
    clientId: "store-service"
[1566941230764] INFO  (mock-api/68093 on QL000000870): kafkajs: Failed to connect to seed broker, trying another broker from the list: Connection timeout
    module: "KafkaConnector"
    namespace: "BrokerPool"
    retryCount: 0
    retryTime: 346
[1566941232119] INFO  (mock-api/68093 on QL000000870): kafkajs: Connection timeout
    module: "KafkaConnector"
    namespace: "Connection"
    broker: "localhost:9092"
    clientId: "store-service"
[1566941232119] INFO  (mock-api/68093 on QL000000870): kafkajs: Failed to connect to seed broker, trying another broker from the list: Connection timeout
    module: "KafkaConnector"
    namespace: "BrokerPool"
    retryCount: 1
    retryTime: 566
[1566941232830] INFO  (mock-api/68093 on QL000000870): kafkajs: Consumer has joined the group
    module: "KafkaConnector"
    namespace: "Runner"
    groupId: "products-11c947b08e778a786280cf16c51ceb28f0a5095a0a79860724202d98977df3d2"
    memberId: "store-service-dc827a15-a317-4d5d-a8fa-f3124e9cf6bc"
    leaderId: "store-service-dc827a15-a317-4d5d-a8fa-f3124e9cf6bc"
    isLeader: true
    memberAssignment: {
      "fake-qwell.products": [
        0
      ]
    }
    groupProtocol: "RoundRobinAssigner"
    duration: 3034
[1566941232866] INFO  (mock-api/68093 on QL000000870): Processing message at offset 0
    module: "ProductsConsumer"
    offset: "0"
    key: "27694a02-6410-43dd-9fd4-2377fb6017fa"
[1566941232866] INFO  (mock-api/68093 on QL000000870): Upserting product: 27694a02-6410-43dd-9fd4-2377fb6017fa
    module: "Repository"
[1566941233690] INFO  (mock-api/68093 on QL000000870): kafkajs: Connection timeout
    module: "KafkaConnector"
    namespace: "Connection"
    broker: "localhost:9092"
    clientId: "store-service"
[1566941233691] INFO  (mock-api/68093 on QL000000870): kafkajs: Failed to connect to seed broker, trying another broker from the list: Connection timeout
    module: "KafkaConnector"
    namespace: "BrokerPool"
    retryCount: 2
    retryTime: 1206
[1566941235903] INFO  (mock-api/68093 on QL000000870): kafkajs: Connection timeout
    module: "KafkaConnector"
    namespace: "Connection"
    broker: "localhost:9092"
    clientId: "store-service"
[1566941235903] INFO  (mock-api/68093 on QL000000870): kafkajs: Failed to connect to seed broker, trying another broker from the list: Connection timeout
    module: "KafkaConnector"
    namespace: "BrokerPool"
    retryCount: 3
    retryTime: 2384
[1566941239295] INFO  (mock-api/68093 on QL000000870): kafkajs: Connection timeout
    module: "KafkaConnector"
    namespace: "Connection"
    broker: "localhost:9092"
    clientId: "store-service"
[1566941239296] INFO  (mock-api/68093 on QL000000870): kafkajs: Failed to connect to seed broker, trying another broker from the list: Connection timeout
    module: "KafkaConnector"
    namespace: "BrokerPool"
    retryCount: 4
    retryTime: 3926
[1566941244230] INFO  (mock-api/68093 on QL000000870): kafkajs: Connection timeout
    module: "KafkaConnector"
    namespace: "Connection"
    broker: "localhost:9092"
    clientId: "store-service"
[1566941244230] INFO  (mock-api/68093 on QL000000870): kafkajs: Failed to connect to seed broker, trying another broker from the list: Connection timeout
    module: "KafkaConnector"
    namespace: "BrokerPool"
    retryCount: 5
    retryTime: 7920
[1566941244237] ERROR (KafkaJSNumberOfRetriesExceeded/68093 on QL000000870): Connection timeout
    KafkaJSNumberOfRetriesExceeded
      Caused by: KafkaJSConnectionError: Connection timeout
        at Timeout.onTimeout [as _onTimeout] (/REDACTED/node_modules/kafkajs/src/network/connection.js:149:23)
        at ontimeout (timers.js:436:11)
        at tryOnTimeout (timers.js:300:5)
        at listOnTimeout (timers.js:263:5)
        at Timer.processTimers (timers.js:223:10)

[ThisIsMissEm]

That was with the await on consumer.run removed. Even increasing the authenticationTimeout to 10,000 has no effect. We're usually operating against a single broker (development + staging environments, in production, we do have more brokers, but we still would like to see things reliable in our non-production environments.

[tulios]

I can see that you are not using authentication in your test environment, so the authenticationTimeout won't be used. It's common for a single broker to delay things a bit, try to increase the connectionTimeout, the default value is 1s.

It should be reliable on your non-prod environments; I think the connection timeout will do the trick. It's quite stable for us at the moment, but let's see if it solves your problem.

[ThisIsMissEm]

Aha! Okay, yeah, increasing the connectionTimeout up to 10,000 helped, and the service started as expected. What’s a generally good amount of time to allow here? 1 second seems too low

On 28. Aug 2019, at 10:43, Túlio Ornelas [e-mail] wrote:

I can see that you are not using authentication in your test environment, so the authenticationTimeout won't be used. It's common for a single broker to delay things a bit, try to increase the connectionTimeout, the default value is 1s.

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub https://github.com/tulios/kafkajs/issues/480?email_source=notifications&email_token=AAAHQ242XKII625SERMOTJTQGY3BVA5CNFSM4IPGVLU2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD5KLLYA#issuecomment-525645280, or mute the thread https://github.com/notifications/unsubscribe-auth/AAAHQ2Y4JQW4R45C3KACMP3QGY3BVANCNFSM4IPGVLUQ.

[tulios]

I don't know how to answer that :D
We have our services colocated with the Kafka cluster on AWS, so they have minimal latency. Since KafkaJS have long-lived connections, I would say that more significant value is not the problem, we need a timeout to catch connectivity errors or machines going down.

I think 10s looks good, we keep the defaults small so you have to think about your setup, but maybe we should increase this value a bit.

[JaapRood]

Having worked with time-outs before in previous deployments with other kafka clients: it should be less than the session timeout to prevent big issues, but network latency is definitely something to be tolerated too. We actually found 10s to be a good trade-off as well!