GH-4230: Enhance listener observation handling with additional adapter support #4238

Flamme1004K · 2026-01-10T11:28:55Z

I'm submitting a PR for issue #4230. Although the KafkaBackoffAwareMessageListenerAdapter class delegates to RecordMessagingMessageListenerAdapter, it was observed that metrics were being recorded incorrectly because it wasn't adding metrics despite RecordMessagingMessageListenerAdapter.

To resolve this, I modified the code to set isListenerAdapterObservationAware to true even when it's a KafkaBackoffAwareMessageListenerAdapter. Additionally, when adding observations, I ensured that if the KafkaBackoffAwareMessageListenerAdapter class delegates to RecordMessagingMessageListenerAdapter, it also adds the metrics.

Thank you for your attention.

Signed-off-by: Flamme1004K <[email protected]>

victorpasqualino · 2026-01-10T16:44:07Z

I did something very similar, with a few differences: #4240

Flamme1004K · 2026-01-11T06:18:27Z

@victorpasqualino good Idea!

I got inspired by your code and tried refactoring it. Can you take a look?

211c631

Signed-off-by: Flamme1004K <[email protected]>

victorpasqualino · 2026-01-11T17:29:21Z

@victorpasqualino good Idea!

I got inspired by your code and tried refactoring it. Can you take a look?
* [211c631](https://github.com/spring-projects/spring-kafka/pull/4238/commits/211c6313eee43311ec5cf21e3f1287b915f1fc48)

Looks good to me. Thank you

artembilan

Please, sure you run /gradlew check locally before pushing.

Thanks

artembilan · 2026-01-12T16:32:25Z

...kafka/src/test/java/org/springframework/kafka/support/micrometer/MicrometerMetricsTests.java

 * @author Soby Chacko
- * @since 3.2.7
+ * @author Hyoungjune Kim
+ * @since 4.0.2


This is not correct.
The class was created really in that version, so you cannot modify it.
Plus, the version you target is still not correct since I fill like we need to back-port the fix to 3.3.x as well.
Therefore, if we do @since, we should aim for the lowest version we are going to back-port to.
Luckily, we don't need to use @since anywhere in your change.

Thank you! I rollbacked @since

214d1c1

artembilan · 2026-01-12T16:36:45Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

 		}

