Kafkajs: Is it possible check that Kafka client is successfully connected?

[AsyncLegs]

For health check status checking purposes for example?

[Nevon]

Just checking that the client is connected isn't really enough to say that it's healthy, but there are events that you can listen to in order to know the state of your consumer. https://kafka.js.org/docs/instrumentation-events#a-name-consumer-a-consumer

What we do for our applications is to check if the consumer has done a heartbeat within the configured sessionTimeout. However, this means that the consumer is considered "unhealthy" during rebalances, which happen during deploys, among other things. This could be considered healthy or unhealthy, depending on what the purpose of your healthcheck is, but you can get the state of the group if needed. Something like this:

const SESSION_TIMEOUT = 30000

const consumer = kafka.consumer({
  groupId,
  sessionTimeout: SESSION_TIMEOUT
})

const { HEARTBEAT } = consumer.events
let lastHeartbeat
consumer.on(HEARTBEAT, ({ timestamp }) => lastHeartbeat = timestamp)

const isHealthy = async () => {
  // Consumer has heartbeat within the session timeout,
  // so it is healthy
  if (Date.now() - lastHeartbeat < SESSION_TIMEOUT) {
    return true
  }

  // Consumer has not heartbeat, but maybe it's because the group is currently rebalancing
  try {
    const { state } = await consumer.describeGroup()

    return ['CompletingRebalance', 'PreparingRebalance'].includes(state)
  } catch () {
    return false
  }
}

Now, I'm not sure I would recommend relying on this kind of thing for alerting, but it can be good for an at-a-glance indication of health. One problem with this is that if your consumers are always rebalancing, because they're crashing and restarting over and over, this would still report healthy. But if we don't have that exception, it'll instead report unhealthy when you are deploying, which can be annoying if it triggers alerts.

My recommendation would be to base alerting primarily on metrics, like queue depth and such.

[AsyncLegs]

@Nevon Thanks for explanations! Sounds very useful.
This check just needed as very basic health check for an internal dashboard.
More complex health checks will be done with Kafka JMX exporter on the Kafka server side.

[WaleedAshraf]

@Nevon Above you said,

One problem with this is that if your consumers are always rebalancing, because they're crashing and restarting over and over, this would still report healthy.

How so? I don't think this will return healthy in such cases.

[Nevon]

How so? I don't think this will return healthy in such cases.

It will because the health check is returning true if the state is CompletingRebalance or PreparingRebalance, so even if it is rebalancing over and over, it will still be considered healthy.

[WaleedAshraf]

Got it.
That can be avoided by keeping track of firstHeartbeat = true/false and then doing something like,

return firstHeartbeat && ['CompletingRebalance', 'PreparingRebalance'].includes(state)

Right?

[Nevon]

I'm not sure what you mean by firstHeartbeat, but assuming that it is just a record of the consumer having heartbeat at some point, it doesn't solve the fundamental problem.

One workaround could be to keep track of how long the consumer has been rebalancing, and then tuning a limit for that to how long your deployments usually take, but again, that's just papering over the fundamental problem which is that this is not the right way to monitor the health of a Kafka consumer. Doing that will just mean that your dashboard is lying to you for X minutes until it finally gives up and admits that there is a problem.

The real solution is to monitor your consumer based on the work it is supposed to be doing. If it is not consuming messages fast enough from the queue, it is not healthy. If it is consuming at the rate you need it to, it is healthy. Simple as that.

[WaleedAshraf]

Got it. Thanks.

I was using firstHeartbeat as keeping track of when it becomes healthy for first time. But yes, it doesn't solve problem as it can go in crash loop (PreparingRebalance/CompletingRebalance) after becoming healthy.

[danielcohenlive]

It seems like by running kafka.consumer(..., you are connecting another consumer to the group. Is there a way to check without connecting an additional consumer?

[Nevon]

You don't need to create a new consumer instance in your healthcheck. Just pass in a reference to your existing consumer instance and use that.

[theliuk]

Just checking that the client is connected isn't really enough to say that it's healthy, but there are events that you can listen to in order to know the state of your consumer. https://kafka.js.org/docs/instrumentation-events#a-name-consumer-a-consumer

What we do for our applications is to check if the consumer has done a heartbeat within the configured sessionTimeout. However, this means that the consumer is considered "unhealthy" during rebalances, which happen during deploys, among other things. This could be considered healthy or unhealthy, depending on what the purpose of your healthcheck is, but you can get the state of the group if needed. Something like this:

const SESSION_TIMEOUT = 30000

const consumer = kafka.consumer({
  groupId,
  sessionTimeout: SESSION_TIMEOUT
})

const { HEARTBEAT } = consumer.events
let lastHeartbeat
consumer.on(HEARTBEAT, ({ timestamp }) => lastHeartbeat = timestamp)

const isHealthy = async () => {
  // Consumer has heartbeat within the session timeout,
  // so it is healthy
  if (Date.now() - lastHeartbeat < SESSION_TIMEOUT) {
    return true
  }

  // Consumer has not heartbeat, but maybe it's because the group is currently rebalancing
  try {
    const { state } = await consumer.describeGroup()

    return ['CompletingRebalance', 'PreparingRebalance'].includes(state)
  } catch () {
    return false
  }
}

Now, I'm not sure I would recommend relying on this kind of thing for alerting, but it can be good for an at-a-glance indication of health. One problem with this is that if your consumers are always rebalancing, because they're crashing and restarting over and over, this would still report healthy. But if we don't have that exception, it'll instead report unhealthy when you are deploying, which can be annoying if it triggers alerts.

My recommendation would be to base alerting primarily on metrics, like queue depth and such.

Would it be sound to use this concept of healthiness to the setup the corresponding healthiness probe for a consumer running on the top of kubernetes? (let's take into account the fact that, at deploy time, we can ask kubernetes to wait some time before checking healthiness probes, so that all the transient rebalancing events do not interfere with the healthiness of a consumer).

Other approaches that we have considered consist of:

1. Using a specific topic to check the consumer's ability to dequeue: part of the application, dedicated to the healthiness check, might enqueue a message and then try to dequeue it. If this end2end test at runtime breaks, the consumer is considered unhealthy.
2. Listening for STOP, DISCONNECT and CRASH (and maybe REQUEST_TIMEOUT) events to decide whether the consumer is healthy or not.