Skip to content

Fix #5415: add new KafkaConsumerRecodItemStreamReader - #5454

Open
djechelon wants to merge 1 commit into
spring-projects:mainfrom
djechelon:GH-5415
Open

Fix #5415: add new KafkaConsumerRecodItemStreamReader#5454
djechelon wants to merge 1 commit into
spring-projects:mainfrom
djechelon:GH-5415

Conversation

@djechelon

Copy link
Copy Markdown

As discussed in #5415, goal is to implement a new Kafka-based ItemReader that returns not only the message body (generic V), but the whole envelope (as ConsumerRecord<K,V>).

In order to maintain backwards compatibility, the following has been done:

  • KafkaItemReader was mostly moved to an abstract class, with 3 generics. Of these, <K,V> are the usual Kafka key-value, and T is the actual type returned by the reader
  • KafkaItemReader changed to extend the abstract class and set T to be V
  • KafkaConsumerRecordItemReader to extend the abstract class and set T to be ConsumerRecord<K,V>
  • Changed the builder to return the consumer record instance with a different build() method
  • Worked on the unit tests
  • Signed off commit and updated copyright notice

Please consider that locally I am experiencing mvn verify even on main as I started working on the fork. I was able to verify that my affected tests work, failing tests occur on unaffected classes.

I wish to thank whoever will spend their precious time paying attention to this contribution.

Split KafkaItemReader into base implementation and alternate implementation that returns whole message envelope
Copy same test from KafkaItemReaderTests into KafkaConsumerRecordItemReaderTests
Add few test code to KafkaItemReaderBuilderTests

Signed-off-by: /usr/local/ΕΨΗΕΛΩΝ <djechelon@zighinetto.org>
@quaff

quaff commented Jul 9, 2026

Copy link
Copy Markdown
Contributor

JmsItemReader should be refactored like this if the PR is accepted.

@djechelon

Copy link
Copy Markdown
Author

About that, consider also that AmqpItemReader shows the same mechanism as Jms.

Refactoring Jms and Amqp would mean that the itemType mechanism becomes broken (or turns into a backwards-compatibility-porkaround), as currently it is used to choose between returning the Message<T> wrapper or the plain T value

@quaff

quaff commented Jul 10, 2026

Copy link
Copy Markdown
Contributor

About that, consider also that AmqpItemReader shows the same mechanism as Jms.

Refactoring Jms and Amqp would mean that the itemType mechanism becomes broken (or turns into a backwards-compatibility-porkaround), as currently it is used to choose between returning the Message<T> wrapper or the plain T value

We could deprecate current reader and add new readers.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants