GH-2068: Add lightweight converting adapter #2165

breader124 · 2022-03-15T00:41:23Z

I've added lightweight message listener adapter converting consumer records to desired type specified by developer as mentioned in #2068

artembilan

I'm not fully sure in the purpose behind this class, but here you are some my review at a glance.

Thanks

artembilan · 2022-03-15T14:44:28Z

...rg/springframework/kafka/listener/adapter/ConvertingAndDelegatingMessageListenerAdapter.java

+	 */
+	public ConvertingAndDelegatingMessageListenerAdapter(Object delegate, Class<V> desiredValueType) {
+		validateMessageListener(delegate);
+		Objects.requireNonNull(desiredValueType);


We prefer to use org.springframework.util.Assert instead of blind NullPointerException

Solved in 603202f

artembilan · 2022-03-15T14:45:57Z

...rg/springframework/kafka/listener/adapter/ConvertingAndDelegatingMessageListenerAdapter.java

+
+	private void validateMessageListener(Object messageListener) {
+		Objects.requireNonNull(messageListener);
+		if (!(messageListener instanceof MessageListener)) {


Why then just don't require a MessageListener delegate in the constructor?

Solved in 603202f

artembilan · 2022-03-15T14:47:32Z

...pringframework/kafka/listener/adapter/ConvertingAndDelegatingMessageListenerAdapterTest.java

+ * @since 3.0.0
+ *
+ */
+class ConvertingAndDelegatingMessageListenerAdapterTest {


We prefer to name test classes as *Tests.

Solved in 603202f

artembilan · 2022-03-15T14:49:43Z

...rg/springframework/kafka/listener/adapter/ConvertingAndDelegatingMessageListenerAdapter.java

+ * @since 3.0
+ * @see AcknowledgingConsumerAwareMessageListener
+ */
+public class ConvertingAndDelegatingMessageListenerAdapter<T, U, V> implements AcknowledgingConsumerAwareMessageListener<T, U> {


I would call it just ConvertingMessageListener. Doesn't look like it adapts to something not compatible in API: https://www.baeldung.com/java-adapter-pattern

Solved in 603202f

artembilan · 2022-03-15T14:56:19Z

...rg/springframework/kafka/listener/adapter/ConvertingAndDelegatingMessageListenerAdapter.java

+		}
+	}
+
+	private ConsumerRecord<T, V> convertConsumerRecord(ConsumerRecord<T, U> data) { // NOSONAR


I don't think we care somehow about T & U in the logic of this class.
In the end what we only need is a V for the data type we would like to convert to.

I tried to replace T and U with just ?, but then it turned out that I can't override onMessage method without compiler yelling at me with communicate "same erasure, yet neither overrides the other". Also from code readability perspective it's kinda useful in my opinion, because it's clearly visible, that key (T) stays the same and U is at the end converted to V. Anyway, if you don't feel convinced I'll try to put some additional effort to investigate how could I avoid using T and U keeping only V as a type of a value after conversion.

artembilan · 2022-03-16T15:41:30Z

...afka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java

+ * @since 3.0
+ * @see AcknowledgingConsumerAwareMessageListener
+ */
+public class ConvertingMessageListener<T, U, V> implements AcknowledgingConsumerAwareMessageListener<T, U> {


I look at this from the end-users perspective where they have to think what exactly need to be specified for the T and U when they really don't care about them since T is always going to be an original one and U is always going to be converted to V independently of U.
Therefore I find them redundant and want to avoid questions about them and trying to explain that we need them merely to satisfy an internal code consistency for compiler.
I believe we still can find some hard casting behavior internally even if it could be handled with a rawtypes suppression...

Changed in eeb9961

artembilan · 2022-03-16T15:44:47Z

...afka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java

+	 * @param messageConverter the {@link MessageConverter} to use for conversion.
+	 * @param desiredValueType the {@link Class} setting desired type of {@link ConsumerRecord}'s value.
+	 */
+	public ConvertingMessageListener(MessageListener<T, V> delegate, MessageConverter messageConverter, Class<V> desiredValueType) {


If messageConverter is optional and we have a default one, it is better to extract it into a setter instead.

Changed in eeb9961

Please, stop making comments like this.
You just push commit to the PR and we are notified.
We also will see your changes and how they are related to the review we have left before.
Otherwise it is spamming my mail box: first your new commit and this "duplication" comments in their individual emails 😄

artembilan · 2022-03-16T15:45:28Z

...afka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java

+	}
+
+	@Override
+	public void onMessage(ConsumerRecord<T, U> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer) { // NOSONAR


I'm not sure that we really need a // NOSONAR over here. The method doesn't look so complicated.

Changed in eeb9961

artembilan · 2022-03-16T15:45:48Z

...afka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java

+		return rebuildConsumerRecord(data, converted);
+	}
+
+	private ConsumerRecord<T, V> rebuildConsumerRecord(ConsumerRecord<T, U> data, V converted) {


This method should be static.

Changed in eeb9961

artembilan · 2022-03-16T15:49:25Z

...afka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java

+		this.delegate = delegate;
+		this.desiredValueType = desiredValueType;
+
+		this.messageConverter = new SimpleMessageConverter();


The SimpleMessageConverter does not make too much sense for a logic of this listener.
It just does this: return (ClassUtils.isAssignableValue(targetClass, payload) ? payload : null);. So, if we expect in this converter whatever is there has been deserialized for us, then we just don't need to use this converter at all.
Perhaps a GenericMessageConverter would be better...

Changed in eeb9961

artembilan

I'm OK with this one.
Let's see what @garyrussell thinks and into what version we merge it.

Thanks

breader124 · 2022-03-27T19:59:58Z

Hi @garyrussell, what's your opinion about this PR?

garyrussell · 2022-03-28T14:37:08Z

I will try to look at it this week.

garyrussell

Thanks for the contribution; I was anticipating a call to a POJO after conversion, but this is a valid implementation too (creating a new ConsumerRecord).

We can add a POJO adapter later.

garyrussell · 2022-03-31T15:57:01Z

...afka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java

+ * @see AcknowledgingConsumerAwareMessageListener
+ */
+@SuppressWarnings("rawtypes")
+public class ConvertingMessageListener<V> implements AcknowledgingConsumerAwareMessageListener<Object, Object> {


Also implement DelegatingMessageListener so the container can determine what type of listener the delegate is.

garyrussell · 2022-03-31T15:59:16Z

...afka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java

+			this.delegate.onMessage(convertedConsumerRecord, acknowledgment);
+		}
+
+		this.delegate.onMessage(convertedConsumerRecord);


Needs to be in an else block, otherwise the listener is called twice.

garyrussell · 2022-03-31T16:04:33Z

...afka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java

+				data.partition(),
+				data.offset(),
+				data.key(),
+				converted


What about the headers and other metadata (timestamp etc.) ?

garyrussell · 2022-03-31T16:06:45Z

...afka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java

+	private ConsumerRecord convertConsumerRecord(ConsumerRecord data) {
+		Header[] headerArray = data.headers().toArray();
+		Map<String, Object> headerMap = Arrays.stream(headerArray)
+				.collect(Collectors.toMap(Header::key, Header::value));


We should use a KafkaHeaderMapper here (configurable, with SimpleKafkaHeaderMapper default).

I analysed the code one more time and figured out that creating GenericMessage with headers is pointless, because they don't play any role in conversion of message payload. Also in newly built ConsumerRecord they are just copied form received record. Based on that I deleted these lines related to extracting headers and take care only about payload in conversion. What's your opinion?

The only thought I had was if someone might want access to the headers to decide the conversion strategy.

Let's say, the listener is expecting ConsumerRecord<String, Foo> where Foo is an interface (or abstract class) with two implementations Bar and Baz; there might be type information in the headers to help the converter decide which one to instantiate.

But, I agree, it would be unfortunate to take the overhead of mapping the headers when the converter doesn't need them.

I would therefore recommend supporting a header mapper property, but only map the headers if one is provided.

…ata and fix doubled invokation of message listener

…cover it with tests

breader124 · 2022-05-29T19:10:22Z

Hi @garyrussell , will this PR be merged?

garyrussell · 2022-05-31T17:37:45Z

Yes, sorry; I have been busy and was not working last week.

spring-projectsGH-2068: Add lightweight converting adapter

9003aa5

breader124 mentioned this pull request Mar 15, 2022

Add a Lightweight MessageListener Adapter Supporting Message Conversion #2068

Closed

artembilan requested changes Mar 15, 2022

View reviewed changes

spring-projectsGH-2068: Introduce amendments after code review

603202f

breader124 force-pushed the GH-2068 branch from d13f07f to 603202f Compare March 15, 2022 23:41

artembilan requested changes Mar 16, 2022

View reviewed changes

spring-projectsGH-2068: Introduce further amendments after code review

eeb9961

artembilan approved these changes Mar 18, 2022

View reviewed changes

garyrussell suggested changes Mar 31, 2022

View reviewed changes

breader124 added 2 commits April 2, 2022 19:58

spring-projectsGH-2068: Extend content of copied record by more metad…

8430134

…ata and fix doubled invokation of message listener

spring-projectsGH-2068: Add possibility to specify header mapper and …

a21b3fb

…cover it with tests

breader124 requested a review from garyrussell April 8, 2022 10:24

garyrussell approved these changes Jun 1, 2022

View reviewed changes

garyrussell merged commit 8034c2c into spring-projects:main Jun 1, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

GH-2068: Add lightweight converting adapter #2165

GH-2068: Add lightweight converting adapter #2165

breader124 commented Mar 15, 2022

artembilan left a comment

artembilan Mar 15, 2022

breader124 Mar 15, 2022

artembilan Mar 15, 2022

breader124 Mar 15, 2022

artembilan Mar 15, 2022

breader124 Mar 15, 2022

artembilan Mar 15, 2022

breader124 Mar 15, 2022

artembilan Mar 15, 2022

breader124 Mar 15, 2022

artembilan Mar 16, 2022

breader124 Mar 18, 2022

artembilan Mar 16, 2022

breader124 Mar 18, 2022

artembilan Mar 18, 2022

artembilan Mar 16, 2022

breader124 Mar 18, 2022

artembilan Mar 16, 2022

breader124 Mar 18, 2022

artembilan Mar 16, 2022

breader124 Mar 18, 2022 •

edited

Loading

artembilan left a comment

breader124 commented Mar 27, 2022

garyrussell commented Mar 28, 2022

garyrussell left a comment

garyrussell Mar 31, 2022

garyrussell Mar 31, 2022

garyrussell Mar 31, 2022

garyrussell Mar 31, 2022

breader124 Apr 2, 2022

garyrussell Apr 5, 2022

breader124 commented May 29, 2022

garyrussell commented May 31, 2022

GH-2068: Add lightweight converting adapter #2165

GH-2068: Add lightweight converting adapter #2165

Conversation

breader124 commented Mar 15, 2022

artembilan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

breader124 Mar 18, 2022 • edited Loading

Choose a reason for hiding this comment

artembilan left a comment

Choose a reason for hiding this comment

breader124 commented Mar 27, 2022

garyrussell commented Mar 28, 2022

garyrussell left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

breader124 commented May 29, 2022

garyrussell commented May 31, 2022

breader124 Mar 18, 2022 •

edited

Loading