Hi, I want to disable the offset auto commit feature, in php code,
$topicConf->set("auto.commit.enable", false);
it says 'Bool configuration property "auto.commit.enable" cannot be set to empty value',
I also try to change this configuration in kafka's configuration,
auto.commit.enable = false
it doesn't work either.
Hi
You have to specify the value as a string: $topicConf->set("auto.commit.enable", "false");
Yeah, I tried this way before in high level consumer
$topicConf->set("auto.commit.enable", "false");
But the offset still auto commit.
I also tried to specify the config below to make the auto-commit period longer:
$topicConf->set("auto.commit.interval.ms", 86400);
Unfortunately, it does't work either.
Here is my code
$conf = new RdKafka\Conf();
// Set a rebalance callback to log partition assignemts (optional)
...
// Configure the group.id. All consumer with the same group.id will consume
// different partitions.
$conf->set('group.id', 'consumerGroup2');
// Initial list of Kafka brokers
$conf->set('metadata.broker.list', "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
$topicConf = new RdKafka\TopicConf();
// Set where to start consuming messages when there is no initial offset in
// offset store or the desired offset is out of range.
// 'smallest': start from the beginning
$topicConf->set("auto.commit.enable", 'false');
//$topicConf->set("enable.auto.commit", 'false');
$topicConf->set("auto.commit.interval.ms", 86400);
$topicConf->set('auto.offset.reset', 'smallest');
// Set the configuration to use for subscribed/assigned topics
$conf->setDefaultTopicConf($topicConf);
$consumer = new RdKafka\KafkaConsumer($conf);
// Subscribe to topic 'test'
$consumer->subscribe(['testReplicaCreate']);
echo "Waiting for partition assignment... (make take some time when\n";
echo "quickly re-joinging the group after leaving it.)\n";
while (true) {
$message = $consumer->consume(120*1000);
switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR:
var_dump($message);
break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF:
echo "No more messages; will wait for more\n";
break;
case RD_KAFKA_RESP_ERR__TIMED_OUT:
echo "Timed out\n";
break;
default:
var_dump($message);
throw new \Exception($message->errstr(), $message->err);
break;
}
}
Hope for your answer.Thanks
@arnaud-lb
OK, I test more.
When I use the low level consumer, this config $topicConf->set("auto.commit.enable", 'false'); doesn't work either. The low level consumer offset auto-commit period is much more longer than high level consumer. But when I specify $topicConf->set('auto.commit.interval.ms', 100);, it can work as I expected.
However I can't commit the offset immediately by using$topic->offsetStore($message->partition, $message->offset);.
No matter what storage I use (file or broker), it has the same problem
Here is my code:
$conf = new RdKafka\Conf();
// Set the group id. This is required when storing offsets on the broker
$conf->set('group.id', 'myConsumerGroup');
$rk = new RdKafka\Consumer($conf);
$rk->addBrokers("127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
$topicConf = new RdKafka\TopicConf();
$topicConf->set("auto.commit.enable", "false");
//$topicConf->set('auto.commit.interval.ms', 100);
// Set the offset store method to 'file'
//$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.sync.interval.ms', 0);
// Alternatively, set the offset store method to 'broker'
$topicConf->set('offset.store.method', 'broker');
// Set where to start consuming messages when there is no initial offset in
// offset store or the desired offset is out of range.
// 'smallest': start from the beginning
$topicConf->set('auto.offset.reset', 'smallest');
$topic = $rk->newTopic("testReplicaCreate", $topicConf);
// Start consuming partition 0
$topic->consumeStart(4, RD_KAFKA_OFFSET_STORED);
while (true) {
$message = $topic->consume(4, 120*10000);
switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR:
var_dump($message);
$topic->offsetStore($message->partition, $message->offset);
break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF:
echo "No more messages; will wait for more\n";
break;
case RD_KAFKA_RESP_ERR__TIMED_OUT:
echo "Timed out\n";
break;
default:
throw new \Exception($message->errstr(), $message->err);
break;
}
}
My kafka version is 0.10.0.1
Sorry for my poor English.
Hope for your answer. Thanks
kafka 9.0.1 has same problem
Hey @GodTou, have you find out how to prevent autocommitting the offsets?
Most helpful comment
OK, I test more.
When I use the low level consumer, this config
$topicConf->set("auto.commit.enable", 'false');doesn't work either. The low level consumer offset auto-commit period is much more longer than high level consumer. But when I specify$topicConf->set('auto.commit.interval.ms', 100);, it can work as I expected.However I can't commit the offset immediately by using
$topic->offsetStore($message->partition, $message->offset);.No matter what storage I use (file or broker), it has the same problem
Here is my code:
My kafka version is 0.10.0.1
Sorry for my poor English.
Hope for your answer. Thanks