auto.offset.reset to largest not working #607

jose9r · 2016-04-12T10:46:25Z

Hi, im using kafka with this library and php-rdkafka from https://github.com/arnaud-lb/php-rdkafka/ , and when i create a KafkaConsumer with a new group with setDefaultTopicConf including auto.offset.reset to largest, and i cant consume new messages, there is an error with commit i supose (auto.commit is enable). The debug output with topic,fetch options is here:

teamcity[testFinishedsetting offset INVALID for commit

edenhill · 2016-04-12T12:44:11Z

Can you paste your setup code? (config and creating the consumer)

jose9r · 2016-04-13T08:38:17Z

   $conf = new Conf();
    $conf->set("group.id", "groupTest");
    $conf->set("offset.store.method", "broker");
    $conf->set("metadata.broker.list", "10.0.0.203:9092");
    $conf->set("debug", "topic,fetch");
    $topicConf = new TopicConf();
    $topicConf->set("auto.offset.reset", "largest");
    $conf->setDefaultTopicConf($topicConf);
    $consumer = new KafkaConsumer($conf);
    $consumer->subscribe(["queueTest"]);
    $msg = $consumer->consume(3600e3);

edenhill · 2016-04-13T12:39:11Z

You have 4 messages in queueTest 0 and the consumer starts reading at the end (since you have auto.offset.reset=largest) so you will not see any messages.
You will need to produce new messages to the topic to see new messages.

jose9r · 2016-04-13T15:27:13Z

I produce new messages, but its the same problem because last commited offset for the first time is invalid and when i try to consume these messages and fetch for the offset is invalid, and finally set committted offset to -1001 and setting offset INVALID for commit, like the first time.

arnaud-lb · 2016-04-13T16:59:20Z

@jose9r You may want to try to publish messages while the script is running. That way, you will receive the messages, and also save a valid offset. Alternatively, auto.offset.reset=smallest would allow to consume messages from the beginning when you haven't consumed any message from the topic, and to start from the saved offset the next time.

edenhill · 2016-04-15T06:49:55Z

@jose9r Any luck with this?

jose9r · 2016-04-15T07:19:45Z

@edenhill no, the problem with the offset continues

edenhill · 2016-04-15T07:29:44Z

So even though you are producing new messages for the given partitions while the consumer is running no new messages are showing up in the consumer?

jose9r · 2016-04-15T07:31:49Z

@edenhill exactly!

edenhill · 2016-04-15T07:32:40Z

Okay, can you run your program with debug=topic,fetch and provide the logs, commenting where in the logs you are starting to produce new messages?
Thanks

edenhill · 2016-04-15T07:46:05Z

Which librdkafka version are you on? You should be using latest master

jose9r · 2016-04-15T07:50:47Z

I have commit version e32127b (7 April). I'm going to test with latest master.

jose9r · 2016-04-15T08:28:13Z

And next execution:

edenhill · 2016-04-15T08:44:15Z

It looks like it doesnt commit your consumed offsets.
Can you show your complete source code?

jose9r · 2016-04-15T08:48:49Z

$conf = new Conf();
$conf->set("group.id", "group15");
$conf->set("offset.store.method", "broker");

    $conf->set("metadata.broker.list", "10.0.0.203:9092,10.0.0.208:9092,10.0.0.209:9092");
    $conf->set("debug", "topic,fetch");
    $topicConf = new TopicConf();
    $topicConf->set("auto.offset.reset", "largest");
    $conf->setDefaultTopicConf($topicConf);
    $consumer = new KafkaConsumer($conf);
    $consumer->subscribe(["queueTest"]);

    $read = false;
    $msg = "msg";
    while ($msg != null) {
        $msg = $consumer->consume(3600e3);
        //$msg2 = $consumer2->consume(3600e3);
        switch ($msg->err) {
            case RD_KAFKA_RESP_ERR_NO_ERROR:
                $read = true;
                echo "offset: " . $msg->offset, "\n";
                echo "message: ".$msg->payload, "\n";
                break;
            case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                $read = true;
                echo "key: " . $msg->key, "\n";
                echo "offset: " . $msg->offset, "\n";
                $msg = "";
                echo "No more messages; will wait for more\n";
                break;
            case RD_KAFKA_RESP_ERR__TIMED_OUT:
                $read = true;
                echo "Timedout\n";
                break;
            default:
                throw new \Exception($msg->errstr(), $msg->err);
                break;
        }
    }

edenhill · 2016-04-15T08:51:53Z

I'm not sure what exits that loop, but you will need to call $consumer->close() before terminating your program so that it can commit final offsets.

jose9r · 2016-04-15T09:00:24Z

$consumer->close() doesnt exist, i think the problem is related with php-rdkafka library, @arnaud-lb can you read this topic please? thank you!!

arnaud-lb · 2016-04-15T11:13:22Z

php-rdkafka calls close() when the KafkaConsumer object is destroyed (e.g. when the object is not referenced anymore). Note however that hitting CTRL-C or killing the program may not let PHP properly destroy objects.

I think that tuning auto.commit.interval.ms will also cause the offsets to be committed earlier.

jose9r · 2016-04-15T12:31:27Z

i change auto.commit.interval.ms to 1000, but the problem continues. Another question is how can i see consumer groups created, because with kafka-consumer-groups.sh script from kafka i only see groups created by console, whose offset is stored in zookeeper, no in kafka __consumer_offsets. thanks

edenhill · 2016-04-15T14:45:43Z

@jose9r Try setting auto.commit.interval.ms to 1000 and add a sleep(5) before exiting your program and see if that helps. If so it is a timing/destructor issue, otherwise there's something else wrong.
Please note @arnaud-lb 's comment on Ctrl-C too.

jose9r · 2016-04-18T10:21:00Z

I try with auto.commit.interval.ms and sleep and it doesnt work. Im using kafka-manager tool for monitoring, and when I run the program with auto.offset.reset to largest, the group doesn't appear . If I set auto.offset.reset to smallest, the group appears and I can see current offset and queue size.

edenhill · 2016-04-18T10:30:58Z

That is an indication that there are no new messages coming in on the topic, librdkafka will only commit offsets for received messages, so if you start a new group to read at the end of the partition ("latest") and no new messages are produced it will just idle at the end, without committing.

edenhill · 2016-04-21T23:41:43Z

Is this still an issue?

jose9r · 2016-04-22T10:58:46Z

Yes, the problem continues.

edenhill · 2016-04-22T16:44:42Z

Can you comment on my comment from 4 days ago about not producing new messages?

charlessoft · 2016-04-29T07:20:18Z

i have same program, i want to receive least message.
i use C++ code

RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
conf->set("bootstrap.servers", brokers, errstr);
conf->set("group.id", group1, errstr);
tconf->set("auto.offset.reset", "latest", errstr);
tconf->set("enable.auto.commit", "true", errstr);
tconf->set("auto.commit.interval.ms", "1000", errstr);
tconf->set("session.timeout.ms", "30000", errstr);

while (run) {
    RdKafka::Message *msg = consumer->consume(topic, partition, 1000);
    msg_consume(msg, NULL);
    delete msg;
    consumer->poll(0);
}

recive all the message 。

edenhill closed this as completed May 11, 2016

fmvin mentioned this issue May 30, 2018

auto.offset.reset=latest not working mfontanini/cppkafka#76

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

auto.offset.reset to largest not working #607

auto.offset.reset to largest not working #607

jose9r commented Apr 12, 2016

edenhill commented Apr 12, 2016

jose9r commented Apr 13, 2016

edenhill commented Apr 13, 2016

jose9r commented Apr 13, 2016

arnaud-lb commented Apr 13, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 15, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 15, 2016

edenhill commented Apr 15, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 15, 2016

jose9r commented Apr 15, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 15, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 15, 2016

arnaud-lb commented Apr 15, 2016

jose9r commented Apr 15, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 18, 2016

edenhill commented Apr 18, 2016

edenhill commented Apr 21, 2016

jose9r commented Apr 22, 2016

edenhill commented Apr 22, 2016

charlessoft commented Apr 29, 2016 •

edited

auto.offset.reset to largest not working #607

auto.offset.reset to largest not working #607

Comments

jose9r commented Apr 12, 2016

teamcity[testFinishedsetting offset INVALID for commit

edenhill commented Apr 12, 2016

jose9r commented Apr 13, 2016

edenhill commented Apr 13, 2016

jose9r commented Apr 13, 2016

arnaud-lb commented Apr 13, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 15, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 15, 2016

edenhill commented Apr 15, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 15, 2016

jose9r commented Apr 15, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 15, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 15, 2016

arnaud-lb commented Apr 15, 2016

jose9r commented Apr 15, 2016

edenhill commented Apr 15, 2016

jose9r commented Apr 18, 2016

edenhill commented Apr 18, 2016

edenhill commented Apr 21, 2016

jose9r commented Apr 22, 2016

edenhill commented Apr 22, 2016

charlessoft commented Apr 29, 2016 • edited

charlessoft commented Apr 29, 2016 •

edited