Add message converter support for ShareKafkaListenerContainerFactory

[husseinvr97]

Expected Behavior

ShareKafkaListenerContainerFactory should support message converters the same way AbstractKafkaListenerContainerFactory does — allowing users to configure a RecordMessageConverter for record listeners, wired through to the MessagingMessageListenerAdapter so that message conversion and result handling work correctly for Share consumer listeners.

Current Behavior

ShareKafkaListenerContainerFactory.createListenerContainer() always passes null as the message converter when calling endpoint.setupListenerContainer(instance, null). This means:

The MessagingMessageListenerAdapter is created with no converter — raw message conversion does not work for Share consumer listeners Return values from listener methods are silently discarded for Share consumers in MessagingMessageListenerAdapter (line 414) with the comment // For share consumers, we don't handle results yet

Features like @ SendTo and reply handling are completely unavailable for Share consumers

This is also explicitly documented as a known limitation in kafka-queues.adoc under Current Limitations.

Context

The regular Kafka listener stack (AbstractKafkaListenerContainerFactory) has full message converter support — it selects between batchMessageConverter and recordMessageConverter based on the listener type and passes it through the entire chain down to the adapter. The Share consumer stack has none of this infrastructure yet.
ShareKafkaListenerContainerFactory does not extend AbstractKafkaListenerContainerFactory — it is a standalone implementation — so none of the converter fields or wiring is inherited. The gap needs to be filled explicitly.

Proposed Approach

The fix follows the exact same pattern already established in AbstractKafkaListenerContainerFactory:

Add recordMessageConverter field to ShareKafkaListenerContainerFactory with a setter method
Replace the hardcoded null in createListenerContainer() with the converter:

// Before
endpoint.setupListenerContainer(instance, null);


// After
endpoint.setupListenerContainer(instance, this.recordMessageConverter);

Handle result forwarding in MessagingMessageListenerAdapter for Share consumers once the converter is in place

The scope is intentionally kept minimal and focused — matching the existing patterns in the codebase without introducing new abstractions.

Happy to submit a PR once this approach is confirmed or any other approach that sounds suitable for solving the problem