Opentracing with Kafka Streams

[muirandy]

Quarkus: 1.0.0.CR1
Java: 1.8.0_232
Maven: 3.6.1

Description
The opentracing already within Quarkus allows reporting to Jaeger. This works for HTTP endpoints, and also allows DB requests to be traced. I'd like my Kafka Streams also to be traced.

Implementation ideas
This confluent blog talks about using Zipkin (in place of Jaeger). It shows how to add tracing to Kafka Streams applications. I have implemented this previously in a non-Quarkus app. It uses openzipkin/brave.

At the time of writing, the way to define a Kafka Streams app in Quarkus is to write a method that returns a Topology. AFAIK, the opentracing for Jaeger doesn't support using a Tracer with a Topology - it needs a KafkaStreams.

Jaeger can be used as a backend for Zipkin, with no changes to the applications being traced. I'm able to trace across KSQL, Kafka Connect and (non-Quarkus) KStreams apps. I hope that a Quarkus implementation would work against either a Zipkin or Jaeger backend, and would work with the traces that would already be in use (ie I'd be able to see the same trace going through all components).

[muirandy]

Ping @gunnarmorling

[geoand]

cc @pavolloffay @objectiser

[objectiser]

@muirandy As an interim solution, I would recommend setting the quarkus.jaeger.propagation=jaeger,b3 property to allow B3 trace context propagation to be supported in your quarkus service, and then use brave to instrument your Kafka Topology.

However it would be a good enhancement to auto-instrument the injected KafkaStream or Topology using opentracing - worth investigating further.

[gunnarmorling]

@muirandy, can you expand a bit on that one:

AFAIK, the opentracing for Jaeger doesn't support using a Tracer with a Topology - it needs a KafkaStreams.

Could the Quarkus KafkaStreams extension automatically apply the required configuration to produced the KafkaStreams object? Perhaps based on some configuration or other produced item(s) you define?

[muirandy]

@gunnarmorling I'll do my best :)

Quarkus Kafka Streams

Following the Quarkus guide on Kafka Streams, we create a Topology which quarkus looks after running for us (we don't have access to the KafkaStreams), ie:

@Produces
public Topology buildTopology() { ... }

Open Tracing

Looking at the Kafka Streams section under open tracing, we need access to a KafkaStreams object:

// Instantiate TracingKafkaClientSupplier
KafkaClientSupplier supplier = new TracingKafkaClientSupplier(tracer);

// Provide supplier to KafkaStreams
KafkaStreams streams = new KafkaStreams(builder.build(), new StreamsConfig(config), supplier);
streams.start();

Comparing with Zipkin instead of Jaeger (for reference)

This is consistent with what I'd need to do if I was using Zipkin in place of Jaeger (the workaround that @objectiser suggested) - I'd also need the KafkaStreams:

Properties kafkaProperties = ...
brave.kafka.streams.KafkaStreamsTracing tracing = ...
KafkaStreams streams = tracing.kafkaStreams(topology, kafkaProperties);
streams.start();

All this means that at present, I can't follow the current Quarkus Guide for Kafka Streams (ie just provide a Topology). Instead, I need something like the following (happily I'd used a previous version of the KafkaStreams extension, and it still works):

void onStart(@Observes StartupEvent event) {
        brave.kafka.streams.KafkaStreamsTracing kafkaStreamsTracing = configureTracing();
        Topology topology = buildTopology();
        Properties kafkaProperties = createStreamingConfig();
        streams = kafkaStreamsTracing.kafkaStreams(topology, kafkaProperties);
        streams.start();
}

Adding spans

The code defining the Topology would likely need to add a span. Again, this can be achieved using brave with Zipkin when building the Topology:

    Topology buildTopology() {
        StreamsBuilder builder = new StreamsBuilder();
        ValueMapper<String, String> jsonToXmlMapper = createValueMapper();
        KStream<String, String> inputStream = builder.stream(converterConfiguration.inputKafkaTopic, Consumed.with(Serdes.String(), Serdes.String()));

        // Important bit here - this adds a span:
        KStream<String, String> xmlStream = inputStream.transformValues(kafkaStreamsTracing.mapValues("json_to_xml", jsonToXmlMapper));

        xmlStream.to(converterConfiguration.outputKafkaTopic);
        return builder.build();
    }

Summary

