OpenTelemetry traces between Kafka producer & consumer

[edeandrea]

I'm trying to figure out if this is expected or not. I have Kafka producer & consumer apps which use the OpenTelemetry otlp exporter extension.

The apps are reactive apps. The producer app listens to REST requests (HTTP POST) using RESTEasy reactive. It then uses the MutinyEmitter to send the message to Kafka:

@WithSpan("FightService.persistFight")
Uni<Fight> persistFight(@SpanAttribute("arg.fight") Fight fight) {
    Log.debugf("Persisting a fight: %s", fight);

    return Fight.persist(fight)
      .replaceWith(fight)
      .map(this.fightMapper::toSchema)
      .chain(this.emitter::send)
      .replaceWith(fight);
}

In the consumer app, it does this:

@Incoming("fights")
@Outgoing("team-stats")
@WithSpan("SuperStats.computeTeamStats")
public Multi<TeamScore> computeTeamStats(Multi<Fight> results) {
	return results.map(this.stats::add)
		.invoke(score -> LOGGER.debugf("Fight received. Computed the team statistics: %s", score));
}

The outgoing channel goes to an in-memory channel which is then served over a websocket.

When I issue the HTTP post to the producer app and look at the trace in Jaeger, I don't see the relationship with the consumer app:

But when I look at the consumer app, I do see a "receive" operation where I do see the relationship:

Going back to the producer in the Jaeger UI, I now see a "send" operation where the relationship does show:

2 parts to my question(s):

Part 1

Is this a bug? Is it working as expected? Am I missing something in my implementation to get this to work?

Part 2

Additionally, are the spans really working the way they should? In

Shouldn't the persistFight span be nested under performFight, since performFight calls persistFight?

@WithSpan("FightService.performFight")
public Uni<Fight> performFight(@SpanAttribute("arg.fighters") @NotNull @Valid Fighters fighters) {
    Log.debugf("Performing a fight with fighters: %s", fighters);
    return determineWinner(fighters)
		.chain(this::persistFight);
}

@WithSpan("FightService.persistFight")
Uni<Fight> persistFight(@SpanAttribute("arg.fight") Fight fight) {
    Log.debugf("Persisting a fight: %s", fight);

    return Fight.persist(fight)
      .replaceWith(fight)
      .map(this.fightMapper::toSchema)
      .map(f -> Message.of(f, Metadata.of(TracingMetadata.withPrevious(Context.current()))))
      .chain(message -> {
        this.emitter.send(message);
        return Uni.createFrom().voidItem();
      })
      .replaceWith(fight);
}

Or are they sequential due to the nature of reactive programming? Its essentially just assembling a pipeline of operations. The order/nesting of methods really doesn't affect the execution of the pipeline?

[quarkus-bot[bot]]

/cc @Ladicek, @alesj, @cescoffier, @ozangunalp, @radcortez

[alesj]

I'm actually surprised you even get "persistFight" in the span chain,
as I would expect it not to get intercepted by the inner call,
but otoh, I'm not familiar exactly on how "WithSpan" mechanism works.

I remember I did have some issues due to OTel's ThreadLocal usage on passing the current Context fwd,
but this might not be it here.

[radcortez]

OTel is not fully integrated with RM yet. While everything is mostly in place, we are still missing the piece to automatically propagate the context to and from RM. Right now this has to be done manually with:

// outgoing
Message.of(OBJECT, Metadata.of(TracingMetadata.withCurrent(Context.current())))
// incoming
message.getMetadata(TracingMetadata.class).ifPresent(metadata -> QuarkusContextStorage.INSTANCE.attach(metadata.getCurrentContext()));

[edeandrea]

I think I got it. My method has both an @Incoming and an @Outgoing, which will publish to one of the channels. As part of the pipeline it will use a MutinyEmitter to publish the other.

Once I clean this up a bit I'll do a PR against quarkusio/quarkus-super-heroes#2. If you wouldn't mind reviewing it there once its done @radcortez / @ozangunalp / @cescoffier I'd really appreciate it!

[ozangunalp]

This makes more sense. Initially I thought that you didn't have any outgoing.

I think you could have used an intermediary in-memory channel to later broadcast to multiple outbound channels.

As in:

@Incoming("fights")
@Outgoing("mem-broadcast")
@Broadcast
TeamScore process(Fight fight)

@Incoming("mem-broadcast")
@Outgoing("destination-1")
TeamScore forward(TeamScore score)

@Incoming("mem-broadcast")
@Outgoing("destination-2")
TeamScore forward2(TeamScore score)

[edeandrea]

Yes initially I didn't have any @Outgoing.

I think you could have used an intermediary in-memory channel to later broadcast to multiple outbound channels.

If I did that though then I wouldn't be able to ack the message after both sets of processing had concluded?

Also, the result of each is different, since they are performing different logic.

It'll probably make more sense for you to see what I did. I should have a PR for the super heroes done later today that has everything, but this is the re-write of the listener class:
https://github.com/edeandrea/quarkus-super-heroes/blob/opentelemetry/event-statistics/src/main/java/io/quarkus/sample/superheroes/statistics/listener/SuperStats.java

[edeandrea]

Also, manually adding in the tracing stuff (retrieving the parent context, adding new spans, etc) does over complicate this because I need to use the Message.

[edeandrea]

FYI @radcortez / @ozangunalp / @cescoffier I created a PR at quarkusio/quarkus-super-heroes#107. Would you mind reviewing/providing your thoughts/inputs there? Thank you very much!

[radcortez]

For the REST Client invocations what happens is that a single context is created to handle the execution, so multiple calls will use the same context. Due to the Span suppression rules from OTel (subsequent spans from the same kind in the same context are ignored), you only get a single span for the REST Client calls.

This is fixed in #25336