配置以在KafkaMessageListenerContainer不在侦听器上启动新跨度（Kafka Sleuth）

Question

我是Kafka侦探的新手。目前我遇到了一个问题，span在MessageListenerMethodInterceptor中创建了一个在KafkaListener上触发的拦截器，但如果出现异常，我们将在ErrorHandler中丢失spanId和traceId。是否可以将sleuth配置为在KafkaMessageListenerContainer中启动span？

Answer 1

我已经处理好了。在我的本地pc中，它工作正常，但我有一些疑问。。。

>

我通过在application.properties中设置spring.sleuth.messaging.kafka.enabled=false来禁用MessageListenermetodInterceptor

我创建了一个记录拦截器，实现了org。springframework。Kafka。侦听器。RecordInterceptor如下所示：

 private final Tracer tracer;
 private final KafkaTracing kafkaTracing;

 private static final ThreadLocal<Span> CURRENT_SPAN = new ThreadLocal<>();
 private static final ThreadLocal<Tracer.SpanInScope> CURRENT_SPAN_IN_SCOPE = new ThreadLocal<>();

 @Override
 public ConsumerRecord<K, V> intercept(ConsumerRecord<K, V> record, Consumer<K, V> consumer) {
     openSpan(record);
     return RecordInterceptor.super.intercept(record, consumer);
 }

 @Override
 public void failure(ConsumerRecord<K, V> record, Exception exception, Consumer<K, V> consumer) {
     Span span = CURRENT_SPAN.get();
     if (span != null) {
         String message = Optional.ofNullable(exception.getMessage()).orElseGet(() -> exception.getClass().getSimpleName());
         span.tag("error", message);
     }
 }

 @Override
 public void clearThreadState(Consumer<?, ?> consumer) {
     closeSpan();
 }

 private void openSpan(ConsumerRecord<K, V> record) {
     Span span = kafkaTracing.nextSpan(record).name("kafka-tracing").start();
     Tracer.SpanInScope spanInScope = tracer.withSpanInScope(span);
     CURRENT_SPAN.set(span);
     CURRENT_SPAN_IN_SCOPE.set(spanInScope);
     if (LOG.isDebugEnabled()) {
         LOG.debug("created span {}", span);
     }
 }

 private void closeSpan() {
     Span span = CURRENT_SPAN.get();
     if (span != null) {
         try {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("finishing span {}", span);
             }
             span.finish();
         } catch (Exception ex) {
             LOG.warn("on finishing Span {} got error {}", span, ex.getMessage());
         } finally {
             CURRENT_SPAN.set(null);
         }
     }

     Tracer.SpanInScope spanInScope = CURRENT_SPAN_IN_SCOPE.get();
     if (spanInScope != null) {
         try {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("closing SpanInScope {}", spanInScope);
             }
             spanInScope.close();
         } catch (Exception ex) {
             LOG.warn("on closing SpanInScope {} got error {}", spanInScope, ex.getMessage());
         } finally {
             CURRENT_SPAN_IN_SCOPE.set(null);
         }
     }
 }

其他方法为空。

此拦截器的bean创建如下：

    @Bean
    public KafkaTracingRecordInterceptor<Object, Object> kafkaRecordInterceptor(Tracer tracer, Tracing tracing) {
        return new KafkaTracingRecordInterceptor<>(tracer, KafkaTracing.create(tracing));
    }

附言：如果我遗漏了什么或以错误的方式实现了它，请添加评论。

Answer 2

记录拦截器是仪器的最佳位置；最近的改进扩展了它的范围，包括错误处理程序。

https://github.com/spring-projects/spring-kafka/pull/1946