Php-rdkafka: increasing threads using setErrorCb on producer

[KernelFolla]

• PHP version: 7.4
• librdkafka version: 0.11.6.255
• php-rdkafka version: 5
• kafka version: 2.2.1

We have an always running service that logs everything in a kafka producer.
It increases number of threads on each flush until we end with a full cpu and thread limited.
Commenting setErrorCb everything works like a charm but we would like to handle those errors

<?php

use RdKafka\Conf;
use RdKafka\Producer;
use RdKafka\ProducerTopic;

class KafkaHelper
{
    private $timeout;
    private ProducerTopic $topic;
    private Producer $producer;

    public function __construct($config)
    {
        $this->timeout = $config['timeout'] ?? env('KAFKA_TIMEOUT', 3000);

        $conf = new Conf();

        $conf->setErrorCb(function ($kafka, int $errorCode, string $errorMsg) {
            throw new \RuntimeException($errorMsg, $errorCode);
        });

        $this->producer = new Producer($conf);
        $this->producer->addBrokers(
            $config['brokers'] ?? config('services.kafka.brokers')
        );

        $this->topic = $this->producer->newTopic(
            $config['topic']
        );
    }


    public function send(string $msg): void
    {
        $this->add($msg);
        $this->flush();
    }

    public function add(string $msg)
    {
        $this->topic->produce(RD_KAFKA_PARTITION_UA, 0, $msg);
    }

    public function flush()
    {
        $this->producer->flush($this->timeout);
    }
}

[nick-zh]

Can you check if this happens with newer librdkafka versions as well (i would use librdkafka:1.x)?
librdkafka:0.11.6 is quite old, many bugs have been fixed since then.
As a side note, for performance, you probably only want to flush when terminating the producer, not after every message :v:

[KernelFolla]

(in my defense😃) I've used librdkafka:0.11.6 because i did apt install librdkafka-dev in docker using the debian based image php:7.4
anyway I've just tested with 1.7 (git clone + config make make install + reinstalling rdkafka with pecl and verified version with php -i and issue is the same.

about the flush, in this case I flush after every message to be sure of its delivery, still not sure if it's the best way😃

[nick-zh]

Thanks for the update, i'll try to have a look this weekend regarding the threads ✌️

[nick-zh]

I had a quick look, so i am always using this playground to reproduce things (i used pure-php).
I changed the docker-compose.yml to reflect php:7.4 and librdkafka:1.7

I changed the producer.php to include flush after every message and added an error callback.
For me the thread count stays constant when i check the output of ps -T and cpu stays under 5%.

Producer code:

<?php

use RdKafka\Conf;
use RdKafka\Message;
use RdKafka\Producer;
use RdKafka\TopicConf;

error_reporting(E_ALL);

$conf = new Conf();
// will be visible in broker logs
$conf->set('client.id', 'pure-php-producer');
// set broker
$conf->set('metadata.broker.list', 'kafka:9096');
// set compression (supported are: none,gzip,lz4,snappy,zstd)
$conf->set('compression.codec', 'snappy');
// set timeout, producer will retry for 5s
$conf->set('message.timeout.ms', '5000');
//If you need to produce exactly once and want to keep the original produce order, uncomment the line below
//$conf->set('enable.idempotence', 'true');

// This callback processes the delivery reports from the broker
// you can see if your message was truly sent, this can be especially of importance if you poll async
$conf->setDrMsgCb(function (Producer $kafka, Message $message) {
    if (RD_KAFKA_RESP_ERR_NO_ERROR !== $message->err) {
        $errorStr = rd_kafka_err2str($message->err);

        echo sprintf('Message FAILED (%s, %s) to send with payload => %s', $message->err, $errorStr, $message->payload) . PHP_EOL;
    } else {
        // message successfully delivered
        echo sprintf('Message sent SUCCESSFULLY with payload => %s', $message->payload) . PHP_EOL;
    }
});

$conf->setErrorCb(function ($kafka, int $errorCode, string $reason) {
    echo sprintf('Code %d, reason %s', $errorCode, $reason) . PHP_EOL;
});

// SASL Authentication
// can be SASL_PLAINTEXT, SASL_SSL
// conf->set('security.protocol', '');
// can be GSSAPI, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER
// $conf->set('sasl.mechanisms', '');
// $conf->set('sasl.username', '');
// $conf->set('sasl.password', '');
// default is none
// $conf->set('ssl.endpoint.identification.algorithm', 'https');


// SSL Authentication
//$conf->set('security.protocol', 'ssl');
//$conf->set('ssl.ca.location', __DIR__.'/../keys/ca.pem');
//$conf->set('ssl.certificate.location', __DIR__.'/../keys/kafka.cert');
//$conf->set('ssl.key.location', __DIR__.'/../keys/kafka.key');

// Add additional output if you need to debug a problem
// $conf->set('log_level', (string) LOG_DEBUG);
// $conf->set('debug', 'all');

$producer = new Producer($conf);
// initialize producer topic
$topic = $producer->newTopic('pure-php-test-topic');
// Produce 10 test messages
$amountTestMessages = 10000000;

// Loop to produce some test messages
for ($i = 0; $i < $amountTestMessages; ++$i) {
    // Let the partitioner decide the target partition, default partitioner is: RD_KAFKA_MSG_PARTITIONER_CONSISTENT_RANDOM
    // You can use a predefined partitioner or write own logic to decide the target partition
    $partition = RD_KAFKA_PARTITION_UA;

    //produce message with payload, key and headers
    $topic->producev(
        $partition,
        RD_KAFKA_MSG_F_BLOCK, // will block produce if queue is full
        sprintf('test message-%d',$i),
        sprintf('test-key-%d', $i),
        [
            'some' => sprintf('header value %d', $i)
        ]
    );
    echo sprintf('Queued message number: %d', $i) . PHP_EOL;

    // Poll for events e.g. producer callbacks, to handle errors, etc.
    // 0 = non-blocking
    // -1 = blocking
    // any other int value = timeout in ms
    $producer->flush(3000);
}

// Shutdown producer, flush messages that are in queue. Give up after 20s
$result = $producer->flush(20000);

if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
    echo 'Was not able to shutdown within 20s. Messages might be lost!' . PHP_EOL;
}

[nick-zh]

I am taking the liberty to close this, since i didn't get any further feedback. If you can provide an example in the playground how to reproduce this, please feel to re-open :v: