New instrumentation for reactive kafka clients

[maciej-gromul]

Because current implementation created and closed spans immediately it was hard to follow the context. New implementation relies on reactor instrumentation to provide better experience where context contains tracing context and is being propagated downstream.

[pivotal-cla]

@maciej-gromul Please sign the Contributor License Agreement!

Click here to manually synchronize the status of this Pull Request.

See the FAQ for frequently asked questions.

[marcingrzejszczak]

@maciej-gromul can you please sign the CLA?

[maciej-gromul]

Hello, i'm trying to settle some legal things on my side. I'm part of https://github.com/MaerskTech but i can't see it in CLA for choice, do you know what might i be missing ?

[marcingrzejszczak]

Eeee no idea - can you check this out https://cla.pivotal.io/about ?

[maciej-gromul]

Setting myself as public contributor within our org helped. Now i'm waiting for confirmation whether i can sign CLA :) Can you in the meantime take a look at spring-projects-experimental/spring-cloud-sleuth-otel#159 ? It's connected to this one as it brings brave and otel support to the same level in this case :)

[pivotal-cla]

@maciej-gromul Thank you for signing the Contributor License Agreement!

[maciej-gromul]

@marcingrzejszczak successfully signed ;)

[marcingrzejszczak]

I've added a couple of small remarks.

Apart from that I want to understand a few things before we proceed. So do I understand correctly that the current behavior immediately creates and closes the span? And your change now for each element in a flux will instrument the Reactor context with properly extracted span from the consumerRecord?

Also I wonder about any breaking changes. I mean this seems like fixing a bug but I would like to ensure that we're not breaking anybody if we don't need to. I don't see any breaking changes at a first glance - do you?

[marcingrzejszczak]

BTW have you seen this work here ? reactor/reactor-kafka#325

[maciej-gromul]

I've added a couple of small remarks.

Apart from that I want to understand a few things before we proceed. So do I understand correctly that the current behavior immediately creates and closes the span? And your change now for each element in a flux will instrument the Reactor context with properly extracted span from the consumerRecord?

Also I wonder about any breaking changes. I mean this seems like fixing a bug but I would like to ensure that we're not breaking anybody if we don't need to. I don't see any breaking changes at a first glance - do you?

The only thing i can think of is the TracingKafkaConsumerFactory being removed and it's autoconfiguration for reactive consumers. We can leave TracingKafkaConsumerFactory as it is and it's bean autoconfiguration but then if someone has his own ConsumerFactory implementation registered as a bean it will get wrapped again leading to duplicated spans.

BTW have you seen this work here ? reactor/reactor-kafka#325

Haven't seen that one, seems to be for spring boot 3 with observations ? I've already sent and issue to reactor-core since there's propagation issue there reactor/reactor-core#3395

[marcingrzejszczak]

The only thing i can think of is the TracingKafkaConsumerFactory being removed and it's autoconfiguration for reactive consumers. We can leave TracingKafkaConsumerFactory as it is and it's bean autoconfiguration but then if someone has his own ConsumerFactory implementation registered as a bean it will get wrapped again leading to duplicated spans.

But nobody should directly use the TracingKafkaConsumerFactory. We can leave the class, mark it as deprecated, point to the new solution. The whole instrumentation should in fact be fully transparent so from the point of view of the user it should work OOB.

Haven't seen that one, seems to be for spring boot 3 with observations ? I've already sent and issue to reactor-core since there's propagation issue there reactor/reactor-core#3395

Correct. As for the other issue you're referring to I was debugging this partially with Darek today ;)

[marcingrzejszczak]

Congrats @maciej-gromul on your PR!

[maciej-gromul]

But nobody should directly use the TracingKafkaConsumerFactory. We can leave the class, mark it as deprecated, point to the new solution. The whole instrumentation should in fact be fully transparent so from the point of view of the user it should work OOB.

I can leave the class TracingKafkaConsumerFactory but to make it transparent i guess we would need to bring back in the TracingReactorKafkaAutoConfiguration the bean of TracingKafkaConsumerFactory. That way it would be transparent but im not 100% sure whether it wouldn't collide with new solution. Umm you've closed the task without that deprecation ? Will you bring that factory back in separate issue ?

[marcingrzejszczak]

Umm you've closed the task without that deprecation ? Will you bring that factory back in separate issue ?

Oops, I've forgotten about that. Care to open another PR that will add the deprecation? I wouldn't add the bean back cause that bean is really internal to our instrumentation and can cause trouble.

[maciej-gromul]

I've added separate PR here: #2272