Potential context leak in kafka-clients instrumentation

[trask]

TracingIterator wraps Iterator<ConsumerRecord> returned from kafka, and starts scopes in next() and closes them in hasNext() for each record in the iterator.

This is a clever way to put the context in scope, and may be acceptable, but worth reviewing alternatives, including removing it and only automatically putting context in scope when kafka is used from higher-level APIs, e.g. spring-kafka's MessageListener.onMessage() and vertx-kafka's KafkaReadStream.poll() which takes a handler (we can start a span for the handler and create links to all of the messages in the batch).

[pavolloffay]

I have find one Kafka issue that might be related to this. I have seen multiple spans with the same ID coming from the Kafka instrumentation.

[trask]

This instrumentation is definitely leaky.

I sent a PR to reduce the impact of this leakiness: #5826

But it's still worth considering dropping this instrumentation, and recommending manual instrumentation of all poll use cases, e.g.

    ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));

    for (ConsumerRecord<K, V> record : records) {
      Context context =
          GlobalOpenTelemetry.getPropagators()
              .getTextMapPropagator()
              .extract(Context.root(), record, KafkaConsumerRecordGetter.INSTANCE);
      try (Scope ignored = context.makeCurrent()) {
        ...
      }
    }

Where KafkaConsumerRecordGetter can be copied locally or imported from io.opentelemetry.instrumentation:opentelemetry-kafka-clients-common.

[mateuszrzeszutek]

But it's still worth considering dropping this instrumentation, and recommending manual instrumentation of all poll use cases, e.g.

I 100% agree with that 💯 -- we could even provide some utilities for correct manual instrumentation (along with examples) as a part of the library instrumentation.

But, before doing that we'd have to implement a dedicated spring-kafka instrumentation (the current one only covers batch processing, kafka-clients covers the 1-record scenario), a vertx-kafka instrumentation (I think it also depends on kafka-clients instrumentation), and perhaps consider doing something with reactor-kafka (I'm ~95% sure it's broken right now). And there's kafka-streams too, it also relies on kafka-clients instrumentation.

[trask]

as @anuraaga pointed out in SIG meeting, even with PR #5826, you can still end up with an unended (and so unreported) consumer span when an exception is thrown out of the iterator loop.

this is not nearly as bad as leaking the context and then getting same trace id stamped forever after, but it's still problematic and probably reason enough to deprecate the ConsumerRecords iterator instrumentation and provide a simple library/guidance for users who are using poll. this simple library/guidance could be used by both non-javaagent and javaagent users.

[62mkv]

Hi @mateuszrzeszutek any news/plans on this one? we're also facing leaking trace id's and missing spans for reactor-kafka based consumers.. 😢 javaagent, 1.31

[almogtavor]

@62mkv are you using Schedulers.parallel()? I've been facing issues only when using it, and resolved them (according to what I have seen until now) by manually reactivating the original Context for each message