Kafka symfony transport

[dheineman]

Adds the TransportFactory and Driver required to use kafka in the Symfony bundle

[dheineman]

not sure if this is going to be needed.

[makasim]

It will, keep it.

[dheineman]

Symfony does not allow dots in configuration nodes but the librdkafka configuration keys do contain them, need to figure out how to maybe normalize them?

[makasim]

you can define a variable node (which allows everything to be put in it).

[dergriewatz]

Why just do you not use underscores?

[dheineman]

As far as i can tell even a variable node requires a name and will throw an exception when it contains any dots.

I can just remove the metadata.broker.list scalarNode and just leave the global ArrayNode. Any suggestions?

[dergriewatz]

By default you can configure the broker over the config in this way:

enqueue:
    transport:
        default: 'kafka://localhost:9092'

So I think it is a good solution to provide a alternative config way like this:

enqueue:
    transport:
        kafka:
            metadata_broker_list: localhost:9092
            ...

The easier way is to use ->scalarNode('metadata_broker_list')->end()

[makasim]

I'd suggest moving all the broker options to a dedicated configuration option. The option should be a variable node (no validation) like this one for example

Developers are free to put there whatever the broker supports.

[dergriewatz]

Another solution I found is to add a parser in the RdKafkaConnectionFactory class like the parseDsn method and translate the config properties where the dot is needed.

    public function __construct($config = 'kafka:')
    {
        if (empty($config) || 'kafka:' === $config) {
            $config = [];
        } elseif (is_string($config)) {
            $config = $this->parseDsn($config);
        } elseif (is_array($config)) {
            $config = $this->parseDots($config); // or some other naming ;)
        } else {
            throw new \LogicException('The config must be either an array of options, a DSN string or null');
        }

        $this->config = array_replace($this->defaultConfig(), $config);
    }

[makasim]

Another solution I found is to add a parser in the RdKafkaConnectionFactory class like the parseDsn method and translate the config properties where the dot is needed.

I am not in favor of it. If you use factory in plain php you can pass doted options. Symfony does not allow it so it has to be fixed there.

[dergriewatz]

But is not the configuration to validate the better option?
I think it is a valid choice for a library like this that supports many different libraries to use a self defined config format and map it to the specific library formats.

[dheineman]

I went with the variable node for now as parsing or normalizing the configuration nodes would require maintaining a map of all rdkafka configuration properties (as it consists of keys with both dots and underscores). And i am not particularly in favour of this.

[dheineman]

The entire configuration needs more work/checking

[makasim]

It looks good to me, would be happy to merge it.

[dergriewatz]

It does not have to be checked.
The default of the field $contentType is null and the default return value of $message->getHeader('content_type') is also null if the header content_type not set.

[dergriewatz]

Why just do you not use underscores?

[dergriewatz]

Does not the transport factory have to be set in the EnqueueBundle class?

if (class_exists(RdKafkaConnectionFactory::class)) {
    $extension->setTransportFactory(new RdKafkaTransportFactory('kafka'));
} else {
    $extension->setTransportFactory(new MissingTransportFactory('kafka', ['enqueue/rdkafka']));
}

[dheineman]

@makasim This should be done

[makasim]

@dheineman Thank you very much

[makasim]

Tagged 0.8.27 with this feature included.