How to know deterministically know when a consumer is ready?

[apeloquin-agilysys]

In KafkaJS we make use of the consumer.group_join event for our application to know a consumer is ready to receive messages.

consumer.on(consumer.events.GROUP_JOIN, (event: ConsumerGroupJoinEvent) => {
  // inform app this consumer is ready
});

Per #38, the on methods don't appear to be on the immediate road map.

About the on methods, we don't plan to implement them yet.

I'm wondering if you have an alternative approach we can use to ascertain consumer readiness?

Originally we were using this:

consumer.assignment().length > 0

Unfortunately, it turns out that when you have a topic with more consumers than partitions, the consumers that don't get a partition also never satisfy this condition.

I see we get log entries:

{
  "message": "Consumer connected",
  "extra": {
    "fac": "BINDING",
    "name": "undefined#consumer-1"
  }
}

...but even if we were to hijack these events with our own Logger, I see no way to tie the name back to the consumers created (in our case we may have dozens).

[apeloquin-agilysys]

Sorry, I just noticed this very much overlaps with #65.

[apeloquin-agilysys]

Reopening. While this does overlap with #65, my use case is focused on readiness at startup.

The following test appears to show that the Consumer connected event that is logged and associated with the consumer.connect() await response is not indicative of the consumer actually be available to receive all messages sent to the topic from that point forward.

import {KafkaJS as Confluent} from "@confluentinc/kafka-javascript";
import {fail} from "assert";

const topicCount = 6;

const groupId = (t: number) => `test-confluent-group-${t}`;
const topic = (t: number) => `test-confluent-topic-${t}`;

describe("Consumer ready", () => {
  let kafka: Confluent.Kafka;
  let admin: Confluent.Admin;
  let producer: Confluent.Producer;
  let consumers: Confluent.Consumer[] = [];
  let messagesReceived: string[] = [];

  beforeEach(async () => {
    consumers = [];
    messagesReceived = [];

    kafka = new Confluent.Kafka({kafkaJS: {brokers: ["localhost:9092"], logger: new ConfluentLogger()}});
    admin = kafka.admin();
    await admin.connect();

    // clear all groupIds to avoid contamination from previous runs
    const groupIds: string[] = [];
    for (let t = 0; t < topicCount; t++) {
      groupIds.push(groupId(t));
    }
    await admin.deleteGroups(groupIds);

    const topics: Confluent.ITopicConfig[] = [];
    for (let t = 0; t < topicCount; t++) {
      topics.push({topic: topic(t)});
    }
    await admin.createTopics({topics});

    producer = kafka.producer();
    await producer.connect();

    let promises: Promise<any>[] = [];
    for (let t = 0; t < topicCount; t++) {
      promises.push(setupConsumer(t));
    }
    await Promise.all(promises);
  });

  afterEach(async () => {
    const promises: Promise<void>[] = [];
    promises.push(admin.disconnect());
    promises.push(producer.disconnect());
    for (const consumer of consumers) {
      promises.push(consumer.disconnect());
    }
    await Promise.all(promises);
  });

  it("sends/receives immediately after consumer connected", async () => {
    await sendMessagesAndWait();
  });

  it("sends/receives after consumer assignment", async () => {
    // this approach is not available when there are more app instances than partitions
    for (const consumer of consumers) {
      await until(() => consumer.assignment().length > 0);
    }

    await sendMessagesAndWait();
  });

  async function setupConsumer(t: number) {
    const consumer = kafka.consumer({kafkaJS: {groupId: groupId(t)}});
    await consumer.connect();
    await consumer.subscribe({topic: topic(t)});
    await consumer.run({
      eachMessage: async ({message}) => {
        const value = message.value?.toString() ?? "";
        messagesReceived.push(value);
        console.log(`[Received] @ ${topic(t)} (#${messagesReceived.length} of ${topicCount}): ${value}`);
      }
    });
    consumers.push(consumer);
  }

  async function sendMessagesAndWait() {
    const promises: Promise<any>[] = [];
    for (let t = 0; t < topicCount; t++) {
      const value = `Message ${t + 1}`;
      console.log(`[Sending] @ ${topic(t)}: ${value}`);
      promises.push(producer.send({topic: topic(t), messages: [{value}]}));
    }
    await Promise.all(promises);

    await until(() => messagesReceived.length === topicCount);
  }

  async function until(condition: () => boolean) {
    const timeout = 10000;
    const finish = Date.now() + timeout;
    while (Date.now() <= finish) {
      const result = await condition();
      if (result) return;
      await new Promise(resolve => setTimeout(resolve, 50));
    }
    fail(`Failed within ${timeout!}ms`);
  }
});

class ConfluentLogger implements Confluent.Logger {
  private logLevel: Confluent.logLevel;

