Node-rdkafka: What's the properly way to get the last offset

[Rick83600]

My use case : I want to get an history of messages, for example get the N last logs in a particular topic.
So I need to know the last offset for use this.assign().

How can I do that cleanly ?

:)

[webmakersteve]

If you're just trying to get the last offset so you can use it with assign, why not just subscribe to the topic and have it read from the last offset?

Otherwise, you can get the last commited offset by running

consumer.commited(timeout, (err, topicPartitions) => {
   // topicPartitions is structured like so: { topic: 'topic', partition: 0, offset: 20 }
});

Otherwise you can configure the behavior in subscribe by configuring the auto.offset.reset and make it largest to mean the last offset.

[Rick83600]

Ok,

I just try to use committed()

const ts = 1000; consumer.committed(ts || 5000, (err, t) => { // return null [] });

Return an empty array, maybe this feature is in validation ? :)

I'm using v0.6.4

[edoardocomar]

@Rick83600 committed was buggy we've fixed it this week.
https://github.com/Blizzard/node-rdkafka/issues/65

[edited to say]
oops I saw the version bump - do you have some partitions assigned before invoking committed ?
and committed provides you with the list in the callback, not as synchronous return value

[Rick83600]

@edoardocomar My fault, I had assign after committed(), you're alright.

Thanks for your great work !

[karteek-gamooga]

@webmakersteve Does committed return offsets for all the partitions or only the partition for the particular consumer group. If it is the earlier is there any way to know what all partitions the current consumer group is consuming?

Thanks

[affair]

@webmakersteve
Is there a way to get current group offset without calling consume() ?
Currently rebalance_cb is the only place in code where I can get current offset and partitions

```
var consumer = new Kafka.KafkaConsumer({
'debug': 'all',
'group.id': 'consumer-test',
'metadata.broker.list': '192.168.2.60:19092',
'fetch.wait.max.ms': 100,
'fetch.message.max.bytes': 1024 * 1024,
'enable.auto.commit': false,
'rebalance_cb': function(err, assignment) {
console.log("+OK - rebalance cb");

if (err.code === Kafka.CODES.ERRORS.ERR__ASSIGN_PARTITIONS) {
  console.log("+OK - rebalance cb. assign: ", assignment);
  // Note: this can throw when you are disconnected. Take care and wrap it in
  // a try catch if that matters to you
  this.assign(assignment).committed(10000, (err, data) => {
    console.log("+OK - rebalance cb. committed data: ", data);


  });
} else if (err.code == Kafka.CODES.ERRORS.ERR__REVOKE_PARTITIONS){
  console.log("+OK - rebalance cb. unassign: ", assignment);
  // Same as above
  this.unassign().committed(10000, (err, data) => {
    console.log("+OK - rebalance cb. committed data: ", data);


  });
} else {
  // We had a real error
  console.error("+OK - rebalance cb. err: ", err);
}

},
'offset_commit_cb': function(err, topicPartitions) {

if (err) {
  console.error("+OK - offset commit cb. err: ", err);
  // There was an error committing
  console.error(err);
} else {
  console.error("+OK - offset commit cb. topicPartitions: ", topicPartitions);
  // Commit went through. Let's log the topic partitions
  // console.log(topicPartitions);
}

}
}, {});

consumer
.on('ready', arg => {
console.log("+OK - is ready ", JSON.stringify(arg, null, 4));

// Subscribe to the librdtesting-01 topic
// This makes subsequent consumes read from that topic.
consumer.subscribe(['librdtesting-01']);

consumer.consume();

})
.on('data', function(data) {
console.log('Message found! Contents below.');
console.log("data: ", data.value.toString(), " topic: ", data.topic, "offset: ", data.offset, " partition: ", data.partition, " timestamp: ", data.timestamp);
});

// Non-flowing mode
consumer.connect();
````

[webmakersteve]

@affair
rebalance_cb isn't the best place to get the "current committed offsets" because it is only giving you the assignment for your client. If there are multiple consumers in the group you will only get a fraction of the offsets.

committed will return the offsets for your group but you will need to specify a list of topic partitions for which you want committed offsets. You can get these from the broker metadata. That would be my suggestion.