Php-rdkafka: Exception during producer shutdown since 4.0.3 (race condition, RdKafka\Kafka::__construct() has not been called)

[sofire]

A few days ago, I added kafka to my project, and deployed in product env, and some error logs occurred (log in below).

I'm not sure this fault caused by rdkafka, but it doesn't happen before. If i remove the kafka code, the error disappeared.
I'd write a simple test script, but doesn't reproduce the same error.

Can someone help me to find the real reason.
Thanks.

• PHP version: 7.2.31
• librdkafka version: 1.4.0
• php-rdkafka version: 4.0.3
• kafka version: kafka_2.12-2.4.1

[17-May-2020 10:50:49] WARNING: [pool www] child 17854 said into stderr: "NOTICE: PHP message: PHP Fatal error:  Uncaught Exception: RdKafka\Kafka::__construct() has not been called in [no active file]:0"
[17-May-2020 10:50:49] WARNING: [pool www] child 17854 said into stderr: "Stack trace:"
[17-May-2020 10:50:49] WARNING: [pool www] child 17854 said into stderr: "#0 {main}"
[17-May-2020 10:50:49] WARNING: [pool www] child 17854 said into stderr: "  thrown in [no active file] on line 0"
[17-May-2020 10:50:49] WARNING: [pool www] child 17854 said into stderr: "[no active file](0) : Fatal error - Uncaught Exception: RdKafka\Kafka::__construct() has not been called in [no active file]:0"
[17-May-2020 10:50:49] WARNING: [pool www] child 17854 said into stderr: "Stack trace:"
[17-May-2020 10:50:49] WARNING: [pool www] child 17854 said into stderr: "#0 {main}"
[17-May-2020 10:50:49] WARNING: [pool www] child 17854 said into stderr: "  thrown"
[17-May-2020 10:50:49] WARNING: [pool www] child 17854 said into stderr: "/home/www/softwares/php-7.2.31/ext/opcache/ZendAccelerator.c:652: accel_rep
lace_string_by_process_permanent: Assertion `!((char*)(str) >= (accel_shared_globals->interned_strings_start) && (char*)(str) < (accel_shared_globals
->interned_strings_end))' failed."
[17-May-2020 10:50:50] WARNING: [pool www] child 17854 exited on signal 6 (SIGABRT - core dumped) after 2750.439072 seconds from start
[17-May-2020 10:50:50] NOTICE: [pool www] child 25102 started

#0  0x00007f296baf72c7 in raise () from /lib64/libc.so.6
#1  0x00007f296baf89b8 in abort () from /lib64/libc.so.6
#2  0x00007f296baf00e6 in __assert_fail_base () from /lib64/libc.so.6
#3  0x00007f296baf0192 in __assert_fail () from /lib64/libc.so.6

#4  0x00007f2969753f1b in accel_replace_string_by_process_permanent (str=0x7f295c78d8b0)
    at /home/www/softwares/php-7.2.31/ext/opcache/ZendAccelerator.c:652
#5  0x00007f2969753606 in accel_copy_permanent_strings (new_interned_string=0x7f2969753e49 <accel_replace_string_by_process_permanent>)
    at /home/www/softwares/php-7.2.31/ext/opcache/ZendAccelerator.c:528
#6  0x00007f2969753fee in accel_use_permanent_interned_strings () at /home/www/softwares/php-7.2.31/ext/opcache/ZendAccelerator.c:679

#7  0x00007f2969759194 in accel_shutdown () at /home/www/softwares/php-7.2.31/ext/opcache/ZendAccelerator.c:2874
#8  0x00007f296975b878 in zm_shutdown_zend_accelerator (type=1, module_number=43)
    at /home/www/softwares/php-7.2.31/ext/opcache/zend_accelerator_module.c:435
#9  0x0000000000a38858 in module_destructor (module=0x2180080) at /home/www/softwares/php-7.2.31/Zend/zend_API.c:2564
#10 0x0000000000a2c976 in module_destructor_zval (zv=0x7ffdbe856890) at /home/www/softwares/php-7.2.31/Zend/zend.c:690
#11 0x0000000000a42b7e in _zend_hash_del_el_ex (ht=0x1682100 <module_registry>, idx=42, p=0x1fd0520, prev=0x0)
    at /home/www/softwares/php-7.2.31/Zend/zend_hash.c:998
#12 0x0000000000a42c49 in _zend_hash_del_el (ht=0x1682100 <module_registry>, idx=42, p=0x1fd0520)
    at /home/www/softwares/php-7.2.31/Zend/zend_hash.c:1021
#13 0x0000000000a441a2 in zend_hash_graceful_reverse_destroy (ht=0x1682100 <module_registry>)
    at /home/www/softwares/php-7.2.31/Zend/zend_hash.c:1477
#14 0x0000000000a36647 in zend_destroy_modules () at /home/www/softwares/php-7.2.31/Zend/zend_API.c:2008
#15 0x0000000000a2ce7a in zend_shutdown () at /home/www/softwares/php-7.2.31/Zend/zend.c:905
#16 0x000000000099bac4 in php_module_shutdown () at /home/www/softwares/php-7.2.31/main/main.c:2458
#17 0x0000000000b26f8d in main (argc=1, argv=0x7ffdbe856db8) at /home/www/softwares/php-7.2.31/sapi/fpm/fpm/fpm_main.c:2020

[sofire]

I disable opcache extension，the signal changed from SIGABRT to SIGSEGV.

[18-May-2020 11:56:51] WARNING: [pool www] child 8376 said into stderr: "NOTICE: PHP message: PHP Fatal error:  Uncaught Exception: RdKafka\Kafka::__construct() has not been called in [no active file]:0"
[18-May-2020 11:56:51] WARNING: [pool www] child 8376 said into stderr: "Stack trace:"
[18-May-2020 11:56:51] WARNING: [pool www] child 8376 said into stderr: "#0 {main}"
[18-May-2020 11:56:51] WARNING: [pool www] child 8376 said into stderr: "  thrown in [no active file] on line 0"
[18-May-2020 11:56:51] WARNING: [pool www] child 8376 said into stderr: "[no active file](0) : Fatal error - Uncaught Exception: RdKafka\Kafka::__construct() has not been called in [no active file]:0"
[18-May-2020 11:56:51] WARNING: [pool www] child 8376 said into stderr: "Stack trace:"
[18-May-2020 11:56:51] WARNING: [pool www] child 8376 said into stderr: "#0 {main}"
[18-May-2020 11:56:51] WARNING: [pool www] child 8376 said into stderr: "  thrown"

[18-May-2020 11:56:51] WARNING: [pool www] child 8376 exited on signal 11 (SIGSEGV) after 1759.068766 seconds from start

[nick-zh]

@sofire can you tell me what command from the extension is causing this crash?

[sofire]

Thanks for your reply.

I try to find which function is wrong, but find nothing.
The program runs ok, except the crash when program exit.
The kafka producer is ok, the message is not lost, and no error logs.

[nick-zh]

@sofire do you have any callbacks registered?

[nick-zh]

So i tried to reproduce this with:
OS: alpine:3.11.6
PHP: 7.2.31
ext: 4.0.3
librdkafka: 1.4.0

with the following example code:

$conf = new Conf();
$conf->set('message.timeout.ms', '5000');
$conf->set('metadata.broker.list', 'jobcloud-message-bus:9095');

$message = 'test';

$producer = new Producer($conf);
$topic = $producer->newTopic('test');

try {
    for ($i = 0; $i < 5; $i++) {
        echo 'Writing message '.$i."\n";
        $topic->producev(RD_KAFKA_PARTITION_UA, 0, json_encode($message), 'test');
    }
} catch (\RuntimeException $e) {
    echo 'ERROR: ' , $e->getMessage() , ' (code: ' , $e->getCode() , ')' , PHP_EOL;
    exit;
}

$producer->flush(10000);

and i was not able to reproduce.
In earlier versions this used to happen when the producer / consumer was already deconstructed but there was a callback or internal call still trying to execute something. Since php-rdkafka:4.x this should not happen anymore. I also added callbacks in my example setDrMsgCb and setErrorCb but i was not able to reproduce this.

[sofire]

Yes , I add callbacks: setLogCb / setErrorCb / setDrMsgCb / setRebalanceCb
Tomorrow I try to test these callbacks.

[nick-zh]

@sofire in the producer you shouldn't use setRebalanceCb this callback is only for the high level consumer RdKafka\KafkaConsumer. Let me know if you find anything more that can help reproduce this.

[sofire]

I remove setRebalanceCb, the error disappear.

But when I remove all callbacks: setLogCb / setErrorCb / setDrMsgCb / setRebalanceCb, produce the same error.

[nick-zh]

@sofire thx for the update. It is strange that this still happens when you have no callbacks, but unless you have some example docker setup with code, so i can reproduce this, it's going to be hard to track this down since i was not able to reproduce this.

[Steveb-p]

@sofire are you creating multiple instances of Producer by any chance?

[sofire]

@nick-zh Thanks very much. I will do more testing, if I find some more information, I will tell . Maybe my script is too complicated.

@Steveb-p No, I use php singleton pattern , and the script call the producer only once, and exit.

[nick-zh]

@sofire thanks for the update and sry that i couldn't be of more help. As i said, it seems that after destruct somehow there is still a call that gets executed. I tried to pinpoint it, but sadly i was not able to see where this could happen, in 4.x this should not happen anymore, so this is really intriguing.
Do you use a framework in your project or just plain PHP? If i find time this week, i will try to investigate more.

[sofire]

Use our own framework, and redis , redis cluster extensions.

[sofire]

In cli mode, have the same errors. In cgi / php-fpm mode, no these errors.
I used setLogCb / setErrorCb / setDrMsgCb to write logs.

[22-May-2020 10:20:06 PRC] PHP Fatal error:  Uncaught Exception: RdKafka\Kafka::__construct() has not been called in [no active file]:0
Stack trace:
#0 {main}
  thrown in [no active file] on line 0

[nick-zh]

So running this script, which produces 5 messages and has the same callbacks registered:

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

$conf = new Conf();
$conf->set('message.timeout.ms', '5000');
$conf->set('metadata.broker.list', 'kafka:9092');

$conf->setDrMsgCb(function (Producer $kafka, Message $message) {
    if ($message->err) {
        switch ($message->err) {
            case RD_KAFKA_RESP_ERR__MSG_TIMED_OUT:
                $errorStr = rd_kafka_err2str($message->err);
                break;
            default:
                $errorStr = 'Unknown error';
        }

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

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

$conf->setLogCb(function ($kafka, $level, $facility, $message) {
    echo sprintf('Level: %s, Fac: %s, Msg: %s', $level, $facility, $message);
});

$message = 'test';

$producer = new Producer($conf);
$topic = $producer->newTopic('test');

try {
    for ($i = 0; $i < 5; $i++) {
        echo 'Writing message '.$i."\n";
        $topic->producev(RD_KAFKA_PARTITION_UA, 0, json_encode($message), 'test');
    }
} catch (\RuntimeException $e) {
    echo 'ERROR: ' , $e->getMessage() , ' (code: ' , $e->getCode() , ')' , PHP_EOL;
    exit;
}

$producer->flush(10000);

works totally fine for me. Let me know if you find a way to adjust this example so that it causes the same error you get.
Are you doing any kind of special caching in your framework or in your application?

[sofire]

@nick-zh

#0  0x00007f45c51ac337 in raise () from /lib64/libc.so.6
#1  0x00007f45bbaf3f40 in get_kafka_object (zrk=zrk@entry=0x7f45c327b808) at /home/www/softwares/phpext/php-rdkafka-4.0.3/rdkafka.c:180
#2  0x00007f45bbaf8edd in kafka_topic_free (object=0x7f45c327b818) at /home/www/softwares/phpext/php-rdkafka-4.0.3/topic.c:51

#3  0x000000000081b99a in zend_objects_store_free_object_storage (objects=objects@entry=0x1320ff8 <executor_globals+824>, 
    fast_shutdown=fast_shutdown@entry=1 '\001') at /home/www/softwares/php-7.2.29/Zend/zend_objects_API.c:106
#4  0x00000000007cf868 in shutdown_executor () at /home/www/softwares/php-7.2.29/Zend/zend_execute_API.c:266
#5  0x00000000007e03f5 in zend_deactivate () at /home/www/softwares/php-7.2.29/Zend/zend.c:1032
#6  0x000000000077ead9 in php_request_shutdown (dummy=dummy@entry=0x0) at /home/www/softwares/php-7.2.29/main/main.c:1904
#7  0x000000000088760b in do_cli (argc=4, argv=0x2b3fb80) at /home/www/softwares/php-7.2.29/sapi/cli/php_cli.c:1178
#8  0x000000000043ff4f in main (argc=4, argv=0x2b3fb80) at /home/www/softwares/php-7.2.29/sapi/cli/php_cli.c:1403

[nick-zh]

@sofire great stuff on locating where it happens, thx! I will have a look this evening to see how this can happen. It is during topic deconstruct, but i need to debug how this can actually happen. I will keep you posted :v:

[nick-zh]

Ok i think i know why, pretty sure it is related to the change i did here #333 probably we still need to have this check as well. I am assuming it can happen that rkt is already freed, so it will end in this exception. I will try to find a way to reproduce to make sure, very good find @sofire :v:

[nick-zh]

@sofire i was not able to reproduce, but i think #371 should fix it. If you want you can try to build the extension from this branch to see, if it doesn't happen for you anymore.
Sry it took so long :bowing_man:

[sofire]

I added a log, then the log and Fatal error occurred at the same time.

if (ZE_ISDEF(intern->zrk)&& !intern->rkt) {
    zend_error(E_WARNING,"intern->rkt IS NULL");
}

[01-Jun-2020 19:56:20 PRC] PHP Warning:  intern->rkt IS NULL in Unknown on line 0
[01-Jun-2020 19:56:20 PRC] PHP Fatal error:  Uncaught Exception: RdKafka\Kafka::__construct() has not been called in [no active file]:0
Stack trace:
#0 {main}
  thrown in [no active file] on line 0

Then I modified the topi.c:

if (ZE_ISDEF(intern->zrk)&& intern->rkt) {

The Fatal error disappeared.

I think it's resolved the problem. Thanks.

[nick-zh]

@sofire thx a lot for the feedback and sry for the inconvenience

[Vortelf]

Hello,

Managed to reproduce the issue and it's still present.

OS: Debian 10
PHP: 7.4.11
ext: 4.0.4
librdkafka: 1.4.2

Here's the code we're using with Laravel framework

<?php

use Enqueue\RdKafka\RdKafkaConnectionFactory;
use Illuminate\Support\Facades\Log;
use RdKafka\Conf;

class KafkaProducer
{
    /**
     * timeout in milliseconds that librdkafka flush waits for all outstanding producer requests to be handled.
     * It ensures messages produced properly.
     */
    const FLUSH_TIMEOUT = 10000;

    /**
     * Times to retry flushing before giving up and throwing an exception
     */
    const FLUSH_RETRIES = 10;

    /**
     * Topic missing error message
     */
    const BROKER_MISSING_ERROR_MESSAGE = 'Kafka broker is not set';

    /**
     * Flush error message
     */
    const FLUSH_ERROR_MESSAGE = 'Message not send';

    private $producer;

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

    /**
     * Create Kafka producer
     *
     * @param array $config
     * @param string $topic
     * @return $this
     * @throws RuntimeException
     */
    public function create(array $config = [], $topic = '')
    {
        $conf = new Conf();

        $broker = $config['broker'] ?? $config('kafka.default_broker');

        if (empty($broker)) {
            throw new RuntimeException(self::BROKER_MISSING_ERROR_MESSAGE);
        }

        $conf->set('metadata.broker.list', $broker);

        $this->populateGlobalConfig($conf);
        if(!empty($topic)){
            $this->populateTopicConfig($conf, $topic);
        }

        $this->producer = new \RdKafka\Producer($conf);

        return $this;
    }

    /**
     * Send message to Kafka topic
     *
     * @param string $topic
     * @param string $message
     * @param null $key
     */
    public function send(string $topic, string $message, $key = null): void
    {
        $topic = $this->producer->newTopic($topic);

        $topic->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);
        $this->producer->poll(0);
    }

    /**
     * Insert topic and global config
     * @param Conf $conf
     * @return Conf
     */
    protected function populateGlobalConfig(Conf $conf): Conf
    {
        $globalConfig = config('kafka.globalConfig', []);
        foreach ($globalConfig as $key => $value) {
            $conf->set($key, $value);
        }
        return $conf;
    }

    public function populateTopicConfig(Conf $conf, string $topic)
    {
        $topicsConfig = config('kafka.topics');
        if (isset($topicsConfig[$topic]['config'])) {
            foreach ($topicsConfig[$topic]['config'] as $key => $value) {
                $conf->set($key, $value);
            }
        }
        return $conf;
    }

    public function flush(): void
    {
        for ($flushRetries = 0; $flushRetries < self::FLUSH_RETRIES; $flushRetries++) {
            $result = $this->producer->flush(self::FLUSH_TIMEOUT);
            if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
                break;
            }
        }

        if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
            throw new RuntimeException(self::FLUSH_ERROR_MESSAGE);
        }
    }
}

class KafkaService
{
    private $context;
    public function __construct(){}
    public function setContext($kafkaProducer): void { $this->context = $kafkaProducer; }
    public function getContext() {return $this->context;}
}

app()->singleton('KafkaService', function () {
    $instance = new KafkaService();
    $kafkaInstance = app(KafkaProducer::class)->create(['broker' => config('kafka.broker')],'default');
    $instance->setContext($kafkaInstance);
    return $instance;
});

$instance = app('KafkaService');

After destructor for KafkaProducer gets called when EOF is reached and throws

PHP Fatal error:  Uncaught Exception: RdKafka\Kafka::__construct() has not been called

[sofire]

try to update librdkafka to 1.4.4