Allow configuration of default media type for kafka consumer when no content type header available

[tandev]

Description

When no Content-Type is set, PayloadConverter falls back to MediaType::createApplicationXPHP() for $sourceMediaType and tries to find a MediaTypeConverter towards the target format in an KafkaConsumer.

For interoperability with other systems it should be possible to allow to configured Kafka-Consumer in a way that allows to define Source-Media-Type to for example JSON if no Content-Type Header is set in the kafka message headers and this is the defined standard for kafka usage in a team

Example
Having a Consumer like

final class ScheduleMeetingKafkaConsumer
{
    #[KafkaConsumer(
        'kafka_consumer_schedule_meeting',
        'testTopic'
    )]
    public function handle(ScheduleMeeting $scheduleMeeting): void
    {
        $this->receivedMeetings[] = $scheduleMeeting;
    }
}

with Message on Kafka without headers being processed leads to

Ecotone\Messaging\Support\InvalidArgumentException: Can not call Test\Ecotone\Kafka\Fixture\Calendar\ScheduleMeetingKafkaConsumer::handle. Lack of Media Type Converter for application/x-php:string to application/x-php:Test\Ecotone\Kafka\Fixture\Calendar\ScheduleMeeting

(Note i have also no MediaTypeConverter for JSON, but it didnt look for one, so having one, and a way to define json as default, would probably make that work)

Reproduced in tests via 31c17c8

[dgafka]

What if we would add #[AddHeader('contentType', 'application/json')]. Can you test it out?

[tandev]

Adding the AddHeader makes this work indeed! Think a solution per configuration probably still would be nice to avoid having this be forgotten or unclear

[dgafka]

@tandev so you would like to have on KafkaConsumer attribute something like sourceMediaType which would have always precedence over anything else?

That could be achieved in simpler way and generic way. We could introduce #[SourceMediaType], which under the hood would do AddHeader. Take a look on Delayed attribute as an example.
That would be descriptive and would work for other cases too

[tandev]

My thought was that this could be part of KafkaConsumerConfiguration to have a globally defined input media type, similarly to what we added to the KafkaPublisherConfiguration. But it could be that this works different on inbound handling. But there also sometimes is good case to make it explicit