Php-rdkafka: Question: Does Rdkafka handle hard and graceful shutdown?

[anam-hossain]

Hi guys
I would like to apologies if the question already answered.

We use Kubernetes for automating application deployment, scaling, and management. When Kubernetes wants to kill a pod it does the following:

• A SIGTERM signal is sent to the main process (PID 1) in each container, and a “grace period” countdown starts (defaults to 30 seconds).
• Upon the receival of the SIGTERM, each container should start a graceful shutdown of the running application and exit.
• If a container doesn’t terminate within the grace period, a SIGKILL signal will be sent and the container violently terminated.

I have two questions:

1. Does PHP Rdkafka handle the graceful shutdown process (SIGTERM)?
2. What happen when shutdown happened forcefully (SIGKILL)? Does the client lose any messages as the messages are not being fully processed?

I am using High-level consumer to receive the messages. Any example of how to handle this would be very helpful.

• PHP version: 7.3
  
  
  
  • librdkafka version: 1.2.2
  
  
  • php-rdkafka version: 4.0
  
  
  • kafka version: 2.2
  
  

Thanks,

[nick-zh]

Hey @anam-hossain

From what i read, you are only talking about consuming, right?
I think the safest approach would probably be to use ext-pcntl and let the application react to the SIGTERM.
On destruction of the consumer, rd_kafka_consumer_close is called, which according to
the documentation does the following:

This call will block until the consumer has revoked its assignment,
calling rebalance_cb if it is configured, committed offsets
to broker, and left the consumer group.
The maximum blocking time is roughly limited to session.timeout.ms.

So if the PHP process is already shutting down on SIGTERM you might be fine.
The worst case that can happen is, that you might read messages again,
depending on your usecase, this might be ok.
But it really depends if you commit the message immediately or after processing it.
Let me know if i can assist further with this matter or if you need more info.

Cheers,
Nick

[anam-hossain]

@nick-zh Thank you for the detail response. Could you able to check the code below to see I am on the right track or not?

<?php

use Exception;
use RdKafka\Conf;
use RdKafka\KafkaConsumer;

class ConsumerHandler
{
    protected $shouldQuit = false;

    /**
     * consume messages from kafka
     *
     * @param int $timeout "Timeout in milliseconds. Default 2 minutes"
     * @return void
     */
    public function consume(int $timeout = 120000)
    {
        $this->listenForSignals();

        $conf = new Conf();

        // Configure the group.id. All consumer with the same group.id will consume
        // different partitions.
        $conf->set('group.id', 'myConsumerGroup');

        // Initial list of Kafka brokers
        $conf->set('metadata.broker.list', '127.0.0.1');

        // 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
        $conf->set('auto.offset.reset', 'smallest');

        $consumer = new KafkaConsumer($conf);

        // Subscribe to topic 'test' and 'test2'
        $consumer->subscribe(['test', 'test2']);

        while (true) {
            try {
                $message = $consumer->consume($timeout);

                // $consumer->commitAsync($message);
                switch ($message->err) {
                    case RD_KAFKA_RESP_ERR_NO_ERROR:
                        var_dump($message);
                        break;
                    case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                        break;
                    case RD_KAFKA_RESP_ERR__TIMED_OUT:
                        break;
                    default:
                        throw new Exception($message->errstr(), $message->err);
                        break;
                }
            } catch (Exception $exception) {
                //
            }

            $this->reactToSignal();
        }
    }

    protected function listenForSignals()
    {
        pcntl_async_signals(true);

        // Interrupted (Ctrl-C is pressed)
        pcntl_signal(SIGINT, function () {
            $this->shouldQuit = true;
        });

        // Kill process was called
        pcntl_signal(SIGTERM, function () {
            $this->shouldQuit = true;
        });
    }

    protected function reactToSignal()
    {
        if ($this->shouldQuit) {
            exit(0);
        }
    }
}

Questions:

1. It seems messages are automatically committed with High-level consumer eventhough $consumer->commitAsync($message); not called.
   Is it expected behaviour? Is there any way we can manually commit the message?
2. Is it good practice to subscribe to multiple topics?
   $consumer->subscribe(['test', 'test2']);

[nick-zh]

@anam-hossain so far looks good to me.

It seems messages are automatically committed with High-level consumer

You want to set enable.auto.commit=false this way you have control over the offsets.

Is it good practice to subscribe to multiple topics

I'd say there is no silver bullet here. Depends on different factors, one of them being your topics. If you only have a few partitions with low / medium traffic, then it should be perfectly fine. If you have tons of partitions and traffic, you might want to use multiple consumers for a topic.

[anam-hossain]

@nick-zh Thank you very much.

[Steveb-p]

@nick-zh We might actually add a comment regarding enable.auto.commit in main readme, alongside termination signals handling sample.

[nick-zh]

@Steveb-p agreed, i think this would be a good addition for docs and readme, will take care of it this week :+1:

[anam-hossain]

@nick-zh I have one more question. Sorry to bother you. We have a producer which push message to Kafka when HTTP request comes in via Nginx (public endpoint). By design, this producer can not be in a job/queue.

Can we use the pcntl_sigprocmask(SIG_BLOCK, array(SIGIO)); with Nginx process? Does it terminate the PHP process before executing the rest of the PHP code?

According to the doc, the following code does this
This allows a PHP process / request to send messages ASAP and to terminate quickly.

<?php

$conf = new \RdKafka\Conf();
$conf->set('socket.timeout.ms', 50); // or socket.blocking.max.ms, depending on librdkafka version
if (function_exists('pcntl_sigprocmask')) {
    pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
    $conf->set('internal.termination.signal', SIGIO);
} else {
    $conf->set('queue.buffering.max.ms', 1);
}

$producer = new \RdKafka\Producer($conf);

[nick-zh]

@anam-hossain this is perfectly fine to do, please don't forget to use flush in the producer (needed since rdkafka:4.x)