Kafkajs: How should one perform orderly shutdown with committing offsets?

[manjabes]

I'm experimenting with kafkajs and was wondering if there is a "built-in" way of shutting down the consumer and committing offsets.

So, suppose I'm doing something like this:

await consumer.subscribe({topic: 'mytopic', fromBeginning:true});

await consumer.run({
  autoCommitInterval : 120000, // to have the library autocommit every now and then
  autoCommit : false, //to disable committing offsets after each message
  eachMessage: async ({ topic, partition, message }) => {
            ...
  });

and then want to add a signal handler for orderly shutdown

process.on('SIGINT', ()=> {
  consumer.disconnect();
  process.exit(1);
});

there the disconnect() will stop the consumer, leave the consumer group and disconnect from the broker. But not commit the offsets.

Theoretically I could track the offsets myself in the eachMessage callback and in the SIGINT handler commit them manually, but:

• that is kind of excessive considering that the library keeps track of the offsets anyway
• for me to be able to commit those offsets, the consumer should be running, otherwise I die with UnhandledPromiseRejectionWarning: KafkaJSNonRetriableError: Consumer group was not initialized, consumer#run must be called first but if the consumer is running then it keeps on consuming and processing messages further.

Is there another way that this can or should be done?

[tulios]

Hi @manjabes , you don't have to do anything manually. The library will constantly commit the offsets and disconnect will wait for your consumers to finishing processing and then it will commit the offsets, leave the group and close the connection. You just have to await for disconnect.

[JaapRood]

@tulios is that behaviour dependent on autoCommit having been enabled?

[tulios]

Yes @JaapRood , only if autoCommit is enabled, but it is enabled by default.

[manjabes]

@tulios note that I've disabled autoCommit because normally I wouldn't want to commit after each message. In that case (only relying on autoCommitInterval) the offsets will not automatically be committed (I've verified this by consuming the __consumer_offsets topic).
The debug log of kafkajs from the shutting down process:

{"level":"DEBUG","timestamp":"2019-09-25T13:07:58.522Z","logger":"kafkajs","message":"[ConsumerGroup] Fetching from 1 partitions for 1 out of 1 topics","topics":["mytopic"],"activeTopicPartitions":[{"topic":"mytopic","partitions":[0]}],"pausedTopicPartitions":[]}
{"level":"DEBUG","timestamp":"2019-09-25T13:07:58.523Z","logger":"kafkajs","message":"[Connection] Request Fetch(key: 1, version: 7)","broker":"mybroker:9092","clientId":"myclient","correlationId":9,"expectResponse":true,"size":96}
going to disconnect! 2019-09-25T13:08:02.140Z
{"level":"DEBUG","timestamp":"2019-09-25T13:08:02.141Z","logger":"kafkajs","message":"[Runner] stop consumer group","groupId":"mygroup","memberId":"myclient-d0a5c5ac-35c4-41de-8ed1-f0350f65dd56"}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:02.141Z","logger":"kafkajs","message":"[Runner] waiting for consumer to finish...","groupId":"mygroup","memberId":"myclient-d0a5c5ac-35c4-41de-8ed1-f0350f65dd56"}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:03.141Z","logger":"kafkajs","message":"[Runner] waiting for consumer to finish...","groupId":"mygroup","memberId":"myclient-d0a5c5ac-35c4-41de-8ed1-f0350f65dd56"}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:03.525Z","logger":"kafkajs","message":"[Connection] Response Fetch(key: 1, version: 7)","broker":"mybroker:9092","clientId":"myclient","correlationId":9,"size":70,"data":"[filtered]"}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:03.531Z","logger":"kafkajs","message":"[Connection] Request Heartbeat(key: 12, version: 1)","broker":"mybroker:9092","clientId":"myclient","correlationId":10,"expectResponse":true,"size":100}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:03.532Z","logger":"kafkajs","message":"[Connection] Response Heartbeat(key: 12, version: 1)","broker":"mybroker:9092","clientId":"myclient","correlationId":10,"size":10,"data":{"throttleTime":0,"errorCode":0}}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:03.532Z","logger":"kafkajs","message":"[Runner] consumer not running, exiting","groupId":"mygroup","memberId":"myclient-d0a5c5ac-35c4-41de-8ed1-f0350f65dd56"}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:04.147Z","logger":"kafkajs","message":"[Connection] Request LeaveGroup(key: 13, version: 1)","broker":"mybroker:9092","clientId":"myclient","correlationId":11,"expectResponse":true,"size":96}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:04.149Z","logger":"kafkajs","message":"[Connection] Response LeaveGroup(key: 13, version: 1)","broker":"mybroker:9092","clientId":"myclient","correlationId":11,"size":10,"data":{"throttleTime":0,"errorCode":0}}
{"level":"INFO","timestamp":"2019-09-25T13:08:04.149Z","logger":"kafkajs","message":"[Consumer] Stopped","groupId":"mygroup"}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:04.149Z","logger":"kafkajs","message":"[Consumer] consumer has stopped, disconnecting","groupId":"mygroup"}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:04.149Z","logger":"kafkajs","message":"[Connection] disconnecting...","broker":"mybroker:9092","clientId":"myclient"}
{"level":"DEBUG","timestamp":"2019-09-25T13:08:04.149Z","logger":"kafkajs","message":"[Connection] disconnected","broker":"mybroker:9092","clientId":"myclient"}
disconnected 2019-09-25T13:08:04.149Z