Could the Quarkus KafkaStreams extension automatically apply the required configuration to produced the KafkaStreams object? Perhaps based on some configuration or other produced item(s) you define?

I think it could. The "Open Tracing" stuff from above looks like generic code to me, and I guess under the covers you're already creating a KafkaStreams object from the Topology that is supplied.

However, we'd still need a mechanism within the creation of the Topology to add a span. I'm not sure whats already in place for JDBC in Quarkus - maybe something similar might help. Otherwise, we'd need something similar to what we get from brave and Zipkin.

HTH. Let me know what you think!

[muirandy]

If its of help, you can find what I'm doing with the Zipkin workaround here.

[gunnarmorling]

@muirandy, so for OpenTracing, would it help if the extension supported a CDI producer for a KafkaClientSupplier? If the extension finds such bean, it could use it when creating the KafkaStreams instance.

However, we'd still need a mechanism within the creation of the Topology to add a span

Isn't the producer method for the Topology the place to do that? Is anything else needed?

you can find what I'm doing

Thanks, I'll take a look into that to be better grasp the needs.

[gunnarmorling]

Btw. I think also with the latest version it should work if you just create your own KafkaStreams bean. You wouldn't benefit from topic being awaited and lifecycle management, but e.g. enablement of native still would be present.

[pavolloffay]

Are we talking here about microprofile reactive messaging for kafka?

If so I dig into it some time ago. As far as I remember there was a problem with context propagation (extract the context from the message if consumer method accepts objects, link spans in a processor - sending span to a different topic than receiving):
Here are some references:

• Add context to Message class smallrye/smallrye-reactive-messaging#214
• Context propagation eclipse/microprofile-reactive-messaging#72
• POC for smallrye Add tracing to kafka smallrye/smallrye-reactive-messaging#205 - it requires the context propagation to be done first.

The context propagation work in microprofile will take some time, in the meantime smallrye could provide API to install kafka interceptors - or better to expose kafka configuration.

[muirandy]

Sorry for the delay folks, I've been flat out on other things that the Zipkin workaround gave me.

@gunnarmorling I pretty much brain dumped what I have on this. I need to learn more about Quarkus and how its working to be able to comment more. If there's something I can do to help try something out then let me know.

@pavolloffay I don't think so - I think this guide is looking at microProfile streaming whereas here we're looking at Kafka Streams. I don't understand all the magic that @gunnarmorling has put in as yet, so I could be wrong!

[iabughosh]

Is there a possibility to implement OpenTracing Instrumentation for Kafka Clients: both Kafka-Streams & Kafka-Reactive-messaging?

[geoand]

@pavolloffay ^

[pavolloffay]

I haven't found a way in Quarkus to automatically pull in OT instrumentation for Kafka client. Even if we do that it will not be a full solution. There will be spans for each receive/send operation but the traces will be likely broken due to fact that reactive messaging does not propagate context in the message - see eclipse/microprofile-reactive-messaging#72. First we need to get this done and then build tracing on top of it.

[pilhuhn]

Is that something which could be done in SmallRye as "tech-preview", where the experiences could then be fed back into the MicroProfile specification processes?

[pavolloffay]

I am not sure, when I tried I need the changes in the spec first. Maybe somebody with a better understanding of Smallrye (smallrye/smallrye-reactive-messaging#214) could give it a try.

[LeonardoBonacci]

Any update on this issue?

[pat-karakun]

In contrast to what's written on https://github.com/opentracing-contrib/java-kafka-client wrt how to setup opentracing with KafkaStreams, it's (now) possible to just provide a KafkaClientSupplier for injection:

@Produces
public KafkaClientSupplier buildTracingKafkaClientSupplier() {
    return new TracingKafkaClientSupplier(GlobalTracer.get());
}

But what's the current state regarding context propagation, namely inject spanId from a consumed message into the tracing context and later add it from there to those messages produced.
Is the following comment from Nov 2019 still valid or is there a better approach in the meantime?

@muirandy As an interim solution, I would recommend setting the quarkus.jaeger.propagation=jaeger,b3 property to allow B3 trace context propagation to be supported in your quarkus service, and then use brave to instrument your Kafka Topology.

However it would be a good enhancement to auto-instrument the injected KafkaStream or Topology using opentracing - worth investigating further.

[EKrol2]

Are there any updates on this?

[brunobat]

OpenTracing is deprecated and this issue will not be implemented.