feat: add ContentType attribute for default message content type

[dgafka]

Why is this change proposed?

Resolves #658. When a consumed message carries no contentType header — common when messages are produced by non-Ecotone systems (e.g. Kafka records published by other teams' services) — Ecotone assumes application/x-php and skips deserialization, failing with Lack of Media Type Converter for application/x-php:string to <TargetClass>. There was no declarative way to state "messages arriving at this endpoint are JSON unless stated otherwise"; the only workaround was #[AddHeader('contentType', ...)], which unconditionally overrides and is easy to forget.

Description of Changes

• New #[ContentType] endpoint attribute (extends AddHeader, same family as #[Delayed], #[TimeToLive]) that sets the contentType message header before the endpoint processes the message.
• By default it only fills the header when missing; replaceIfExists: true makes it override the incoming one.
• Media type is validated at configuration time — a malformed value fails at bootstrap, not at first message.
• Handled in EndpointHeadersInterceptor; works transport-agnostically for any endpoint (Kafka consumers, asynchronous command/event handlers, etc.).

Usage

final class ScheduleMeetingKafkaConsumer
{
    #[ContentType('application/json')]
    #[KafkaConsumer('kafka_consumer_schedule_meeting', 'testTopic')]
    public function handle(ScheduleMeeting $scheduleMeeting): void
    {
        // message without contentType header is now deserialized as JSON
    }
}

#[ContentType('application/json', replaceIfExists: true)]
#[Asynchronous('async')]
#[CommandHandler('order.place')]
public function place(PlaceOrder $order): void
{
    // incoming contentType header is always overridden with application/json
}

Use cases

1. Interoperability with foreign producers — a team standardizes on JSON for Kafka topics, but upstream services don't set a contentType header. Declaring #[ContentType('application/json')] on the consumer makes deserialization work without touching producers.
2. Mixed sources on one endpoint — messages from Ecotone services carry proper headers (kept untouched), while legacy publishers send bare payloads (default applies).
3. Correcting mislabeled payloads — an upstream system sets a wrong content type; replaceIfExists: true enforces the known-correct format.

Flow

flowchart LR
    A[Message arrives at endpoint] --> B{contentType header present?}
    B -- no --> C[Set header from ContentType attribute]
    B -- yes --> D{replaceIfExists?}
    D -- true --> C
    D -- false --> E[Keep incoming header]
    C --> F[PayloadConverter deserializes to handler parameter type]
    E --> F

Loading

Pull Request Contribution Terms

• I have read and agree to the contribution terms outlined in CONTRIBUTING.