Php-rdkafka: How to configure asynchrony in PHP?

[niwsmbulai1989]

• PHP version:7.2
• librdkafka version:4.0.3
• php-rdkafka version:1.2.1
• kafka version:2.5

$conf = new \RdKafka\Conf();
$rk = new \RdKafka\Producer($conf);
$rk->addBrokers(self::$broker_list);
$cf = new \RdKafka\TopicConf();
$rk->newTopic(self::$topic, $cf);
$topic->produce(RD_KAFKA_PARTITION_UA, self::$partition, $message);

asynchrony in java
Properties props = new Properties();
props.put('producer.type', 'sync');

But I don't see such a configuration in the rdkafka PHP document

[nick-zh]

@niwsmbulai1989 so there is poll it can be done either blocking (by passing -1) or non-blocking (by passing 0 or a timeout in ms). If you want a sync producer, either call flush(-1) after produce or poll indefinitely for an event:

$producer->poll(-1);

[realpeanut]

https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/rdkafka-producertopic.produce.html
public RdKafka\ProducerTopic::produce ( integer $partition , integer $msgflags , string $payload [, string $key = NULL ] ) : void

Produce and send a single message to broker.

This is a async and non-blocking. @niwsmbulai1989

[niwsmbulai1989]

@nick-zh @plediii

sync

$conf = new \RdKafka\Conf();
$rk = new \RdKafka\Producer($conf);
$rk->addBrokers(self::$broker_list);
$cf = new \RdKafka\TopicConf();
$rk->newTopic(self::$topic, $cf);
$topic->produce(RD_KAFKA_PARTITION_UA, self::$partition, $message);
$rk->flush(-1);
$rk->poll(-1);

Async
$conf = new \RdKafka\Conf();
$rk = new \RdKafka\Producer($conf);
$rk->addBrokers(self::$broker_list);
$cf = new \RdKafka\TopicConf();
$rk->newTopic(self::$topic, $cf);
$topic->produce(RD_KAFKA_PARTITION_UA, self::$partition, $message);

Take a look at my code synchronization and asynchrony, right?

[nick-zh]

As @realpeanut pointed out correctly, the produce call on its own is async, BUT depending on how you use poll, the message will be handled sync or async. Now let's see why that is:

Sync Produce: You call produce to queue / send your message (async) but with poll(-1) you will wait until you get the delivery report for that message, this is what makes it synchronous.

$topic->produce(RD_KAFKA_PARTITION_UA, self::$partition, $message);
$rk->poll(-1);

Async produce: You call produce to queue / send your message (async) and with poll(0) you will NOT wait until you get the delivery report for that message, this is what makes it asynchronous.

$topic->produce(RD_KAFKA_PARTITION_UA, self::$partition, $message);
$rk->poll(0);

Producing synchronous is mostly done, when you want to be sure the message has been sent in this instant and when you want to fail / handle errors immediately if it wasn't successful. It is a bit slower of course, since you have some overhead of waiting for message events.

Producing asynchronous is mostly done, when you do not care about the immediate result, but want to achieve higher throughput. It poses the challenge that you need to handle failures asynchronously as well.

So decide which is more important to you.
As closing note, flush is called on producer shutdown to make sure, all outstanding messages will be sent. In the sync case everything should already be sent, but it doesn't hurt to call it anyway, since it will be really fast if there are no more outstanding messages.
Hope this helps.

[niwsmbulai1989]

Call to undefined method RdKafka\\Producer::asyncCommit()

$conf = new \RdKafka\Conf();
$rk = new \RdKafka\Producer($conf);
$rk->addBrokers(self::$broker_list);
$cf = new \RdKafka\TopicConf();
$topic=$rk->newTopic(self::$topic, $cf);
$topic->produce(RD_KAFKA_PARTITION_UA, self::$partition, $message);
$rk->flush(0);
$rk->poll(0);
$rk->asyncCommit($message);

@nick-zh why ?

[nick-zh]

@niwsmbulai1989 sry typo, it's commitAsync <- also i editet the answers, since commit is only needed for consumers, sry i got carried away, too early on a sunday 😄
But flush(0) is not really needed before shutdown in the async case. You can do it on producer shutdown.

[niwsmbulai1989]

Call to undefined method RdKafka\\Producer::commitAsync()

Still not ？
@nick-zh

[niwsmbulai1989]

  $conf = new \RdKafka\Conf();
  $rk = new \RdKafka\Producer($conf);
  $rk->addBrokers(self::$broker_list);
  $cf = new \RdKafka\TopicConf();
  $topic=$rk->newTopic(self::$topic, $cf);
  $topic->produce(RD_KAFKA_PARTITION_UA, self::$partition, $message);
  $rk->flush(0);
  $rk->poll(0);
  $rk->commitAsync($message);

@nick-zh This is the complete code, but still error!

[Steveb-p]

commit and it's sibling, commitAsync are methods that are used high-level consumer. They are used to inform Kafka broker that message has been processed and acknowledged (when auto-commit is disabled).

Your code is already asynchronous at this point: you're not waiting for messages to be produced. That's what flush(0) and poll(0) do. Actually they both do similar things: poll is part of flush afaik. Just remove the last line:

  $conf = new \RdKafka\Conf();
  $rk = new \RdKafka\Producer($conf);
  $rk->addBrokers(self::$broker_list);
  $cf = new \RdKafka\TopicConf();
  $topic=$rk->newTopic(self::$topic, $cf);
  $topic->produce(RD_KAFKA_PARTITION_UA, self::$partition, $message);
  $rk->flush(0);
  $rk->poll(0);
