allow context propagation for kafka tombstones (DataDog/dd-trace-java#1754)
MODIFY
instrumentation/kafka-clients-0.11/src/main/java/io/opentelemetry/instrumentation/auto/kafkaclients/KafkaProducerInstrumentation.java
instrumentation/kafka-clients-0.11/src/main/java/io/opentelemetry/instrumentation/auto/kafkaclients/KafkaProducerInstrumentation.java
@@ -95,8 +95,7 @@ public final class KafkaProducerInstrumentation extends Instrumenter.Default {
callback = new ProducerCallback(callback, parent, span);
- boolean isTombstone = record.value() == null && !record.headers().iterator().hasNext();
- if (isTombstone) {
+ if (record.value() == null) {
span.setAttribute("tombstone", true);
}
@@ -108,9 +107,7 @@ public final class KafkaProducerInstrumentation extends Instrumenter.Default {
// headers attempt to read messages that were produced by clients > 0.11 and the magic
// value of the broker(s) is >= 2
if (apiVersions.maxUsableProduceMagic() >= RecordBatch.MAGIC_VALUE_V2
- && Config.get().isKafkaClientPropagationEnabled()
- // Must not interfere with tombstones
- && !isTombstone) {
+ && Config.get().isKafkaClientPropagationEnabled()) {
Context context = withSpan(span, Context.current());
try {
OpenTelemetry.getPropagators()

MODIFY
instrumentation/kafka-clients-0.11/src/main/java/io/opentelemetry/instrumentation/auto/kafkaclients/TracingIterator.java
instrumentation/kafka-clients-0.11/src/main/java/io/opentelemetry/instrumentation/auto/kafkaclients/TracingIterator.java
@@ -91,9 +91,7 @@ public class TracingIterator implements Iterator<ConsumerRecord> {
long startTimeMillis = System.currentTimeMillis();
spanBuilder.setStartTimestamp(TimeUnit.MILLISECONDS.toNanos(startTimeMillis));
Span span = spanBuilder.startSpan();
- // tombstone checking logic here because it can only be inferred
- // from the record itself
- if (next.value() == null && !next.headers().iterator().hasNext()) {
+ if (next.value() == null) {
span.setAttribute("tombstone", true);
}
decorator.afterStart(span);

MODIFY
instrumentation/kafka-clients-0.11/src/test/groovy/KafkaClientTest.groovy
instrumentation/kafka-clients-0.11/src/test/groovy/KafkaClientTest.groovy
@@ -260,8 +260,8 @@ class KafkaClientTest extends AgentTestRunner {
received.value() == null
received.key() == null
- assertTraces(2) {
- trace(0, 1) {
+ assertTraces(1) {
+ trace(0, 2) {
// PRODUCER span 0
span(0) {
operationName SHARED_TOPIC
@@ -272,16 +272,12 @@ class KafkaClientTest extends AgentTestRunner {
"tombstone" true
}
}
- }
- // when a user consumes a tombstone a new trace is started
- // because context can't be propagated safely
- trace(1, 1) {
// CONSUMER span 0
- span(0) {
+ span(1) {
operationName SHARED_TOPIC
spanKind CONSUMER
errored false
- parent()
+ childOf span(0)
attributes {
"partition" { it >= 0 }
"offset" 0
@@ -292,9 +288,6 @@ class KafkaClientTest extends AgentTestRunner {
}
}
- def headers = received.headers()
- !headers.iterator().hasNext()
-
cleanup:
producerFactory.stop()
container?.stop()

allow context propagation for kafka tombstones (DataDog/dd-trace-java#1754)