Simple Consumer.c - No effect of setting offset configuration (Same result in earliest and latest

[sandippatel-incedo]

Same result after setting 2 different configuration

1. Offset : latest - It should give latest message after consumer is up and running but giving data from where it left

    if (rd_kafka_conf_set(conf, "auto.offset.reset", "latest",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
            fprintf(stderr, "%s\n", errstr);
            rd_kafka_conf_destroy(conf);
            return 1;
    }
   

2. Offset : earliest - Its working fine as it should receive all message from where it left

    if (rd_kafka_conf_set(conf, "auto.offset.reset", "earliest",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
            fprintf(stderr, "%s\n", errstr);
            rd_kafka_conf_destroy(conf);
            return 1;
    }
   

Can you please guide what other setting do i need to do to get latest offset while starting consumer app.

[edenhill]

The auto.offset.reset policy is only used if there is no previously committed offset for the topic,partition,group combo.

[edenhill]

If you are not interested in the committed offsets you have two options:

1. Set enable.auto.commit=false (to not commit new offsets) and use a unique group.id (to avoid reading previously committed offsets)
2. Set enable.auto.commit=false (..) and set up a rebalance_cb, from the rebalance callback you manually call assign() with the start offset (the per-partition .offset field) set to RD_KAFKA_OFFSET_END.

[sandippatel-incedo]

@edenhill , Its working fine as per my requirements, Thanks a lot for your prompt support and guidance.

[sandippatel-incedo]

@edenhill
Want to use "latest" offset instead of "earliest" in consumer.c example program, Tried by changing only that setting but didn't work can you please guide?

[sandippatel-incedo]

@edenhill Can you please take a look & guide?

1. if i set enable.auto.commit=false, C consumer start with same committed message
   e.g
   step1 start c consumer and producer, it start with offset 100 & process till 1000 offset, terminate only c consumer application
   step2 producer is running so after C consumer stopped it reaches to offset 2000
   step3 restart c consumer, it start with offset 100, which i want to start with 2000 (latest offset)

[edenhill]

enable.auto.commit=false turns off comitting of messages, not reading committed offsets, it still does that. You will need to use assign() with the .offset field of each partition set to RD_KAFKA_OFFSET_END - if you use a subscribe()ing consumer you will need to set up a rebalance_cb and do this assign from within.