fix: preserve env placeholders in serialized message attributes

[dgafka]

Why is this change proposed?

Since 1.313.0, consuming messages in a Symfony application fails with DefinitionHelper::unserializeSerializedObject(): Return value must be of type object, false returned whenever an attribute contains an environment placeholder — e.g. #[KafkaConsumer(topics: 'orders.%env(SUFFIX)%')] or #[Asynchronous(asynchronousExecution: [new ErrorChannel('%env(ERROR_CHANNEL)%')])].

Ecotone stored such attributes as PHP-serialize()d strings in the container. Symfony then rewrites the %...% placeholder inside that string during container compilation, which changes the string's byte length but not the serialized length prefix (s:NN:"…"), so unserialize() returns false and the consumer/handler can never be built. It only reproduces under a real Symfony container (Symfony resolves %...% in service arguments). Fixes #669.

Description of Changes

• DefinitionHelper now builds attribute container definitions from DefinedObject::getDefinition() (a structured class + arguments) instead of serialize(), so each placeholder-bearing string is a normal container argument that Symfony resolves correctly — no length-prefix corruption, and the environment value is still applied.
• #[Asynchronous], #[ErrorChannel], #[DelayedRetry], #[WithoutMessageCollector], #[WithoutDatabaseTransaction] now implement DefinedObject (#[KafkaConsumer] already did). All AsynchronousEndpointAttribute elements must be DefinedObject because #[Asynchronous] now exposes asynchronousExecution structurally.
• Scoped to the attribute paths only (buildAttributeDefinitionFromInstance + resolvePotentialComplexAttribute); the generic buildDefinitionFromInstance is left as-is to avoid affecting service configurations.
• Added booted-kernel Symfony regression tests for both cases; the Kafka test publishes to and consumes from an env-placeholder topic end-to-end.
• Added ecotone/kafka + ext-rdkafka to the Symfony package require-dev.

Usage examples

// Kafka consumer with an environment-driven topic name
final class OrderConsumer
{
    #[KafkaConsumer('orders', topics: 'orders.%env(ENV_SUFFIX)%')]
    public function handle(string $payload): void
    {
        // ENV_SUFFIX=production  -> subscribes to "orders.production"
    }
}

// Async command handler with an environment-driven error channel
final class OrderHandler
{
    #[Asynchronous('orders', asynchronousExecution: [new ErrorChannel('errors.%env(ERROR_CHANNEL)%')])]
    #[CommandHandler('order.place')]
    public function place(PlaceOrder $command): void
    {
        // ERROR_CHANNEL=high_priority -> failures route to "errors.high_priority"
    }
}

Use cases

• Per-environment Kafka topic naming (orders.staging vs orders.production) driven by env vars.
• Configuring an error / dead-letter channel name from the environment.
• Any attribute argument that should be sourced from %env(...)% in a Symfony app.

Flow

sequenceDiagram
    participant App as Symfony container (compile)
    participant Runtime
    Note over App: Before — serialize() blob: s:28:"orders.%env(SUFFIX)%"
    App->>Runtime: %env% resolved inside blob -> s:28:"orders.production" (length mismatch) -> unserialize() = false
    Note over App: After — structured Definition arg: "orders.%env(SUFFIX)%"
    App->>Runtime: %env% resolved as a whole argument -> new KafkaConsumer('orders.production')

Loading

Pull Request Contribution Terms

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