Kafkajs: How to tweak the consumer fetch interval

[Sudheer312]

[tulios]

Hi @Sudheer312 , what is the issue?

[Sudheer312]

Hi @tulios , How can i control fetch interval as i cannot poll when ever i want like in java clients??

lets take an example. I want to poll new set of messages only after processing my current batch . so i want to stop the polling while i am processing my current batch.

if i use autoCommitInterval as 5000 ms will i get new set of messages only after 5 secs??

[tulios]

@Sudheer312 the API is different than the Java API. You can use eachBatch and then pause the consumer after you receive the batch. After processing the batch, you can call resume to unblock the consumer. KafkaJS doesn't have a poll or fetch API, although we are considering adding such a feature at some point.

[Sudheer312]

@Sudheer312 the API different than the Java API. You can use eachBatch and then pause the consumer after you receive the batch. After processing the batch, you can call resume to unblock the consumer. KafkaJS doesn't have a poll or fetch API, although we are considering adding such a feature at some point.

How to control the batch size? is there any parameter like no of messages per batch that i can recieve?

[tulios]

Take a look here: https://kafka.js.org/docs/consuming#a-name-options-a-options

You can configure maxBytes and maxBytesPerPartition, so if you know more or less the size of your messages, you can control the size of the batch. The Java client uses the same constraints to fetch data, but it discards the overflow and returns the number you requested.

If you care about the number of messages, you can implement the same logic. You can configure a reasonable amount on maxBytes and then you always process a fixed number of messages, only resolving the ones in your list.

[Sudheer312]

hi @tulios ,
tried the way you suggested . But it is running into infinitely consuming the same messages. this code is deployed in 2 server (2 consumers).

const options = {
groupId: config.KAFKA_RENDERER_CONSUMER_GROUP_ID,
maxBytes: 200,
maxBytesPerPartition: 200,
autoCommit: true,
encoding: 'utf8'
};

const consumingTopic = config.KAFKA_RENDERER_CONSUMER_TOPIC;
const kafkaConsumer = kafkaClient.consumer(options);

function sleep(time) {
return new Promise(resolve => setTimeout(resolve, time));
}

const run = async () => {
await kafkaConsumer.connect();
await kafkaConsumer.subscribe({ topic: consumingTopic });
await kafkaConsumer.run({
autoCommitInterval: 5000,
eachBatch: async ({ batch, commitOffsetsIfNecessary }) => {
const { topic } = batch;
await commitOffsetsIfNecessary();
kafkaConsumer.pause([{ topic }]);

  for (let message of batch.messages) {
    console.log(
      'message from consumer',
      topic,
      batch.partition,
      JSON.parse(message.value)
    );

    await sleep(300).then(() => {
          console.log('sleeping');
        }); 

  }
  kafkaConsumer.resume([{ topic }]);
}

});
};

[tulios]

Hi @Sudheer312 , take a look at the docs for eachBatch. There are a couple off issues with this code:

1) await commitOffsetsIfNecessary() you probably want to do this after you have processed the batch
2) When using eachBatch you have to resolve your offsets, otherwise, they don't advance and the client won't fetch the next batch

something like:

consumer.run({
  eachBatch: async ({ batch, resolveOffset, heartbeat }) => {
    consumer.pause([{ topic }])

    for (let message of batch.messages) {
      console.log(
        "message from consumer",
        topic,
        batch.partition,
        JSON.parse(message.value)
      )

      await sleep(300).then(() => {
        console.log("sleeping")
      })

      await resolveOffset(message.offset)
      await heartbeat()
    }

    consumer.resume([{ topic }])
  }
})

but given your example, I think eachMessage would fit you well. KafkaJS will only fetch a new batch of messages after you finish processing the current batch. It would look like this:

consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    console.log(
      'message from consumer',
      topic,
      partition,
      JSON.parse(message.value)
    )

    await sleep(300).then(() => {
      console.log('sleeping')
    }) 
  }
})

[Durairaj]

@Sudheer312 the API is different than the Java API. You can use eachBatch and then pause the consumer after you receive the batch. After processing the batch, you can call resume to unblock the consumer. KafkaJS doesn't have a poll or fetch API, although we are considering adding such a feature at some point.

@tulios Is someone working on the above feature? Let me know if u welcome any PR on the issue?

[Nevon]

No one is currently working on that, and it's not something that we've planned to work on in the near future.

Before receiving a PR, I would prefer that we discuss the use case and proposed API. We are not interested in just reproducing whatever the Java API does, unless there's a use-case that we currently don't solve for and their API solves the problem in a usable and idiomatic way. So far I don't see anything in this thread to indicate that this is something we need, but there may very well be use-cases I'm not aware of.

[Durairaj]

No one is currently working on that, and it's not something that we've planned to work on in the near future.

Before receiving a PR, I would prefer that we discuss the use case and proposed API. We are not interested in just reproducing whatever the Java API does, unless there's a use-case that we currently don't solve for and their API solves the problem in a usable and idiomatic way. So far I don't see anything in this thread to indicate that this is something we need, but there may very well be use-cases I'm not aware of.

Right, makes sense. Is there any way how to consume next messages in offset? I think java has that API right? I want to replicate the same behaviour as in Java client to port something to Nodejs. what u think?

[mschuber123]

I am playing with eaxmple https://github.com/mapr-demos/kafka-sample-programs.git. I added an kafkajs-Client consuming on topic "fast-messages". The Java-Client consuming with "consumer.poll(200)" receives messages minimum 3 times faster than kafkajs-Client. For my use-case "replay messages from a realtime system" consuming-speed is important. Is there any way to speed up kafkajs-Client ?