GH-11183: Fix batch mode message ID generation in KafkaMessageDrivenChannelAdapter#11184
Conversation
…aMessageDrivenChannelAdapter Fixes: spring-projectsgh-11183 Signed-off-by: zernov-a <zernov.a.i.90@gmail.com>
artembilan
left a comment
artembilan
left a comment
There was a problem hiding this comment.
Please, add your name to the @author list of the affected classes.
Looking good!
Thank you!
| @@ -153,7 +153,13 @@ else if (JacksonPresent.isJackson2Present()) { | |||
| .toArray(new String[0])); | |||
| messageConverter.setHeaderMapper(headerMapper); | |||
| this.recordListener.setMessageConverter(messageConverter); | |||
There was a problem hiding this comment.
I think the this.recordListener.setMessageConverter() should go outside of the if..else, too.
…nChannelAdapter Signed-off-by: zernov-a <zernov.a.i.90@gmail.com>
Thank you for the review! I've added the |
artembilan
left a comment
artembilan
left a comment
There was a problem hiding this comment.
Looks good.
Just couple minors.
Thanks
| * @author Artem Bilan | ||
| * @author Urs Keller | ||
| * @author Jooyoung Pyoung | ||
| * @author Zernov Oleksii |
There was a problem hiding this comment.
I think the international naming convention is to express first name first.
See an example of my name in this list which is Ukrainian, too.
|
|
||
| @Test | ||
| void testBatchListener() { | ||
| this.kafkaTemplate.send(TEST_TOPIC10, "foo"); |
There was a problem hiding this comment.
Please, don't use foo/bar language in the code.
I know we have a lot of them already there, but we prefer to get rid of it eventually.
Thanks
| KafkaMessageDrivenChannelAdapter.ListenerMode.batch, TEST_TOPIC10) | ||
| .configureListenerContainer(c -> | ||
| c.ackMode(ContainerProperties.AckMode.MANUAL) | ||
| .id("topic10ListenerContainer"))) |
There was a problem hiding this comment.
I don't think that we need any of these customization to proof the concept.
I mean the goal of this specific test to make sure splitter doesn't fail.
So, what container id or what mode, doesn't matter.
Therefore redundant from the readability perspective.
Less code in the test - less chance of bugs in tests 😄
Signed-off-by: zernov-a <zernov.a.i.90@gmail.com>
artembilan
left a comment
artembilan
left a comment
There was a problem hiding this comment.
LGTM.
Will be merged automatically when PR build is green.
Thank you for contribution; looking forward for more!
For the future: all the info you have included into PR description, could simply go to the first commit - and it becomes that PR description.
Now I had to add it to the final commit manually.
2 participants
Fixes #11183
When using
Kafka.messageDrivenChannelAdapterwithListenerMode.batchfollowed by a.split(), the splitter fails because the batch message doesn't have a message ID generated. This is required by the splitter to properly set correlation headers on split messages.The root cause was that the constructor was setting the same
MessagingMessageConverter(intended for record mode) on the batch listener instead of creating a properBatchMessagingMessageConverterwith message ID generation enabled.This fix ensures that when in batch mode, a
BatchMessagingMessageConverteris created withsetGenerateMessageId(true)andsetGenerateTimestamp(true), similar to how it's done for record mode.Changes
KafkaMessageDrivenChannelAdapterconstructor to properly configureBatchMessagingMessageConverterfor batch modetestBatchListener()inKafkaDslTeststo verify the fix