Add a Lightweight MessageListener Adapter Supporting Message Conversion

[garyrussell]

See https://stackoverflow.com/questions/70684560/concurrentkafkalistenercontainerfactory-message-converter-is-ignored-when-config/70685111#70685111

Similar to RabbitMQ, we could provide a thin adapter that performs message conversion.

[breader124]

Hi, I'll try to start working on this issue

[breader124]

Hi once again, it seems that after update introduced in 4e2f9fd project doesn't compile anymore due to failures in resolving micrometer related imports. What's more, one test fails even after I resolved problem with imports. I guess you'll be fixing it all soon (as the commit I mentioned is the last one on main branch and it was pushed last Friday), so I'll just wait for it and start working whenever it's ready, from my perspective it doesn't make sense to double the work

[garyrussell]

@breader124 My sincere apologies - I was off on Friday and we have a holiday tomorrow, so I was trying to get a jump on Tuesday's releases (they were delayed from tomorrow because of the holiday). I should have checked the CI builds (I actually did, but on my 'phone and failed to notice the failure). Fixing now...

[garyrussell]

Fixed. @breader124 once again sorry about that.

[breader124]

@garyrussell, it's absolutely not a problem, thanks so much for quick reaction. Now I'm able to jump in

[garyrussell]

@breader124 More apologies - I have had to revert the Micrometer upgrade for compatibility with other projects at this time.

Please rebase your branch before pushing a PR.

Thanks.

[breader124]

Hi @garyrussell, no problem, I'll do it. I'd also like to ask you for a little guidance and validation of my initial thoughts. So basically as I understand this task, I need to provide concrete implementation of MessageListener interface, right? In order to do that, I'd propose to create another interface extending MessageListener and called ConvertingMessageListener. In this new interface, I'd provide default implementation for onMessage(ConsumerRecord<K, V> data) simply throwing UnsupportedOperationException and add unimplemented method void onMessage(T data, Consumer consumer). This way I'd be able to provide then concrete implementation of ConvertingMessageListener named let's say SimpleConvertingMessageListenerImpl. This class during creation could accept desired type and Consumer for already converted message. Example usage could look similarly to

kafkaMessageListenerContainer.setupMessageListener(new ConvertingMessageListenerAdapter<>(Type desiredType, Consumer consumer));

What's your opinion about your solution? Please bear with me if I'm proposing something totally incorrect, it's the first time with this codebase from my perspective :)

I'd also like to ask question about the difference between ConsumerRecord and Message. In order to convert ConsumerRecord to Message I'd need to use MessagingMessageConverter, but I don't know if that's the way to go and most probably I need more information about the real purpose of those classes.

Thanks in advance! I'm looking forward to your answer.

[garyrussell]

Sorry; I am tied up on another task at the moment; I will try to get back to this soon, but we definitely don't want another flavor of the MessageListener interface.

You might want to look at the adapters in spring-rabbit; the legacy adapter is close to what this lightweight adapter might be.

https://github.com/spring-projects/spring-amqp/blob/main/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java

It might be easier to look at it on an old branch before it was refactored.

https://github.com/spring-projects/spring-amqp/blob/1.3.x/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java

[breader124]

Okay, thanks for an advice. I think I understand what there's need to be done better right now after looking at the classes you linked. Unfortunately I didn't have enough time previous week to code it, I'll do my best to do it as soon as possible.

[breader124]

Hi @garyrussell, I just opened PR (#2165). Looking forward to your review!