Librdkafka: Question of RdKafka::Consumer use OFFSET_STORED

[coolulu]

hi,edenhill
my environment is kafka 0.9, I use RdKafka::Consumer only set OFFSET_STORED and 'group.id','metadata.broker.list', 'event_cb'

steps：
1.start my program 
2.send some message, for example "1","2","3"
3.start my program recv "1","2","3" right
4.kill my program 
5.send some message, for example "4","5","6"
6.restart my program recv message

In step 6 my program recv messge
sometimes stored is right number of messges -> "4","5","6"
sometimes stored is recv more messges -> "1","2","3","4","5","6" or "3","4","5","6" and so on
sometimes stored is recv empty -> Consume ERR__PARTITION_EOF : Broker: No more messages

Is it missed other configure in OFFSET_STORED ?

[edenhill]

With 0.9 you should be using the KafkaConsumer instead as it provides a more convenient interface to the broker based offset storage and consumer balancing (if you want that).

See examples/rdkafka_consumer_example.cpp for an example.

To use balanced consumer groups you use subscribe() to a set of topics, if you want to consume a fixed set of partitions you should use assign() instead.

[coolulu]

I test rdkafka_consumer_example.cpp.

rdkafka_consumer_example listen a new topic

First of all,send some messages to new topic
next, start rdkafka_consumer_example was loss old messages

Is it missed other configure??

[edenhill]

If there is no previous stored offset for a group it will start consuming from the end of the partition. To start from the beginning use -X topic.auto.offset.reset=smallest

[coolulu]

thanks

after I modify rdkafka_consumer_example.cpp.
on recv message after commit,
I use commitAsync() was core dumped,
and how to use commitAsync (Message *message), *message Need to be delete pointer

rdkafka_consumer_example.cpp 

/*
   * Consume messages
   */
  while (run) {
    if (use_ccb) {
      std::cerr << "Use callback: Not implemented" << std::endl;
      break;
    }

    RdKafka::Message *msg = consumer->consume(10);
    msg_consume(msg, NULL);
//  consumer->commitSync(msg);
    consumer->commitSync();        
//  consumer->commitAsync(msg);    //how to use
//  consumer->commitAsync();          //core dumped
    delete msg;
  }

[edenhill]

Do you have a gdb backtrace from the crash?
Which version are you using? Please use latest master

I'm not sure I understand your question on commitAsync(msg), what are you seeing?

[coolulu]

I was use old version.

kafkaConsumer Gruop can automatic manage stored offset.
such kafkaConsumer Gruop not need to commit offset ?

[edenhill]

Yes, KafkaConsumer performs automatic commits by default.