Php-rdkafka: php code cannot commit msg offset

[geopamplona]

Hi,

I am using the library to consume a Kafka queue, with this config:
$ topicConf-> set ('enable.auto.commit', 'false');

When I consume a message I commit manually:
$ consumer-> commit ($ message);

so far everything works correctly, I consume the queue and according to messages everything works fine.

I used a kafka cluster with 4 brokers, and now I'm trying to balance the cluster to turn off 3 brokers, leave the cluster with 1 broker and raise it again. When I remove 3 kafkas with kafkatools from linkedin everything works fine, the code in php continues to consume the only broker that is raised, i supose. When the other kafka brokers come back to life, everything works fine too, so php script consume queue apparently.

However, I realize that although the php code is able to consume the queue (the data flow does not stop), it cannot commit, or this commit is not being distributed well.

As I am doing to manage the cluster there should be issues of storage of issues or offsets.

The error comes when restarting the php code, suddenly it begins to consume old messages from when I stop committing.

This would be the graph of the situation:

I think it could be controlled from the php code; My question, is it possible to control with a callback or similar if the code cannot commit?
why if cannot commit de msg kafka send new msg?
How can I configure it to be more strict when confirming the message and maintain consistency in the order?
I just realized that I use an earlier version of the libraries; Do you recommend that my version be updated?

• PHP version: 7.2
• librdkafka version: 0.9.3
• php-rdkafka version: 3.0.5
• kafka version: Apache Kafka 1.1.0-cp1 from confluentinc/cp-kafka:4.1.0

rdkafka
rdkafka support => enabled
version => 3.0.5
build date => Jul 27 2018 10:55:21
librdkafka version (runtime) => 0.9.3
librdkafka version (build) => 0.9.3.0

[Steveb-p]

@geopamplona What consumer group ID did you assign to the consumer?

EDIT: librdkafka version 0.9.3 is 2-3 years old if I remember correctly. Phprdkafka actually will be dropping support for anything below 0.10 (as far as I remember, it might be even higher). You should update librdkafka to either version 0.11.6 (last before major overhaul) or 1.0.

[geopamplona]

i assign gropu-ib and client.id:

    $conf->set('group.id', 'client');
    $conf->set('client.id', 'client');

Does the problem have the same value?

Edit:
ok i will try upgrade libs, thanks

[Steveb-p]

No, but not assigning group.id would cause Kafka 2.0 to discard commit instructions (since it would not know what consumer group the process belongs to and consequently would start over each time you connect).

Try to upgrade first and see if it helps.

[geopamplona]

perfect, i will try upgrade first
I not using kafka 2.0, i using kafka 1.1.x
I will try next week,
thanks you very much !

[nick-zh]

@geopamplona besides what @Steveb-p said, i have the following inputs

My question, is it possible to control with a callback or similar if the code cannot commit?

This might help: https://arnaud.le-blanc.net/php-rdkafka/phpdoc/rdkafka-conf.setoffsetcommitcb.html
But you only need this if you use commitAsync, commit is synchronous and throws an exception on error

why if cannot commit de msg kafka send new msg?

Actually Kafka doesn't send messages, rather a consumer fetches them. Committing offsets to Kafka is like acknowledging a message. When you restart your PHP process, it will start consuming from the last offset you committed, which is actually a good thing.

How can I configure it to be more strict when confirming the message and maintain consistency in
the order?

If you can't handle a message (or a commit fails), you can't move to the next message (at least in that partition), if you do and commit that next's message offset, you "lose" your previous message.

[geopamplona]

I upgrade libs

 rdkafka support => enabled
 version => 3.1.2
 build date => Sep 30 2019 06:38:13
 librdkafka version (runtime) => 1.1.0
 librdkafka version (build) => 1.1.0.255

now I get this warning

