I am using Symfony and Enqueue Kafka to consume a Kafka topic. The problem is that Enqueue doesn't know how to serialize the messages on that topic and it throws a 'Undefined array key "body"' error.

In the doc I find that I can create my own serializer: https://php-enqueue.github.io/transport/kafka/#serialize-message

How can I do it in a Symfony project?

解决方案

I found out the solution. I decorated the transport's connection factory to set my custom serializer.

In services.yaml I added:

    App\Messenger\MyConnectionFactory:
        decorates: 'enqueue.transport.default.connection_factory'
        arguments:
            $decorated: '@App\Messenger\MyConnectionFactory.inner'

And my connection factory looks like this:

<?php

namespace App\Messenger;

use Interop\Queue\ConnectionFactory;
use Interop\Queue\Context;

class MyConnectionFactory implements ConnectionFactory
{

    public function __construct(
        private ConnectionFactory $decorated
    ) {
    }

    public function createContext(): Context
    {
        $context = $this->decorated->createContext();

        if (method_exists($context, 'setSerializer')) {
            $context->setSerializer(new MySerializer());
        }

        return $context;
    }
}

This might not be the best way to do it. Are there any suggestions?