the code for which is:

    console.log('going to disconnect!', new Date().toJSON())
    await consumer.disconnect();
    console.log('disconnected', new Date().toJSON());

As can be seen, no offset committing happens.

[tulios]

@manjabes the client doesn't commit after each message, if you don't have a clear use case for disabling the auto commit I would recommend you to keep it enabled. This process is already optimized from the client perspective and it give you control over the frequency. If auto commit is on, when disconnecting, the consumer will commit even if the interval doesn't match your configuration.

If you want to control this process you can use stop instead of disconnect and call commitOffsets without any argument, then you call disconnect.

*In you example you probably haven't received any messages from fetch

[tulios]

Actually, I'm wrong. You can't call consumer.commitOffsets without arguments, you will need to either keep track of the offsets yourself or use eachBatch and keep a reference of commitOffsetsIfNecessary, which you can call without arguments to commit already resolved offsets.

[manjabes]

How should the automatic commits work when using the eachMessagecallback? Because in my experiments with a low-volume topic, it certainly seemed to me (based on consuming the __consumer_offsets in another client) that when autoCommitwas set to true(or not configured explicitly at all) then the library would commit offsets after each message (irrespective of the autoCommitIntervalsetting). Are You saying that with a higher-volume topic, the results would be different?

[manjabes]

So, I experimented with a higher-volume topic, autoCommitundefined (which should effectively be ==true), autoCommitInterval: 30000 and when the consumer started from the beginning and was chewing through the backlog, it indeed did not commit after every message ( https://pastebin.com/3jKUfmq2 ), but when the consumer group lag was in single digits, it still hammered the __consumer_offsets topic almost every record ( https://pastebin.com/eXcxWpWZ ).
When stopping the consumer, the offsets were committed, yes, but I'd like to avoid hammering the offset topic like that. Is there anything I'm missing?

[manjabes]

So, the "commit after each message" behaviour that I observed is because of this

https://github.com/tulios/kafkajs/blob/master/src/consumer/runner.js#L316

which causes the commits to be done after each fetch, contrary to what the API docs state:

The eachMessage handler provides a convenient and easy to use API, feeding your function one message at a time. It is implemented on top of eachBatch, and it will automatically commit your offsets and heartbeat at the configured interval for you.

(emphasis mine).

The configured intervals have no effect unless autoCommit is explicitly disabled.
If this behaviour cannot be changed, is it at least possible to provide a method such as this

  commitUncommitted() {
      this.commitOffsets(this.consumerGroup.uncommittedOffsets());
  }

in runner.js so that I could pause consumption, call this method to commit the uncommitted offsets and then disconnect from the broker?