Absence of metrics and traces in spring-cloud-stream-binder-kafka-reactive

[patpatpat123]

Describe the issue
Absence of metrics and traces in spring-cloud-stream-binder-kafka-reactive

To Reproduce
Steps to reproduce the behavior:
Create a simple spring cloud stream reactive kafka producer and consumer
Start producing and consuming messages.
Even with the actuator and micrometer configured, no Kafka related metrics observed.
reactor/reactor-kafka#262
Ditto for traces

version
4.0.2-SNAPSHOT

Expected behavior
Similar to spring-cloud-stream-binder-kafka which supports both metrics and traces, kafka-reactive counterpart does not support any of the two.

Starting an issue here tracking this.

Thank you for the amazing work Spring Cloud Stream team for the reactive support.
Having metrics and traces for spring-cloud-stream-binder-kafka-reactive would allow it to be in production for companies which requires metrics and logs as part of production launch of a business flow based on this framework.

[kschlesselmann]

We are missing the offset lag metric too. Any workarounds available to get at least those basic metrics?

[sobychacko]

@kschlesselmann We will take a look at this and see how we can support metrics in the reactive binder.

[garyrussell]

reactor-kafka now supports adding Micrometer listeners to instrument the producers and consumers (since 1.3.17).

https://projectreactor.io/docs/kafka/release/reference/#_micrometer_metrics

reactor/reactor-kafka#323

[patpatpat123]

Indeed, big thanks @garyrussell for the metrics support.
Amy ETA for the tracing support please?

[kschlesselmann]

@garyrussell I enabled reactor kafka metrics like

@Configuration(proxyBeanMethods = false)
class ReactorKafkaConfiguration {

    @Bean
    fun receiverOptionsCustomizer(meterRegistry: MeterRegistry): ReceiverOptionsCustomizer<Any, Any> =
        ReceiverOptionsCustomizer { _, receiverOptions ->
            receiverOptions.consumerListener(MicrometerConsumerListener(meterRegistry))
        }
}

and while it seems to work fine there are a lot of metrics/tags now. Any way to reduce those metrics? I tried something like

@Bean
fun kafkaMetricsMeterFilter(): MeterFilter = object : MeterFilter {

    private val tagsToRemove = setOf(
        "client.id",
        "reactor-kafka.id",
        "kafka.version",
        "node.id",
    )

    override fun map(id: Meter.Id): Meter.Id = if (id.name.startsWith("kafka.")) {
        id.tags
            .filter { it.key !in tagsToRemove }
            .let { id.replaceTags(it) }
    } else {
        id
    }
}

but this doesn't seem to work. Any suggestions?

[garyrussell]

I would expect a filter to work; if you can't figure it out, I suggest you reach out to the Micrometer folks.

We don't support Micrometer internals here.

[garyrussell]

Probably the first place to start is debug the MeterRegistryPostProcessor bean in Boot's MetricsAutoConfiguration - it looks like it should apply all filter bean instances. If that all looks good, next step is to step into the micrometer code.

[patpatpat123]

Hello team,

Just checking on this issue.
As mentioned in the title, there are two elements missing in spring cloud stream binder kafka reactive.
The two items are:

1 - metrics
2 - traces

For item number 1, metrics, I believe this issue is now resolved with the help of this PR:
reactor/reactor-kafka#323
I believe it has been properly integrated into spring cloud stream.

If not anything else, thank you for the help provided.
If it is confirmed metrics are now available, I am more than happy to rename the title (or someone else) in order to properly reflect the accuracy of the issue.

With that said, I believe that as of today, November 2023, traces are yet to be available in spring cloud stream.
Reator kafka just released GH-321 PR-325
Please correct me if I am wrong, but I do not believe that as of today, traces are integrated into spring cloud stream.
May I ask if I am correct?
And if yes, may I also ask for help to make this happen?

Thank you for your time.