+		private MessageListener<?, ?> unwrapDelegateIfAny(MessageListener<?, ?> listener) {
+			if (listener instanceof KafkaBackoffAwareMessageListenerAdapter<?, ?> backoffAware) {


The KafkaBackoffAwareMessageListenerAdapter is an AbstractDelegatingMessageListenerAdapter, so I think any of those should be covered here.
If we go this method, it has to be static.

Good! I added static to this method

71fadb5

Good. No need to add a comment with a confirmation.
I'll see an agreement in the next commit you push.

How about an AbstractDelegatingMessageListenerAdapter support?
You didn't comment anything to contradict my suggestion and didn't implement it.
With the 👍 it give a false impression that all the request have been addressed.
Any comments, please?

And another our code style suggestion: the static methods should go in the end of class.
Not a big deal: just an info if you'd like to know more about our developing process.

That was such a good suggestion I ended up hitting the emoji button without thinking. Next time I'll make sure to use them appropriately.

I've been thinking about it, and instead of using AbstractDelegatingMessageListenerAdapter, how about checking if it's an instance of DelegatingMessageListener and extracting it if it is? What do you think?

b75ae23

…iner Signed-off-by: Flamme1004K <[email protected]>

Signed-off-by: Flamme1004K <[email protected]>

…icsTests` Signed-off-by: Flamme1004K <[email protected]>

Signed-off-by: Flamme1004K <[email protected]>

Flamme1004K · 2026-01-13T13:36:03Z

@artembilan Thank you, Code Review! I run /gradlew check. result I modifed source code.

artembilan

Thank you for the update!

I won't mind any objections for my review.

artembilan · 2026-01-13T13:34:13Z

...kafka/src/test/java/org/springframework/kafka/support/micrometer/MicrometerMetricsTests.java

 	}

+	static class RetryBackOffObservationListener {
+		final CountDownLatch latch = new CountDownLatch(1);


Every member in the class (including before the last } of the last inner class) has to be surrounded with blank lines.
That makes the code much easier to read.
And we do here an Open Source, so our code should be as clean as possible and as a good example for community.

I'm learning about open source. Thank you.

artembilan · 2026-01-13T13:36:13Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

+
+		private void setupObservationRegistry(ObservationRegistry observationRegistry) {
+			if (this.listener == null) {
+				return;


I think for better void digesting it would be better to make it opposite: use if (this.listener != null) { instead around the rest of the code in this method.

artembilan · 2026-01-13T13:40:39Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

 		}

+		private MessageListener<?, ?> unwrapDelegateIfAny(MessageListener<?, ?> listener) {
+			if (listener instanceof KafkaBackoffAwareMessageListenerAdapter<?, ?> backoffAware) {


Good. No need to add a comment with a confirmation.
I'll see an agreement in the next commit you push.

How about an AbstractDelegatingMessageListenerAdapter support?
You didn't comment anything to contradict my suggestion and didn't implement it.
With the 👍 it give a false impression that all the request have been addressed.
Any comments, please?

And another our code style suggestion: the static methods should go in the end of class.
Not a big deal: just an info if you'd like to know more about our developing process.

artembilan · 2026-01-13T13:43:14Z

...kafka/src/test/java/org/springframework/kafka/support/micrometer/MicrometerMetricsTests.java

+				// NOTHING
+			}
+			finally {
+				latch.countDown();


I think we don't need any logic here since in the end you still fall back to the await().untilAsserted().
Therefore, such a latch is just redundant barrier which really does not contribute to the async flow.

During local testing, intermittent failures in the test code were observed, so the source code was modified to call latch.countDown().

The reason is that the count for spring.kafka.listener.id can only be obtained after messages are consumed by the listener. If verification occurs before consumption, a null pointer exception occurs regardless of untilAsserted.

Signed-off-by: Flamme1004K <[email protected]>

…delegates Signed-off-by: Flamme1004K <[email protected]>

Flamme1004K · 2026-01-13T14:37:28Z

@artembilan Thank you for your prompt feedback on the code review.

I've addressed the feedback you provided. Please review it.

artembilan · 2026-01-13T14:51:02Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

+		private void setupObservationRegistry(ObservationRegistry observationRegistry) {
+			if (this.listener != null) {
+				MessageListener<?, ?> target = unwrapDelegateIfAny(this.listener);
+				if (target instanceof RecordMessagingMessageListenerAdapter<?, ?> adapter) {


I think we need to think about this change twice.
You see, our existing logic is RecordMessagingMessageListenerAdapter.class.equals(this.listener.getClass()).
I believe even if we extract the delegate, we still have to be sure that we talk about exactly the same object.
So, if we do this instanceof, then any RecordMessagingMessageListenerAdapter extension is a valid candidate for isListenerAdapterObservationAware.
But that is really not what we want here.
We want to check exactly only for the RecordMessagingMessageListenerAdapter and nothing more.
In some other places we do have some custom RecordMessagingMessageListenerAdapter , which must not be treated as an observation aware.
Makes sense?

@artembilan I agree. It's correct to change the code to verify it's specifically RecordMessagingMessageListenerAdapter.class.

I will modify the code to RecordMessagingMessageListenerAdapter.class.equals(this.listener.getClass()).

I changed the code!

878dc39

…st synchronization Signed-off-by: Flamme1004K <[email protected]>

…for listener adapter handling Signed-off-by: Flamme1004K <[email protected]>

artembilan

Sounds good!
Will merge when build is green.

Thank you!

artembilan · 2026-01-13T16:35:15Z

@Flamme1004K ,

thank you for contribution; looking forward for more!

Flamme1004K · 2026-01-15T13:03:39Z

@artembilan
Thank you! I will strive to contribute more in the future.

Enhance listener observation handling with additional adapter support

b73a755

Signed-off-by: Flamme1004K <[email protected]>

Flamme1004K force-pushed the GH-4230 branch from 2f92cab to b73a755 Compare January 10, 2026 11:31

Flamme1004K changed the title ~~Enhance listener observation handling with additional adapter support~~ GH-4230: Enhance listener observation handling with additional adapter support Jan 10, 2026

Add retry backoff metrics test and listener with observation support

2b78ff3

Signed-off-by: Flamme1004K <[email protected]>

Refactor listener observation registry setup for better adapter handling

211c631

Signed-off-by: Flamme1004K <[email protected]>

Flamme1004K force-pushed the GH-4230 branch from cec3cc9 to 211c631 Compare January 11, 2026 06:25

artembilan requested changes Jan 12, 2026

View reviewed changes

Flamme1004K added 2 commits January 13, 2026 22:07

Make unwrapDelegateIfAny method static in KafkaMessageListenerConta…

71fadb5

…iner Signed-off-by: Flamme1004K <[email protected]>

Rollback JavaDoc version in MicrometerMetricsTests class

214d1c1

Signed-off-by: Flamme1004K <[email protected]>

Flamme1004K force-pushed the GH-4230 branch from 1ebd8a9 to 214d1c1 Compare January 13, 2026 13:25

Flamme1004K added 2 commits January 13, 2026 22:33

Replace assertEquals with AssertJ's assertThat in `MicrometerMetr…

f39cf62

…icsTests` Signed-off-by: Flamme1004K <[email protected]>

Add blank line for code readability in KafkaMessageListenerContainer

d8d4a66

Signed-off-by: Flamme1004K <[email protected]>

artembilan requested changes Jan 13, 2026

View reviewed changes

Flamme1004K added 3 commits January 13, 2026 23:16

Remove unused latch

7ef3f54

Signed-off-by: Flamme1004K <[email protected]>

Simplify null check logic in setupObservationRegistry method

2eefc6b

Signed-off-by: Flamme1004K <[email protected]>

Refactor unwrapDelegateIfAny to handle DelegatingMessageListener …

b75ae23

…delegates Signed-off-by: Flamme1004K <[email protected]>

artembilan requested changes Jan 13, 2026

View reviewed changes

Flamme1004K added 2 commits January 13, 2026 23:57

Rollback CountDownLatch to RetryBackOffObservationListener for te…

33b892f

…st synchronization Signed-off-by: Flamme1004K <[email protected]>

Refactor setupObservationRegistry to use explicit class comparison …

878dc39

…for listener adapter handling Signed-off-by: Flamme1004K <[email protected]>

artembilan approved these changes Jan 13, 2026

View reviewed changes

artembilan merged commit 90d8522 into spring-projects:main Jan 13, 2026
3 checks passed

GH-4230: Enhance listener observation handling with additional adapter support #4238

GH-4230: Enhance listener observation handling with additional adapter support #4238

Conversation

Flamme1004K commented Jan 10, 2026

Uh oh!

victorpasqualino commented Jan 10, 2026

Uh oh!

Flamme1004K commented Jan 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

victorpasqualino commented Jan 11, 2026

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Flamme1004K commented Jan 13, 2026

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Flamme1004K Jan 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Flamme1004K commented Jan 13, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Flamme1004K Jan 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

artembilan commented Jan 13, 2026

Uh oh!

Flamme1004K commented Jan 15, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Flamme1004K commented Jan 11, 2026 •

edited

Loading

Flamme1004K Jan 13, 2026 •

edited

Loading

Flamme1004K Jan 13, 2026 •

edited

Loading