Add async propagation flags for kafka consumer.
MODIFY
dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java
dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java
@@ -59,6 +59,7 @@ public class TracingIterator implements Iterator<ConsumerRecord> {
decorator.afterStart(span);
decorator.onConsume(span, next);
currentScope = activateSpan(span, true);
+ currentScope.setAsyncPropagation(true);
}
} catch (final Exception e) {
log.debug("Error during decoration", e);

MODIFY
dd-java-agent/instrumentation/kafka-streams-0.11/src/main/java/datadog/trace/instrumentation/kafka_streams/KafkaStreamsProcessorInstrumentation.java
dd-java-agent/instrumentation/kafka-streams-0.11/src/main/java/datadog/trace/instrumentation/kafka_streams/KafkaStreamsProcessorInstrumentation.java
@@ -81,7 +81,7 @@ public class KafkaStreamsProcessorInstrumentation {
CONSUMER_DECORATE.afterStart(span);
CONSUMER_DECORATE.onConsume(span, record);
- activateSpan(span, true);
+ activateSpan(span, true).setAsyncPropagation(true);
}
}
}
@@ -119,6 +119,7 @@ public class KafkaStreamsProcessorInstrumentation {
@Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
public static void stopSpan(@Advice.Thrown final Throwable throwable) {
+ // This is dangerous... we assume the span/scope is the one we expect, but it may not be.
final AgentSpan span = activeSpan();
if (span != null) {
CONSUMER_DECORATE.onError(span, throwable);

Add async propagation flags for kafka consumer.