[kafka] Useless while?

[TiMESPLiNTER]

I have a question regarding these lines in the kafka consumer:

https://github.com/php-enqueue/enqueue-dev/blob/master/pkg/rdkafka/RdKafkaConsumer.php#L103-L111

        if ($timeout > 0) {
            $message = $this->doReceive($timeout);
        } else {
            while (true) {
                if ($message = $this->doReceive(500)) {
                    break;
                }
            }
        }

just out of curiosity: why isn't this written like this?

        if ($timeout > 0) {
            $timeout = 500;
        }

        $message = $this->doReceive($timeout);

[Steveb-p]

It's done to allow indefinite wait for message (when $timeout is 0). AFAIR phprdkafka always requires a timeout for consumption call.

It could be replaced with early return though, and the same applies to doReceive method.

[Steveb-p]

@TiMESPLiNTER this is even more relevant if you consider possible callbacks set for - for example - rebalancing operations on the underlying consumer: https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/rdkafka-kafkaconsumer.consume.html.

Note that there are two distinct operations - one with timeout explicitly set, and second when timeout was either set to 0 or not passed at all.

[TiMESPLiNTER]

It's done to allow indefinite wait for message (when $timeout is 0). AFAIR phprdkafka always requires a timeout for consumption call.

That's true. That's why I suggest to set it to 500 if you pass 0 and omit the endless while. It removes the busy wait which imo is never a good option. For example if you use it with symfony messenger and you consume multiple (kafka) transports with the same messenger:consume command it blocks forever if there's no new message int this very transport and therefor never consumes from others. I also cannot imagine any use case where you need a forced busy wait. If you just consume from one topic and 0 timeout defaults to 500 its anyway the same as the while solution currently in place. Or am I wrong?

[Steveb-p]

If you just consume from one topic and 0 timeout defaults to 500 its anyway the same as the while solution currently in place.

It doesn't do that.

If $timeout argument is set to 0, then receive method is turned into a blocking method - that's the while part. It will constantly try to get a message until it does.
Counterpart to that is passing a $timeout value higher than 0. In this case it will pass exactly this value to KafkaConsumer.

This way, adapters such as the one done for Symfony Messenger can provide a timeout value to allow itself to switch to different transport if there are no messages. At the same time, users that do not use such an adapter or use the library directly (without bundle) can call this method without arguments (or 0), providing ease of use. Otherwise they would need to put while in their own code. It's common use enough to warrant this IMO.

That's why I suggest to set it to 500 if you pass 0 and omit the endless while

This would be change in behavior as I said. Currently users that do not pass value get a blocking method until a message is received.

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.