- $rk->commitAsync($message);

Remember, that phprdkafka will not attempt to finish producing messages when PHP process is terminated - in this case you are responsible for waiting and handling message delivery. In this case flush(-1) and / or poll(-1) should be used. You can also pass a timeout instead to control how long you intend to wait.

[niwsmbulai1989]

@Steveb-p I'm a bit confused. If I delete the last line of code, what kind of synchronization it should be!

[Steveb-p]

When produce is called message is put into background thread (in librdkafka).

At this point message is already being sent.

poll method controls exactly when you want to wait. When called with -1 it means - wait as long as possible. 0 means - just check queue and call registered callbacks, if any.

This practically makes -1 synchronous, while 0 works as if it was asynchronous.

[niwsmbulai1989]

@Steveb-p Very good, thank you! In addition, I have another question: how to judge whether it has been consumed? I need to write the business logic code after consuming again!

[nick-zh]

@niwsmbulai1989 ah yeah totally my bad as well, it's too early for me, as @Steveb-p pointed out correctly, commit is consumer stuff. apologies. I will correct the answers, so others don't get confused as well.

[nick-zh]

how to judge whether it has been consumed?

Can you maybe elaborate what you are trying to achieve?

[niwsmbulai1989]

I found a problem. When flushis less than or equal to 0, I can't receive the message!
No change was found by the following command, why?
bin/kafka-consumer-groups.sh --describe --bootstrap-server kafka-1.default.svc.cluster.localhost:9092 --group elk
@Steveb-p @nick-zh

[nick-zh]

please check the documentation and the examples on how to use flush:
https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/book.rdkafka.html
you can also find additional examples here:
https://github.com/nick-zh/php-kafka-examples
I will add some more comments to my examples later today.

but basically flush(0) will not do much, choose a higher timeout and also check the return code

[niwsmbulai1989]

I'm looking forward to more examples from you. I hope @ I can have a look @nick-zh

[nick-zh]

@niwsmbulai1989 if you feel a use case is missing or something is unclear, let me know. What i can say is, all the examples are not perfect. Every consumer / producer is very specific to the use case that your application needs to cover. So the examples are just a little help on how to do things.

[niwsmbulai1989]

@Steveb-p @nick-zh How to delete messages after Kafka consumption

[nick-zh]

You don't, this is what Kafka is all about, it's a streaming service. You can read about kafka retention time and kafka compaction to gain more insight about what is possible.

[Steveb-p]

@niwsmbulai1989 short answer: you don't.

It's a design choice that comes from topics in Kafka being represented by logs. All you do instead is acknowledge that a message was read and move to the next one.
There exist concepts like log compacting and maximum log size, but their purpose is to reduce size occupied on disk, not to remove particular messages.

Familiarize yourself with what Kafka is good for and what it is bad for, and then decide for yourself if it is a good tool for the task you set yourself to do. For example, if you intend to build a chat application, Kafka is definitely a bad choice.

As for the issue you've mentioned.

When flushis less than or equal to 0, I can't receive the message!

Do make sure that PHP process has enough time to actually send the message over to the Kafka broker. If you flush(0) and then terminate instantly (reaching end of code, or explicitly exit()) PHP process will be terminated alongside with the background thread that was responsible for sending message (it is possible that the message would still be received, albeit slim).
If you pass -1, it waits, and it isn't received, then it is possible that an error occurs - you will need to use error callbacks to see what happened (see https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/rdkafka-conf.seterrorcb.html).

EDIT: Dammit @nick-zh, you win this time! :smile:

[niwsmbulai1989]

two question
1、How do I know if the message has been consumed?
2、After I confirm the message, if my business code logic fails, how can I send the current message back to the queue and reuse it?

two question is very important！

[nick-zh]

1、How do I know if the message has been consumed?

you the developer will know because you can monitor the lag of the offsets of your consumer group

After I confirm the message, if my business code logic fails, how can I send the current message back to the queue and reuse it?

You should only commit, once all the business logic of your application has successfully processed the message.

I think you should familiarize yourself with the core concepts of Apache Kafka, because these questions are not really related to the extension, but more on how to use Kafka.
Confluent provides some good posts / videos on Kafka fundamentals

[Steveb-p]

I think you should familiarize yourself with the core concepts of Apache Kafka, because these questions are not really related to the extension, but more on how to use Kafka.

^ This.

@niwsmbulai1989 from the Producer point of view you have no knowledge if a message was or ever will be consumed. Consumers will acknowledge messages to Kafka, which means they confirm that they have received and handled the message.

Messages do not dissappear from Kafka. This means you never "re-queue" them. There is no need to. If a Consumer fails all you have to do is to not acknowledge (i.e. confirm) the message. This will cause that the next time a Consumer of a particular group starts it will receive the same message, again.

Please read more about what Kafka is all about, because the issues you've raised now can be easily found in introductions to Kafka, and are part of the basic design of it.

[niwsmbulai1989]

I'm a little confused about the above confirmation,How do consumers confirm messages like Kafka?

[niwsmbulai1989]

@nick-zh @Steveb-p How to inform Kafka that the message has been sent successfully？

[nick-zh]

Consumers commit offsets, this is the same as acknowledge for other messaging systems.

How to inform Kafka that the message has been sent successfully

You don't Kafka informs you, that it was received or rejected.

I will close this issue, since your initial question was answered. The last questions are more about Kafka itself and i think you need more guidance in general about Kafka / patterns. While i am happy to help out more (feel free to reach out), i don't think this is the right place.
Also as mentioned before, there are tons of good resources out there from Confluent and others that will be very useful.