Producer is not working in PHP 8.1

[edipoReboucas]

Estamos migrando alguns de nossos serviços para o PHP 8.1 e nesse processo a produção de mensagens parou de funcionar.

Após algum tempo de análise identificamos que o problema foi causado por uma breaking change do rdkafka 4, onde ele parou de enviar as mensagens durante a "destruição" de seus objetos.

https://github.com/arnaud-lb/php-rdkafka/releases/tag/4.0.0

Mesmo antes da migração o código aparentemente funcionava por um erro. É esperado que as mensagens sejam enviadas conforme um poll de mensagens, mas não é o que acontece.

Ao fim da execução do Metamorphosis::produce todos os objetos do rdkafka e muitos do metamorphosis são destruídos, pois não tem nada que segure uma referencia dos objetos.

O rdkafka 3 faz um flush durante a destruição dos seus objetos, dando a impressão que esta tudo funcionando. Já o rdkafka 4+ não faz, evidenciando o problema.

Segue um código simples que funciona no rdkafa 3 e não funciona no rdkafka 4 que ilustra o mesmo comportamento do Metamorphosis::produce:

function produce () {
    $conf = new RdKafka\Conf();
    $conf->set('metadata.broker.list', 'kafka:9092');

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

    $topic = $producer->newTopic("test");
    $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message payload 1 " . random_int(1000, 9999));
    $producer->poll(0);
}

// Como nada faz referencia aos objetos do rdkafka os mesmos são destruidos.
// no rdkafka 3 um flush é executado durante a destruição dos seus objetos
// no rdfaka 4 esse flush não acontece
produce();
sleep(10);

Segue um outro exemplo de um código que funciona, removendo esse comportamento da destruição dos objetos:

$conf = new RdKafka\Conf();
$conf->set('metadata.broker.list', 'kafka:9092');

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

$topic = $producer->newTopic("test");
$topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message payload 1 " . random_int(1000, 9999));
$producer->poll(0);

// as mensagens são enviadas durante o sleep
sleep(10);

Como uma solução paliativa, configuramos o Metamorphosis para funcionar de forma sincronia e com required_acknowledgment ativado.

Em alguns de nossos projetos o problema não ocorre pois estamos usando as classes internas do Metamorphosis ao invés do métodoMetamorphosis::produce, segurando as instancias necessárias em memória durante toda a execução do script.