fix: resolve RESEND failure strategy crash for dynamic channels wrapping Kafka sub-channels

[dgafka]

Why is this change proposed?

When a dynamic channel (e.g. "async") wraps Kafka sub-channels (e.g. "async_tenant_a", "async_tenant_b"), the RESEND failure strategy crashes with "Publisher configuration for async not found". This happens because KafkaAcknowledgementCallback::resend() uses the polling metadata's endpoint ID to look up the producer, but only the actual Kafka channel names have registered publisher configs. This makes dynamic channels with Kafka unusable when RESEND failure strategy is configured.

Usage Example

// Configure Kafka sub-channels with RESEND strategy
KafkaMessageChannelBuilder::create('async_tenant_a', topicName: 'topic_a')
    ->withFinalFailureStrategy(FinalFailureStrategy::RESEND),
KafkaMessageChannelBuilder::create('async_tenant_b', topicName: 'topic_b')
    ->withFinalFailureStrategy(FinalFailureStrategy::RESEND),

// Wrap them in a dynamic channel — RESEND now works correctly
DynamicMessageChannelBuilder::createRoundRobin(
    thisMessageChannelName: 'async',
    channelNames: ['async_tenant_a', 'async_tenant_b'],
),

Use Case Scenarios

1. Multi-tenant Kafka messaging — Each tenant has a dedicated Kafka topic/channel, wrapped by a dynamic channel for unified consumer management. Failed messages are resent to the correct tenant's topic.
2. AMQP Stream with dynamic routing — Multiple AMQP stream queues behind a dynamic channel with round-robin consumption and RESEND on failure.

sequenceDiagram
    participant Consumer
    participant DynamicChannel as DynamicChannel "async"
    participant SubChannel as KafkaChannel "async_tenant_a"
    participant Handler
    Consumer->>DynamicChannel: poll (endpointId="async")
    DynamicChannel->>SubChannel: poll (polledChannelName="async_tenant_a")
    SubChannel-->>DynamicChannel: message
    DynamicChannel-->>Consumer: message
    Consumer->>Handler: handle message
    Handler-->>Consumer: throws Exception
    Note over Consumer: RESEND strategy triggered
    Consumer->>SubChannel: resend using channelName="async_tenant_a"
    Note over SubChannel: Producer found ✓ (was failing with "async")

Loading

Pull Request Contribution Terms

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