Php-rdkafka: Producer is not sending the msg

[wesleywillians]

• PHP version: 7.3
• php-rdkafka version: 4
• kafka version: 5.3.1-ccs (Commit:03799faf9878a999)

Hello guys,

Since the new version, my producer is not able to send new msgs.
I set the debug to all and got this msg: https://gist.github.com/wesleywillians/b64806d00b8e5c825c169565022da093

I'm using pretty much the same example of the documentation.

Any idea?

[nick-zh]

@wesleywillians can you tell me which librdkafka version you are using?

[Steveb-p]

@wesleywillians Looks to me like you did not heed the warning in release notes.

@nick-zh Looks like we need to update the samples :)

Logs you've shown indicate that Producer was destroyed as part of the shutdown process of PHP. Previously messages in queue prevented shutdown from completing, which caused many issues. 4.0 no longer does that. You will need to check queue state and keep the process up until it's empty.

$producer->produce(RD_KAFKA_PARTITION_UA, 0, $message);

while ($producer->getOutQLen() > 0) {
  $producer->poll(0);
}

...or something like that. @nick-zh can you confirm that this is how it should now look like?

See https://arnaud.le-blanc.net/php-rdkafka/phpdoc/rdkafka.getoutqlen.html
See https://arnaud.le-blanc.net/php-rdkafka/phpdoc/rdkafka.poll.html

[nick-zh]

@nick-zh i think documentation should be inplace, in the readme it is this section: https://github.com/arnaud-lb/php-rdkafka#proper-shutdown
and in the documentation, below the producer example is the shutdown example:
https://arnaud.le-blanc.net/php-rdkafka/phpdoc/rdkafka.examples-producer.html
If something is unclear or badly formulated, please let me know so we can adjust this accordingly.
As @Steveb-p correctly pointed out, it is really important to stick to the new shutdown procedures to make sure all remaining messages were sent (or acknowledge that inflight messages haven't).

[Steveb-p]

We should emphasize that every time we're showing producer in the docs. I'll revisit it when I've got my time off next week, especially since I need to look at enqueue with Kafka finally. Otherwise Timesplinter will kill me, he does all the PRs lately :)

[nick-zh]

@Steveb-p i agree, maybe we could spread this more through out the doc, don't worry i can have a look.
PS: let me know if i can assist on the enqueue side

[nick-zh]

@Steveb-p @wesleywillians i opened 2 PRs for improvements, let me know if you feel it should be added somewhere else as well.

[wesleywillians]

@wesleywillians Looks to me like you did not heed the warning in release notes.

@nick-zh Looks like we need to update the samples :)

Logs you've shown indicate that Producer was destroyed as part of the shutdown process of PHP. Previously messages in queue prevented shutdown from completing, which caused many issues. 4.0 no longer does that. You will need to check queue state and keep the process up until it's empty.

$producer->produce(RD_KAFKA_PARTITION_UA, 0, $message);

while ($producer->getOutQLen() > 0) {
  $producer->poll(0);
}

...or something like that. @nick-zh can you confirm that this is how it should now look like?

See https://arnaud.le-blanc.net/php-rdkafka/phpdoc/rdkafka.getoutqlen.html
See https://arnaud.le-blanc.net/php-rdkafka/phpdoc/rdkafka.poll.html

Thank you very much. Should I use flush or poll to terminate my producer.
Do I need to add this while if I'm using flush?

[wesleywillians]

@Steveb-p @wesleywillians i opened 2 PRs for improvements, let me know if you feel it should be added somewhere else as well.

I think its much better. Basically I just need to add the flush after the message has been produced.

Will test here and let you know.

[Steveb-p]

Should I use flush or poll to terminate my producer.

Use poll if you intend to still send more messages. Use flush as part of preparing the program/application to terminate.

Do I need to add this while if I'm using flush?

Flush is a blocking operation, and accepts maximum timeout in millisends. After that time internal Kafka client will be purged of messages (right @nick-zh ?).
While is used in poll example to show how you can do something else while still polling. Theoretically you could just set timeout parameter to the maximum amount of milliseconds you intend to wait.

