uncommittedOffsets : Clarification and Alternatives

[vikas-goswami]

Hi Team,

We are migrating to Javascript client from KafkaJS and using APIs under: const { Kafka } = require("@confluentinc/kafka-javascript").KafkaJS;

We would like to get clarification on uncommittedOffsets support in the migration documentation present here: Within the eachBatch callback, use of uncommittedOffsets is unsupported Link to documentation

Is there any alternative provided to track and commit offsets with a similar fine-tuned control? I am attaching some example usages below.

Example Usage 1:

await resolveOffsets(batch.messages, resolveOffset)
await commitOffsetsIfNecessary(uncommittedOffsets())

Usage 2: Where we track and update uncommittedOffsets

const uncommittedOffsets = []
....
consumer.uncommittedOffsets = consumer.uncommittedOffsets.concat(uncommittedOffsets)
....
  if (consumer.uncommittedOffsets.length === 0) {
    return done(null, false)
  }
  consumer.commitOffsetsIfNecessary(consumer.uncommittedOffsets)
    .then(() => {
      consumer.uncommittedOffsets = []
      return done(null, true)
    })
    .catch(err => {
      return done(err, false)
    })
}

We do not want to use autocommit :true since, it does not provide the finer-grained control over when and how offsets are committed. So, please let us know if there's any alternative.

Thanks in advance.

[vikas-goswami]

Hi @milindl We have raised a query. Please have a look. Thanks in advance.

[milindl]

Regarding usage 1:

I understand that uncommittedOffsets will return all the offsets across topic partitions that have been processed/manually marked as resolved if not using auto resolve, so you're committing the said offsets. In this case, you can use commitOffsetsIfNecessary() without any argument, and it has the same behaviour of committing every resolved offset (both manually resolved or auto resolved, as applicable). In fact, this is the only way to use commitOffsetsIfNecessary in this library - if you want to pick and choose which offsets to commit, commitOffsets needs to be invoked directly.

For usage 2, it's a bit more difficult. Maybe you can maintain the last 'seen' offset while processing messages, and use the committed method on the consumer to get committed topic partitions offsets for each owned topic partition, commit those offsets where the last seen offset is greater than the last committed one?