From my experience, when the consumer is configured with

enable.auto.offset.store: false
enable.auto.commit: false

the offset is only committed when explicitly doing ack call. This means in case of an error, the message is never acknowledged / the offset is never committed, hence the consumer will start lagging and won't be able to consume new messages. In some/many cases this may be intended, especially when the ordering of messages matter – but it's also possible that one does not want to block the consumer when there's a single failing message. Then if you have retry configured on the same transport, you start producing retry messages which never get consumed because the consumer is stuck on the initial message (given you only have 1 consumer or 1 consumer group).

I'd like to see a configuration variable to make ack call on reject, a bit similar to Redis transports' delete_after_reject, say commit_offset_after_reject or ack_after_reject.

Anyway, would be nice to see this Kafka transport in the core 👍

should be @param. The array shape config is interoperable between tools (at least as long as none of the keys are a non-lowercase variant of a phpdoc keyword as defined by the psalm parser because of vimeo/psalm#10008)

should be @param

should be @var (but is this actually needed ? Can't psalm infer it thanks to the return type of newTopic ?)

why are you ignoring it when passing a KafkaFactory instead of assigning it ?

To me, the code should be like this:

    public function __construct(
        private LoggerInterface $logger = new NullLogger(),
        KafkaFactory $kafkaFactory = null,
    ) {
            $this->kafkaFactory = $kafkaFactory ?? new KafkaFactory(
                new LoggingLogCallback($logger),
                new LoggingErrorCallback($logger),
                new LoggingRebalanceCallback($logger),
            );
    }

is there an actual benefit about lazy-loading those 2 objects ? Their constructor only assigns the properties

And what is the benefit of splitting the logic into those separate objects if they are not available as extension points by allowing to inject them in the constructor, as done in some other bridges ?

This seems to be inline with the other messenger bridge transports. I can update it to be injected if you prefer?

Indeed, only the AmazonSQSTransport properly exposes these dependencies through DI, which makes sense as it's the most recent one. I agree we should do the same here

should those stamps be @internal or no ?

should those be @internal or no ?

why making the second argument nullable and the third argument nullable and optional ? this does not match the documented signature of the rebalance callback where those are always passed.

Making them nullable means that the code inside them has to deal with null values which this may not be necessary.

this is broken. you need to convert the kafka level into a \Psr\Log\LogLevel value. Otherwise, you have no guarantee that things will work fine as the only mandatory supported values for the log method in PSR-3 are those string constants. Any other supported value for the level is implementation-defined and so you cannot rely on it when targeting only the PSR-3 interface without knowing the implementation.

Please follow the contributing rules for changelog entries: https://symfony.com/doc/current/contributing/code/conventions.html#writing-a-changelog-entry

these should also use the proper interfaces respectively instead of the concrete implementations, to allow for decoration.

psalm has stubs for this extension, it should not be required to install it: https://github.com/vimeo/psalm/blob/6.x/stubs/extensions/rdkafka.phpstub

This link is not working.

It would be wise to document the methods rather than linking to a page that may change over time.

This needs to be updated