[nick-zh]

Using poll:
Poll triggers your callbacks (if you have any). So it is vital if you use callbacks, to see if you
had any errors, etc. While you are producing, you should call this at regular intervals, otherwise
your internal queue might fill up (and in the end result in RD_KAFKA_RESP_ERR__QUEUE_FULL`).

Using flush:
Flush waits until all outstanding requests (e.g. produce) are completed.
On success it returns RD_KAFKA_RESP_ERR_NO_ERROR otherwise RD_KAFKA_RESP_ERR__TIMED_OUT.
So i would suggest to try to flush until you get RD_KAFKA_RESP_ERR_NO_ERROR

I hope this helps.

@Steveb-p to your question, if flush results in timeout but you still terminate, you can lose messages, yes.

[Steveb-p]

@nick-zh I assumed (wrong) that flush lead to destruction of internal Kafka instances and was meant as an ending operation before shutdown.

What is the difference between polling and flushing? Since flush calls poll internally (and trigger callbacks), there doesn't seem to be a lot of difference.

[nick-zh]

@Steveb-p i haven't checked the internals of flush it is meant as a blocking operation before you close your application. Before the extension did this (sort of) in the destructor, which was buggy though.
I checked and you are right, flush calls poll as well, so we don't need to do it after flush :+1:
EDIT: i adjusted my old comment, so nobody is misled, thx steve

[wesleywillians]

Using poll:
Poll triggers your callbacks (if you have any). So it is vital if you use callbacks, to see if you
had any errors, etc. While you are producing, you should call this at regular intervals, otherwise
your internal queue might fill up (and in the end result in RD_KAFKA_RESP_ERR__QUEUE_FULL`).

Using flush:
Flush waits until all outstanding requests (e.g. produce) are completed.
On success it returns RD_KAFKA_RESP_ERR_NO_ERROR otherwise RD_KAFKA_RESP_ERR__TIMED_OUT.
So i would suggest to try to flush until you get RD_KAFKA_RESP_ERR_NO_ERROR

I hope this helps.

@Steveb-p to your question, if flush results in timeout but you still terminate, you can lose messages, yes.

So, basically the difference between poll and flush is the callbacks. Because even flushing I do need to add a while(true) in order to guarantee that all messages were delivered. Correct? Or you recommend that I always use poll after I run a produce method and forget the flush. (as I can see, poll also calls flush, right?)

[Steveb-p]

So, basically the difference between poll and flush is the callbacks.

Both will trigger callbacks.

Because even flushing I do need to add a while(true) in order to guarantee that all messages were delivered. Correct? Or you recommend that I always use poll after I run a produce method and forget the flush. (as I can see, poll also calls flush, right?)

Nick corrected his statement: you don't need to poll after flushing. And it's the other way around: flush calls poll internally.

As a general guideline:

flush it is meant as a blocking operation before you close your application

So you are expected to run poll in your application to prevent Kafka client thread from having issues with overflow (due to callbacks, errors, or anything else that goes internally into queue).
Ideally, you would run poll at some interval. If you have no event loop in your application, then you might poll after sending message until queue (getQueueLen) returns 0 (which will basically turn message sending into a synchronous operation, where you wait for it to finish). Or decide for yourself what you want to do and when you want to wait - it all depends on your use case.

[nick-zh]

@wesleywillians sry for the confusion, flush calls poll, so ultimately probably if you use flush you should be fine. Depends on your producer, if you produce a lot of messages in one go, you should call poll from time to time to handle callbacks. If you only produce a few messges and then end, flush should be fine.
Edit: Steve was faster, sry for the duplicate answer :smile: , he explained it perfectly above :+1:

[wesleywillians]

@nick-zh and @Steveb-p Thank you very much for the help.

I'm creating an abstraction layer (lib) in order to facilitate the usage of the php-rdkafka based on middleware. Now with the versoin 4, I need to adapt some methods (including the setDefault that was deprecated).

I'll look into your documentation and if I find something to improve I send a PR. Ok?

[Steveb-p]

I'll look into your documentation and if I find something to improve I send a PR. Ok?

You're more than welcome :)