  constructor() {
    this.logLevel = Confluent.logLevel.INFO;
  }

  setLogLevel(logLevel: Confluent.logLevel) {
    this.logLevel = logLevel;
  }

  info = (message: string, extra?: object) => this.doLog(Confluent.logLevel.INFO, message, extra);
  error = (message: string, extra?: object) => this.doLog(Confluent.logLevel.ERROR, message, extra);
  warn = (message: string, extra?: object) => this.doLog(Confluent.logLevel.WARN, message, extra);
  debug = (message: string, extra?: object) => this.doLog(Confluent.logLevel.DEBUG, message, extra);

  namespace() {
    return this;
  }

  private doLog(level: Confluent.logLevel, message: string, extra?: object) {
    if (this.logLevel >= level) {
      console.log(`[ConfluentKafka] ${message}${extra ? ` ${JSON.stringify(extra)}` : ""}`);
    }
  }
}

The first test, which does nothing more than await on the consumer.connect() times out after only receiving a subset of the sent messages, and the messages are clearly send after the Consumer connected log events.

[ConfluentKafka] Producer connected {"name":"rdkafka#producer-2","fac":"BINDING","timestamp":1728934633476}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-4","fac":"BINDING","timestamp":1728934633484}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-6","fac":"BINDING","timestamp":1728934633486}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-5","fac":"BINDING","timestamp":1728934633487}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-3","fac":"BINDING","timestamp":1728934633488}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-7","fac":"BINDING","timestamp":1728934633495}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-8","fac":"BINDING","timestamp":1728934633497}
[Sending] @ test-confluent-topic-0: Message 1
[Sending] @ test-confluent-topic-1: Message 2
[Sending] @ test-confluent-topic-2: Message 3
[Sending] @ test-confluent-topic-3: Message 4
[Sending] @ test-confluent-topic-4: Message 5
[Sending] @ test-confluent-topic-5: Message 6
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-4","fac":"BINDING","timestamp":1728934633498}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-6","fac":"BINDING","timestamp":1728934633499}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-3","fac":"BINDING","timestamp":1728934633499}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-5","fac":"BINDING","timestamp":1728934633499}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-7","fac":"BINDING","timestamp":1728934634485}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-8","fac":"BINDING","timestamp":1728934634486}
[Received] @ test-confluent-topic-0 (#1 of 6): Message 1
[Received] @ test-confluent-topic-1 (#2 of 6): Message 2
[Received] @ test-confluent-topic-3 (#3 of 6): Message 4
[Received] @ test-confluent-topic-2 (#4 of 6): Message 3

...no more messages are received before the test times out

The second test, which waits for consumer.assignment().length > 0 on all consumers, is much more reliable and all expected messages are delivered. However, as noted, this is not a viable solution for a ready check in a runtime environment where its possible to have more consumers in the consumer group than there are partitions in the topic. Those consumers not assigned to a partition never meet the above condition and are never marked ready.. This is where the group_join event in KafkaJS proves invaluable.

[ConfluentKafka] Producer connected {"name":"rdkafka#producer-10","fac":"BINDING","timestamp":1728934645554}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-12","fac":"BINDING","timestamp":1728934645565}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-13","fac":"BINDING","timestamp":1728934645566}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-14","fac":"BINDING","timestamp":1728934645567}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-16","fac":"BINDING","timestamp":1728934645570}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-15","fac":"BINDING","timestamp":1728934645571}
[ConfluentKafka] Consumer connected {"name":"rdkafka#consumer-11","fac":"BINDING","timestamp":1728934645573}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-12","fac":"BINDING","timestamp":1728934645582}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-16","fac":"BINDING","timestamp":1728934645582}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-13","fac":"BINDING","timestamp":1728934645584}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-14","fac":"BINDING","timestamp":1728934645584}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-15","fac":"BINDING","timestamp":1728934646573}
[ConfluentKafka] Received rebalance event with message: 'Local: Assign partitions' and 1 partition(s), isLost: false {"name":"rdkafka#consumer-11","fac":"BINDING","timestamp":1728934646574}
[Sending] @ test-confluent-topic-0: Message 1
[Sending] @ test-confluent-topic-1: Message 2
[Sending] @ test-confluent-topic-2: Message 3
[Sending] @ test-confluent-topic-3: Message 4
[Sending] @ test-confluent-topic-4: Message 5
[Sending] @ test-confluent-topic-5: Message 6
[Received] @ test-confluent-topic-0 (#1 of 6): Message 1
[Received] @ test-confluent-topic-1 (#2 of 6): Message 2
[Received] @ test-confluent-topic-4 (#3 of 6): Message 5
[Received] @ test-confluent-topic-3 (#4 of 6): Message 4
[Received] @ test-confluent-topic-5 (#5 of 6): Message 6
[Received] @ test-confluent-topic-2 (#6 of 6): Message 3

[milindl]

Hello @apeloquin-agilysys , I think the right way here would be to look at the rebalance callback. It is triggered every time there's a rebalance, even if the said consumer has received no partitions as a result of the rebalance call. It could look something like this, you could wait on firstRebalanceComplete.

  const { ErrorCodes } = require('@confluentinc/kafka-javascript').KafkaJS;
  // ...
  let firstRebalanceComplete = false;
  consumer = kafka.consumer({
    kafkaJS: {
      clientId: cName,
      // Other config properties
    },
    'rebalance_cb': (err, assignment) => {
      switch (err.code) {
        case ErrorCodes.ERR__ASSIGN_PARTITIONS:
          console.log(`Assigned partitions ${JSON.stringify(assignment)}`); // This could be [] too
          if (!firstRebalanceComplete) {
            firstRebalanceComplete = true;
            console.log("Completed first rebalance for consumer " + cName);
          }
          break;
        case ErrorCodes.ERR__REVOKE_PARTITIONS:
          // You don't need to take any action for any of the other cases. Including them just for illustrative reasons
          break;
        default:
          console.error(err);
      }
    },
  });

[apeloquin-agilysys]

Thanks. I arrived at a similar (but less complete) conclusion based on the comment by justjake on #65.

There are a lot of callbacks available in GlobalConfig, ConsumerGlobalConfig, ProducerGlobalConfig, etc. However, the comments and typing config.d.ts don't provide much transparency in terms of usage for folks coming from a KafkaJS/Typescript transition (e.g. any and Function lead to internet searches hoping to find example code). Any plans to improve this?

    /**
     * Error callback (set with rd_kafka_conf_set_error_cb())
     */
    "error_cb"?: any;

    /**
     * Throttle callback (set with rd_kafka_conf_set_throttle_cb())
     */
    "throttle_cb"?: any;

    /**
     * Statistics callback (set with rd_kafka_conf_set_stats_cb())
     */
    "stats_cb"?: any;

    /**
     * Log callback (set with rd_kafka_conf_set_log_cb())
     */
    "log_cb"?: any;

    /**
     * Background queue event callback (set with rd_kafka_conf_set_background_event_cb())
     */
    "background_event_cb"?: any;

    /**
     * Socket creation callback to provide race-free CLOEXEC
     */
    "socket_cb"?: any;

    /**
     * Socket connect callback
     */
    "connect_cb"?: any;

    /**
     * Socket close callback
     */
    "closesocket_cb"?: any;

    /**
     * File open callback to provide race-free CLOEXEC
     */
    "open_cb"?: any;

    /**
     * Address resolution callback (set with rd_kafka_conf_set_resolve_cb()).
     */
    "resolve_cb"?: any;
    
    /**
     * Delivery report callback (set with rd_kafka_conf_set_dr_cb())
     */
    "dr_cb"?: boolean | Function;

    /**
     * Delivery report callback (set with rd_kafka_conf_set_dr_msg_cb())
     */
    "dr_msg_cb"?: boolean;

    /**
     * Callback to verify the broker certificate chain.
     */
    "ssl.certificate.verify_cb"?: any;
    
    /**
     * SASL/OAUTHBEARER token refresh callback (set with rd_kafka_conf_set_oauthbearer_token_refresh_cb(), triggered by rd_kafka_poll(), et.al. This callback will be triggered when it is time to refresh the client's OAUTHBEARER token. Also see `rd_kafka_conf_enable_sasl_queue()`.
     */
    "oauthbearer_token_refresh_cb"?: any;
    
    /**
     * Message consume callback (set with rd_kafka_conf_set_consume_cb())
     */
    "consume_cb"?: any;

    /**
     * Called after consumer group has been rebalanced (set with rd_kafka_conf_set_rebalance_cb())
     */
    "rebalance_cb"?: boolean | Function;

    /**
     * Offset commit result propagation callback. (set with rd_kafka_conf_set_offset_commit_cb())
     */
    "offset_commit_cb"?: boolean | Function;
    
    /**
     * Custom partitioner callback (set with rd_kafka_topic_conf_set_partitioner_cb())
     */
    "partitioner_cb"?: any;

[milindl]

I agree - the config.d.ts is auto-generated, and while the non-callback properties are commented pretty accurately, the callback properties are lacking. I'll look at improving it, and removing those callbacks from the typing which aren't supported.

[milindl]

Closing this out with the resolution, I'll address a bunch of typing and documentation issues together in #139 .

[justjake]

What I do these days is resolve a Promise in the first rebalance_cb call. It’s similar to @milindl’s example but I’m calling resolve() where the example does firstRebalanceComplete = true