%4|1569828611.678|CONFWARN|client#consumer-1| [thrd:app]: Configuration property auto.commit.enable is deprecated: [**LEGACY PROPERTY:** This property is used by the simple legacy consumer only. When using the high-level KafkaConsumer, the global `enable.auto.commit` property must be used instead]. If true, periodically commit offset of the last message handed to the application. This committed offset will be used when the process restarts to pick up where it left off. If false, theapplication will have to call `rd_kafka_offset_store()` to store an offset (optional). **NOTE:** There is currently no zookeeper integration, offsets will be written to broker or local file according to offset.store.method.

the error still occurs

Can you provide some example of a strict implementation of Kafka consumer? Or the only examples are those of the web?

If you can't handle a message (or a commit fails), you can't move to the next message (at least in that partition), if you do and commit that next's message offset, you "lose" your previous message.

something strange happens at this point;

during the error, a snapshot backup of the entire server is performed; Could this influence the connection between php code and kafka? or when confirming messages?

I'm still investigating

[nick-zh]

you are probably setting it for the default topic conf instead of just the conf (global), meaning you need to do this:
$conf->set('enable.auto.commit', 'false');

i would say a rough example looks something like this:

use RdKafka\Conf;
use RdKafka\KafkaConsumer;
use RdKafka\TopicConf;


$conf = new Conf();
$conf->set('group.id', 'consumer-group');
$conf->set('metadata.broker.list', 'kafka:9092');
$conf->set('enable.auto.commit', 'false');

$consumer = new KafkaConsumer($conf);

$consumer->subscribe(['topic-name']);

while (true) {
    $message = $consumer->consume(120 * 1000);
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            var_dump($message); //for debug purposes
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "No more messages; will wait for more\n";
            continue 2;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "Timed out\n";
            continue 2;
    default:
            throw new \RuntimeException($message->errstr(), $message->err);
            break;
    }
    //some business logic

   //this will throw an exception if the commit fails
   $consumer->commit($message);
}

Regarding

during the error, a snapshot backup of the entire server is performed; Could this influence the connection between php code and kafka? or when confirming messages?

I am not sure how this is being done, but basically this shouldn't influence your cluster / broker. If it does you might wan't to do this in a different way (mirrormaker, etc.).
My personal opinion is, a snapshot is kind of useless for a system that is built for streaming, backup should occur constantly, hence there are solutions like MirrorMaker, etc. that are able to replicate your cluster.

[geopamplona]

Thanks @nick-zh

I apply your code, but I have this Exception :

Exception 'RdKafka\Exception' with message 'Invalid argument: Specified Message has an error'

I know that taking a snapshot is not the best way, we are working on changing this

[nick-zh]

@geopamplona what are you passing to the commit call? it needs to be either an instance RdKafka\Message or RdKafka\TopicPartition as documented here https://arnaud.le-blanc.net/php-rdkafka/phpdoc/rdkafka-kafkaconsumer.commit.html
The error suggests you are passing something to commit that is not supported

Edit: The example above is just a rough example (and not to be used for production code), in the error cases you don't want to break but do something else most certainly.
To make the above example work, in the error cases you should probably do continue instead of break. I adjusted the example above accordingly

[geopamplona]

I think that fail when try commit timeout msg; because I am passing rdkafka/message object correctly:

Timeout for my consumer is 1 sec (1000ms); its very low

I have already managed to make it work, it was cause of the low timeout for the consumer.
But in case of timeout how will the code behave? will it fail because it cannot commit?
Maybe I should only commit when it's a message, inside swtich RD_KAFKA_RESP_ERR_NO_ERROR?

thanks

EDIT: i see that you change code; break for continue; i think that is better these

[nick-zh]

@geopamplona so actually what timeout means is: you didn't get a new message in x time (x being the timeout you set). So there is actually nothing to commit. This is more of a "soft" error same as end of partition and is totally fine. Most consumers probably don't care about timeout or eof and just continue trying to read new messages. To your question, yes only commit when RD_KAFKA_RESP_ERR_NO_ERROR and only after you have done everything you need to do with the message (business logic), after that, when you are "done" with the message, it is ok to commit.