No elegant way to get span in Kafka ErrorHandler

[AntonKuzValid]

I need a way to get span in Kafka ErrorHanlder. I do the next thing in error handler -

    @Override
      public void handle(Exception ex, ConsumerRecord<?, ?> consumerRecord) {
        var curSpan = this.kafkaTracing.nextSpan(consumerRecord).name("on-message").start()
       try (Tracer.SpanInScope ws = tracer.withSpanInScope(curSpan)) {
      log.error(ex);
        } finally {
      span.finish();
    }
  }

But method in KafkaTracing nextSpan() - clear all headers before error handler do its job. And KafkaTracing if final and i can't inherit from it. I have a question - is it a bug that you don't give an opportunity not to remove headers from ConsumerRecord?

[AntonKuzValid]

Probably you can add a ThreadLocal, this will be setted in proxy. example -

@Aspect
@Component
public class KafkaTracingAspect {
  /** Текущий Span потока */
  public static final ThreadLocal<Span> currentSpan = new ThreadLocal<>();

  @Autowired private Tracer tracer;

  @Around("@annotation(org.springframework.kafka.annotation.KafkaListener)")
  public Object kafkaListener(ProceedingJoinPoint joinPoint) throws Throwable {
    currentSpan.set(tracer.currentSpan());
    return joinPoint.proceed();
  }
}

[marcingrzejszczak]

Can you describe this in more details? What you're trying to achieve, how the flow of code looks like (e.g. first X method is called and then Y and Y doesn't have the tracing context due to KafkaTracing nextSpan(). Also if you could present how in your opinion things should look like to improve the current situation

[AntonKuzValid]

I'll try to explain. I use spring kafka integration. I add org.springframework.kafka.listener.ErrorHandler for logging all unexpected exceptions when my consumer works. The problem is that when the errorHandler do its job, there is no span in context already. Then i decided to get my span form ConsumerRecord because errorHandler method has the next signature - @Override public void handle(Exception ex, ConsumerRecord<?, ?> consumerRecord). But the method which takes the span from ConsumerRecord it clears its headers. The method is Span nextSpan(ConsumerRecord<?, ?> record) in class brave.kafka.clients.KafkaTracing. The solution is is to remove the code which drop headers from kafka message or make class brave.kafka.clients.KafkaTracing non final to allow devloper to iherit from class and change logic. Now the working solution is to save Span in ThreadLocal and get it in errorHandler

[jfumard]

Same problem here !
It seems that the ConsumerRecord<?,?> received in the @Override public void handle(Exception ex, ConsumerRecord<?, ?> consumerRecord) method has been stripped from its headers, including the traceId.

[marcingrzejszczak]

Calling @shakuzen and @adriancole ! In Sleuth we're just reusing what's there in Brave.

[codefromthecrypt]

what seems to be discussed cc @jeqo, is that there's a spring defined hook org.springframework.kafka.listener.ErrorHandler which is not defined in JMS or Kafka itself. These sorts of abstractions will need a solution at that level as opposed to guessing and reaching underneath. The way I typically look at this is to grab a stack trace, or set a breakpoint where the thing is being invoked, then look at the instrumentation involved.

In some cases, what might need to happen is that the platform specific code needs to instrument directly if it differs significantly in needs or impl from the underlying drivers. For example, in brave, the spring-rabbit code is different than JMS for similar reasons.

[codefromthecrypt]

as I spent a little time looking at the code (KafkaMessageListenerContainer) I figured I'd make some comments of interest

• the ErrorHandler can actually be invoked with no messages/records at all (ex handleConsumerException does this.

KafkaMessageListenerContainer.doInvokeRecordListener is what both invokes the sleuth-wrapped listener (MessageListenerMethodInterceptor) and what folks seem to identify as a gap.. the spring-kafka error handler (executed after that listener is no longer on the stack)

The only way I can see out of this, since there are multiple lifecycles involved anyway, is to use the thread-local approach. MessageListenerMethodInterceptor adds a thread local of the last error span context before propagating the error. A wrapped spring-kafka ErrorHandler looks for that thread local and uses it accordingly. The ErrorHandlers cleanup task clears the thread local.

this might not be the end solution, but I suspect something like this is gonna be needed because the container lifecycle and coordinating that is the key thing here.. since the container lifecycle could be one or batch spans etc. it is easier to couple directly to the existing slueth tracing aspects imho

[marcingrzejszczak]

So until there are other options let's assume that the thread local approach is the correct one as mentioned previously by @AntonKuzValid . If anything changes for that matter we will update this issue.