diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfo.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfo.java index 495847befe6..640fff4cb02 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfo.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfo.java @@ -1,6 +1,8 @@ package datadog.trace.instrumentation.kafka_clients; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import java.util.Objects; +import java.util.concurrent.atomic.AtomicReference; import javax.annotation.Nullable; import org.apache.kafka.clients.Metadata; @@ -9,6 +11,10 @@ public class KafkaConsumerInfo { private final Metadata clientMetadata; private final String bootstrapServers; + // handle to the consume span this consumer left lingering past its last poll loop; not part of + // consumer identity, so excluded from equals/hashCode + private final AtomicReference deferredConsumeSpan = new AtomicReference<>(); + public KafkaConsumerInfo(String consumerGroup, Metadata clientMetadata, String bootstrapServers) { this.consumerGroup = consumerGroup; this.clientMetadata = clientMetadata; @@ -21,6 +27,14 @@ public KafkaConsumerInfo(String consumerGroup, String bootstrapServers) { this.bootstrapServers = bootstrapServers; } + public void setDeferredConsumeSpan(AgentSpan span) { + deferredConsumeSpan.set(span); + } + + public AgentSpan getAndClearDeferredConsumeSpan() { + return deferredConsumeSpan.getAndSet(null); + } + @Nullable public String getConsumerGroup() { return consumerGroup; diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfoInstrumentation.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfoInstrumentation.java index a2fa481491c..01127727c09 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfoInstrumentation.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfoInstrumentation.java @@ -87,6 +87,7 @@ public String[] helperClassNames() { return new String[] { packageName + ".KafkaDecorator", packageName + ".KafkaConsumerInfo", + packageName + ".KafkaConsumerInstrumentationHelper", "datadog.trace.instrumentation.kafka_common.ClusterIdHolder", "datadog.trace.instrumentation.kafka_common.KafkaConfigHelper", "datadog.trace.instrumentation.kafka_common.PendingConfig", @@ -120,6 +121,15 @@ public void methodAdvice(MethodTransformer transformer) { .and(takesArguments(1)) .and(returns(named("org.apache.kafka.clients.consumer.ConsumerRecords"))), KafkaConsumerInfoInstrumentation.class.getName() + "$RecordsAdvice"); + + // close()/unsubscribe() end the poll loop for good, so a consume span deferred past the last + // poll must be finished here rather than left dangling. + transformer.applyAdvice( + isMethod().and(isPublic()).and(named("close")), + KafkaConsumerInfoInstrumentation.class.getName() + "$CloseScopeAdvice"); + transformer.applyAdvice( + isMethod().and(isPublic()).and(named("unsubscribe")).and(takesArguments(0)), + KafkaConsumerInfoInstrumentation.class.getName() + "$CloseScopeAdvice"); } public static class ConstructorAdviceNot27 { @@ -237,9 +247,14 @@ public static void muzzleCheck(ConsumerRecord record) { public static class RecordsAdvice { @Advice.OnMethodEnter(suppress = Throwable.class) public static AgentScope onEnter(@Advice.This KafkaConsumer consumer) { - // Set cluster ID in ClusterIdHolder for Schema Registry instrumentation KafkaConsumerInfo kafkaConsumerInfo = InstrumentationContext.get(KafkaConsumer.class, KafkaConsumerInfo.class).get(consumer); + + // a poll() means any span deferred past the previous poll loop is done -- close it now (no-op + // unless consumer-scope deferral is enabled) + KafkaConsumerInstrumentationHelper.closeLingeringConsumeScope(kafkaConsumerInfo); + + // Set cluster ID in ClusterIdHolder for Schema Registry instrumentation if (kafkaConsumerInfo != null && Config.get().isDataStreamsEnabled()) { Metadata consumerMetadata = kafkaConsumerInfo.getClientMetadata(); if (consumerMetadata != null) { @@ -290,4 +305,19 @@ public static void captureGroup( span.finish(); } } + + /** + * Finishes a {@code kafka.consume} span left active past the last poll when {@code + * close()}/{@code unsubscribe()} ends the poll loop, in case no further poll() arrives. + */ + public static class CloseScopeAdvice { + @Advice.OnMethodEnter(suppress = Throwable.class) + public static void onEnter(@Advice.This KafkaConsumer consumer) { + if (Config.get().isKafkaCreateConsumerScopeEnabled()) { + KafkaConsumerInfo kafkaConsumerInfo = + InstrumentationContext.get(KafkaConsumer.class, KafkaConsumerInfo.class).get(consumer); + KafkaConsumerInstrumentationHelper.closeLingeringConsumeScope(kafkaConsumerInfo); + } + } + } } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java index 756f59aad4b..21835f61b68 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java @@ -123,7 +123,13 @@ public static void wrap( KafkaConsumerInstrumentationHelper.extractBootstrapServers(kafkaConsumerInfo); iterable = new TracingIterable( - iterable, KAFKA_CONSUME, CONSUMER_DECORATE, group, clusterId, bootstrapServers); + iterable, + KAFKA_CONSUME, + CONSUMER_DECORATE, + group, + clusterId, + bootstrapServers, + kafkaConsumerInfo); } } } @@ -145,7 +151,13 @@ public static void wrap( KafkaConsumerInstrumentationHelper.extractBootstrapServers(kafkaConsumerInfo); iterable = new TracingList( - iterable, KAFKA_CONSUME, CONSUMER_DECORATE, group, clusterId, bootstrapServers); + iterable, + KAFKA_CONSUME, + CONSUMER_DECORATE, + group, + clusterId, + bootstrapServers, + kafkaConsumerInfo); } } } @@ -167,7 +179,13 @@ public static void wrap( KafkaConsumerInstrumentationHelper.extractBootstrapServers(kafkaConsumerInfo); iterator = new TracingIterator( - iterator, KAFKA_CONSUME, CONSUMER_DECORATE, group, clusterId, bootstrapServers); + iterator, + KAFKA_CONSUME, + CONSUMER_DECORATE, + group, + clusterId, + bootstrapServers, + kafkaConsumerInfo); } } } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentationHelper.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentationHelper.java index 685a4d4f03f..f48fe4b4398 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentationHelper.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentationHelper.java @@ -1,10 +1,52 @@ package datadog.trace.instrumentation.kafka_clients; +import datadog.context.Context; +import datadog.trace.api.Config; +import datadog.trace.api.InstrumenterConfig; import datadog.trace.bootstrap.ContextStore; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.AgentTracer; import datadog.trace.instrumentation.kafka_common.MetadataState; import org.apache.kafka.clients.Metadata; public class KafkaConsumerInstrumentationHelper { + + /** Whether the last record's consume span should be kept active past the poll loop. */ + public static boolean shouldDeferConsumerScope() { + return Config.get().isKafkaCreateConsumerScopeEnabled(); + } + + /** + * Finishes the {@code kafka.consume} span deliberately left active past the poll loop when + * consumer-scope deferral is enabled, restoring the caller's context. Safe to call from any + * consumer entry point (poll, close, unsubscribe). + */ + public static void closeLingeringConsumeScope(KafkaConsumerInfo info) { + if (!shouldDeferConsumerScope()) { + return; + } + // restore the caller's context when the lingering consume span is still active on this thread + AgentSpan active = AgentTracer.activeSpan(); + if (active != null && KafkaDecorator.KAFKA_CONSUME.equals(active.getOperationName())) { + if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { + AgentTracer.closeLingeringIterationScope(); + } else { + AgentSpan swappedOut = AgentSpan.fromContext(Context.root().swap()); + if (swappedOut != null) { + swappedOut.finishWithEndToEnd(); + } + } + } + // owner-aware: finish the exact deferred span regardless of thread/context (CAS-safe if already + // finished) + if (info != null) { + AgentSpan deferred = info.getAndClearDeferredConsumeSpan(); + if (deferred != null) { + deferred.finishWithEndToEnd(); + } + } + } + public static String extractGroup(KafkaConsumerInfo kafkaConsumerInfo) { if (kafkaConsumerInfo != null) { return kafkaConsumerInfo.getConsumerGroup(); diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterable.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterable.java index 5f05996b3e9..357f28d8817 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterable.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterable.java @@ -10,6 +10,7 @@ public class TracingIterable implements Iterable>, TracingI private final String group; private final String clusterId; private final String bootstrapServers; + private final KafkaConsumerInfo kafkaConsumerInfo; public TracingIterable( final Iterable> delegate, @@ -17,20 +18,28 @@ public TracingIterable( final KafkaDecorator decorator, String group, String clusterId, - String bootstrapServers) { + String bootstrapServers, + KafkaConsumerInfo kafkaConsumerInfo) { this.delegate = delegate; this.operationName = operationName; this.decorator = decorator; this.group = group; this.clusterId = clusterId; this.bootstrapServers = bootstrapServers; + this.kafkaConsumerInfo = kafkaConsumerInfo; } @Override public Iterator> iterator() { // every iteration will add spans. Not only the very first one return new TracingIterator( - delegate.iterator(), operationName, decorator, group, clusterId, bootstrapServers); + delegate.iterator(), + operationName, + decorator, + group, + clusterId, + bootstrapServers, + kafkaConsumerInfo); } @Override diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java index 93179e4e3f2..afa05e938be 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java @@ -47,6 +47,7 @@ public class TracingIterator implements Iterator> { private final String group; private final String clusterId; private final String bootstrapServers; + private final KafkaConsumerInfo kafkaConsumerInfo; public TracingIterator( final Iterator> delegateIterator, @@ -54,32 +55,53 @@ public TracingIterator( final KafkaDecorator decorator, String group, String clusterId, - String bootstrapServers) { + String bootstrapServers, + KafkaConsumerInfo kafkaConsumerInfo) { this.delegateIterator = delegateIterator; this.operationName = operationName; this.decorator = decorator; this.group = group; this.clusterId = clusterId; this.bootstrapServers = bootstrapServers; + this.kafkaConsumerInfo = kafkaConsumerInfo; } @Override public boolean hasNext() { boolean moreRecords = delegateIterator.hasNext(); if (!moreRecords) { - // no more records, use this as a signal to close the last iteration scope - if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { - closePrevious(true); - } else { - final AgentSpan previousSpan = AgentSpan.fromContext(Context.root().swap()); - if (previousSpan != null) { - previousSpan.finishWithEndToEnd(); + if (!KafkaConsumerInstrumentationHelper.shouldDeferConsumerScope()) { + // no more records, use this as a signal to close the last iteration scope + if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { + closePrevious(true); + } else { + final AgentSpan previousSpan = AgentSpan.fromContext(Context.root().swap()); + if (previousSpan != null) { + previousSpan.finishWithEndToEnd(); + } } + } else { + // deferring: leave the last record's scope active past the loop; record a handle to it + captureDeferredConsumeSpan(); } } return moreRecords; } + /** + * Stashes the active {@code kafka.consume} span on the per-consumer {@link KafkaConsumerInfo} so + * a later poll()/close()/unsubscribe() can finish that precise span regardless of thread or + * stack. + */ + protected void captureDeferredConsumeSpan() { + if (kafkaConsumerInfo != null) { + AgentSpan span = AgentTracer.activeSpan(); + if (span != null && KafkaDecorator.KAFKA_CONSUME.equals(span.getOperationName())) { + kafkaConsumerInfo.setDeferredConsumeSpan(span); + } + } + } + @Override public ConsumerRecord next() { final ConsumerRecord next = delegateIterator.next(); @@ -90,7 +112,15 @@ public boolean hasNext() { protected void startNewRecordSpan(ConsumerRecord val) { try { if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { - closePrevious(true); + // real records always close the previous scope; the terminal close is deferred when the + // flag is on -- see hasNext(). + if (val != null || !KafkaConsumerInstrumentationHelper.shouldDeferConsumerScope()) { + closePrevious(true); + } else { + // terminal null record while deferring: leave the last consume span active, record a + // handle so a later poll()/close()/unsubscribe() finishes it + captureDeferredConsumeSpan(); + } } else if (val == null) { // previous message span was the last final AgentSpan previousSpan = AgentSpan.fromContext(Context.root().swap()); if (previousSpan != null) { diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingList.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingList.java index 2cb5f4c698f..8a1c3679f3e 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingList.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingList.java @@ -14,6 +14,7 @@ public class TracingList implements List>, TracingIterableD private final String group; private final String clusterId; private final String bootstrapServers; + private final KafkaConsumerInfo kafkaConsumerInfo; public TracingList( final List> delegate, @@ -21,13 +22,15 @@ public TracingList( final KafkaDecorator decorator, String group, String clusterId, - String bootstrapServers) { + String bootstrapServers, + KafkaConsumerInfo kafkaConsumerInfo) { this.operationName = operationName; this.decorator = decorator; this.delegate = delegate; this.group = group; this.clusterId = clusterId; this.bootstrapServers = bootstrapServers; + this.kafkaConsumerInfo = kafkaConsumerInfo; } @Override @@ -140,7 +143,13 @@ public int lastIndexOf(final Object o) { public ListIterator> listIterator(final int index) { // every iteration will add spans. Not only the very first one return new TracingListIterator( - delegate.listIterator(index), operationName, decorator, group, clusterId, bootstrapServers); + delegate.listIterator(index), + operationName, + decorator, + group, + clusterId, + bootstrapServers, + kafkaConsumerInfo); } @Override @@ -151,7 +160,8 @@ public int lastIndexOf(final Object o) { decorator, group, clusterId, - bootstrapServers); + bootstrapServers, + kafkaConsumerInfo); } @Override diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingListIterator.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingListIterator.java index 34e7bbe00df..01509c3d388 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingListIterator.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingListIterator.java @@ -19,8 +19,16 @@ public TracingListIterator( KafkaDecorator decorator, String group, String clusterId, - String bootstrapServers) { - super(delegateIterator, operationName, decorator, group, clusterId, bootstrapServers); + String bootstrapServers, + KafkaConsumerInfo kafkaConsumerInfo) { + super( + delegateIterator, + operationName, + decorator, + group, + clusterId, + bootstrapServers, + kafkaConsumerInfo); this.delegateIterator = delegateIterator; } @@ -28,14 +36,19 @@ public TracingListIterator( public boolean hasPrevious() { boolean moreRecords = delegateIterator.hasPrevious(); if (!moreRecords) { - // no more records, use this as a signal to close the last iteration scope - if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { - closePrevious(true); - } else { - final AgentSpan previousSpan = AgentSpan.fromContext(Context.root().swap()); - if (previousSpan != null) { - previousSpan.finishWithEndToEnd(); + if (!KafkaConsumerInstrumentationHelper.shouldDeferConsumerScope()) { + // no more records, use this as a signal to close the last iteration scope + if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { + closePrevious(true); + } else { + final AgentSpan previousSpan = AgentSpan.fromContext(Context.root().swap()); + if (previousSpan != null) { + previousSpan.finishWithEndToEnd(); + } } + } else { + // deferring: leave the last record's scope active past the loop; record a handle to it + captureDeferredConsumeSpan(); } } return moreRecords; diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/test/groovy/KafkaClientTestBase.groovy b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/test/groovy/KafkaClientTestBase.groovy index a3168dac5c1..1dcfa626333 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/test/groovy/KafkaClientTestBase.groovy +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/test/groovy/KafkaClientTestBase.groovy @@ -42,9 +42,12 @@ import org.springframework.kafka.listener.MessageListener import org.springframework.kafka.test.rule.KafkaEmbedded import org.springframework.kafka.test.utils.ContainerTestUtils import org.springframework.kafka.test.utils.KafkaTestUtils +import spock.lang.IgnoreIf import spock.lang.Shared +import java.util.concurrent.Callable import java.util.concurrent.ExecutionException +import java.util.concurrent.Executors import java.util.concurrent.Future import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.TimeUnit @@ -165,6 +168,18 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { "kafka.consume" } + // when true, the consumer-scope-deferring flag is enabled for this variant, so same-thread + // post-loop work is expected to nest under the last record's consume span + boolean deferConsumerScope() { + false + } + + // whether this variant runs under the legacy context manager (which has the iteration-keep-alive + // timer backstop); false for the context-swap manager, which has no such backstop + boolean legacyContextManager() { + true + } + String serviceForTimeInQueue() { "kafka" } @@ -594,6 +609,10 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { container?.stop() } + // when the scope is deferred, the single (and therefore last) record's consume span is left + // active past the loop instead of flushing immediately, so this test's immediate-flush + // assumption does not hold; the dedicated deferred-close cases cover the ON behavior + @IgnoreIf({ instance.deferConsumerScope() }) def "test records(TopicPartition) kafka consume"() { setup: // set up the Kafka consumer properties @@ -648,6 +667,8 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { producer.close() } + // see the deferred-scope note on "test records(TopicPartition) kafka consume" above + @IgnoreIf({ instance.deferConsumerScope() }) def "test records(TopicPartition).subList kafka consume"() { setup: // set up the Kafka consumer properties @@ -703,6 +724,8 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { producer.close() } + // see the deferred-scope note on "test records(TopicPartition) kafka consume" above + @IgnoreIf({ instance.deferConsumerScope() }) def "test records(TopicPartition).forEach kafka consume"() { setup: // set up the Kafka consumer properties @@ -758,6 +781,9 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { producer.close() } + // the final forward and final backward record are each "last" for their direction, so both leave + // a deferred, still-active consume span behind when the flag is on + @IgnoreIf({ instance.deferConsumerScope() }) def "test iteration backwards over ConsumerRecords"() { setup: def kafkaPartition = 0 @@ -866,6 +892,396 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { producer.close() } + // when a consumer buffers records and writes to the database *after* the per-record iterator + // loop, the write span is disconnected from the consume span unless the scope is deferred + def "test work done after the consume loop is disconnected from the consume span"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps("sender", "false", embeddedKafka) + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + def greeting = "Hello from MockConsumer!" + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, greeting)) + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + + // No DB work inside the iterator loop -- payloads are only buffered here. + def buffer = [] + for (record in pollResult) { + buffer.add(record.value()) + } + // "db.write" stands in for the application's database write, run after the consume loop. + buffer.each { + runUnderTrace("db.write") {} + } + // close() is a deferred-scope trigger: it finishes the lingering consume span so the (possibly + // merged) trace flushes deterministically, without relying on the iteration-keep-alive timer + // (which the context-swap manager does not have). + consumer.close() + + then: + // the produced record was actually consumed (fail clearly here instead of timing out below) + buffer.size() == 1 + // producer trace + consume trace (+ queue span, same trace) + the db.write trace -- when + // deferConsumerScope() is enabled, the db.write merges into the still-open consume span's + // trace instead of flushing as its own. + TEST_WRITER.waitForTraces(deferConsumerScope() ? 2 : 3) + List spans = TEST_WRITER.flatten() + def consumeSpan = spans.find { + it.operationName.toString() == operationForConsumer() + } + def dbWriteSpan = spans.find { + it.operationName.toString() == "db.write" + } + + consumeSpan != null + dbWriteSpan != null + + if (!deferConsumerScope()) { + // bug: the DB write is its own root trace, not a child of the consume span, and the consume + // span is left childless + assert dbWriteSpan.parentId == 0 + assert dbWriteSpan.traceId != consumeSpan.traceId + assert spans.every { + it.parentId != consumeSpan.spanId + } + } else { + // fixed: the post-loop DB write nests under the deferred consume span + assert dbWriteSpan.traceId == consumeSpan.traceId + assert dbWriteSpan.parentId == consumeSpan.spanId + } + + cleanup: + consumer.close() + producer.close() + } + + // with N buffered records and a single post-loop flush, only the LAST record's consume span + // should stay open long enough to parent the flush; records 1..N-1 still get independent spans + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test single post-loop write nests under the last record's consume span when consumer scope is deferred"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps("sender", "false", embeddedKafka) + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + def recordCount = 3 + (0..(SHARED_TOPIC, kafkaPartition, null, "greeting-$it".toString())).get() + } + TEST_WRITER.waitForTraces(recordCount) + def pollResult = KafkaTestUtils.getRecords(consumer) + + def buffer = [] + for (record in pollResult) { + buffer.add(record.value()) + } + // Single flush after the whole batch, standing in for a buffered DB write. + runUnderTrace("db.write") {} + // close() finishes the lingering last-record consume span so its merged trace flushes + // deterministically (the context-swap manager has no iteration-keep-alive timer backstop). + consumer.close() + + then: + buffer.size() == recordCount + // recordCount producer traces + (recordCount - 1) independent early-record consume traces + + // one merged trace for the last record's consume span and the db.write it now parents. + TEST_WRITER.waitForTraces(2 * recordCount) + List spans = TEST_WRITER.flatten() + def consumeSpans = spans.findAll { + it.operationName.toString() == operationForConsumer() + }.sort { it.getTag(InstrumentationTags.OFFSET) as int } + def dbWriteSpan = spans.find { + it.operationName.toString() == "db.write" + } + + consumeSpans.size() == recordCount + dbWriteSpan != null + def lastConsumeSpan = consumeSpans.last() + def earlierConsumeSpans = consumeSpans[0..-2] + + dbWriteSpan.traceId == lastConsumeSpan.traceId + dbWriteSpan.parentId == lastConsumeSpan.spanId + earlierConsumeSpans.every { + it.traceId != lastConsumeSpan.traceId + } + + cleanup: + consumer?.close() + producer?.close() + } + + // close trigger #1: a second poll() must finish the previous batch's lingering consume span + // (its trace flushes) without the new poll cycle's work nesting under it + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test a second poll() finishes the previous lingering consume span without nesting under it"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps("sender", "false", embeddedKafka) + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "first")).get() + TEST_WRITER.waitForTraces(1) + def firstBatch = KafkaTestUtils.getRecords(consumer) + for (record in firstBatch) { + // no-op: iterating to the end is what arms the deferred close for this batch's last record + } + DDSpan lingering = (DDSpan) activeSpan() + + // Send from a separate thread: the first record's consume span is deliberately still active + // (lingering) on this thread, and we don't want it to become the ambient parent of the + // "second" message's own producer span (which would then propagate into its consume span via + // the Kafka header trace context and defeat this test's own assertions below). + Thread.start { + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "second")).get() + }.join() + TEST_WRITER.waitForTraces(2) + def secondBatch = KafkaTestUtils.getRecords(consumer) + def secondValues = [] + for (record in secondBatch) { + secondValues.add(record.value()) + } + // the second batch's only record is itself now the last record of its loop, so it is left + // lingering too (armed for its own deferred close) -- grab it directly instead of relying on + // TEST_WRITER, which won't have flushed it yet. + DDSpan secondLingering = (DDSpan) activeSpan() + + then: + firstBatch.count() == 1 + secondValues == ["second"] + lingering != null + lingering.operationName.toString() == operationForConsumer() + // the second poll() closes the first batch's lingering span, flushing its trace, without the + // second batch's own (still-lingering) consume span nesting under it + TEST_WRITER.waitUntilReported(lingering, 10, TimeUnit.SECONDS) + lingering.isFinished() + secondLingering != null + secondLingering.operationName.toString() == operationForConsumer() + secondLingering.traceId != lingering.traceId + secondLingering.parentId != lingering.spanId + + cleanup: + consumer?.close() + producer?.close() + } + + // close trigger #2: with no further poll(), consumer.close() must finish a lingering consume span + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test consumer close() finishes a lingering consume span"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps("sender", "false", embeddedKafka) + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "greeting")).get() + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + for (record in pollResult) { + // no-op: iterating to the end is what arms the deferred close for the last record + } + DDSpan lingering = (DDSpan) activeSpan() + consumer.close() + + then: + pollResult.count() == 1 + lingering != null + lingering.operationName.toString() == operationForConsumer() + TEST_WRITER.waitUntilReported(lingering, 10, TimeUnit.SECONDS) + lingering.isFinished() + + cleanup: + producer?.close() + } + + // owner-aware cleanup: a consume span deferred while iterating on one thread must still be + // finished by a close() that runs on a different thread, where it is not top-of-stack + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test a consume span deferred on a worker thread is finished by a close on another thread"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps("sender", "false", embeddedKafka) + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "greeting")).get() + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + + // iterate on a worker thread so the deferred consume span is left active there, never + // top-of-stack on the main thread that closes the consumer + DDSpan lingering = null + Thread.start { + for (record in pollResult) { + // no-op: iterating to the end arms the deferred close for the last record + } + lingering = (DDSpan) activeSpan() + }.join() + + consumer.close() + + then: + pollResult.count() == 1 + lingering != null + lingering.operationName.toString() == operationForConsumer() + // only the per-consumer owner-aware handle can reach this span from the closing thread + TEST_WRITER.waitUntilReported(lingering, 10, TimeUnit.SECONDS) + lingering.isFinished() + + cleanup: + producer?.close() + } + + // owner-aware cleanup across a *reused* worker thread: even when records are iterated on a pooled + // executor thread, a later close() finishes the exact deferred span via the per-consumer handle. + // (The executor thread's own context is not restored cross-thread -- a documented context-swap + // limitation -- so only the finish guarantee is asserted.) + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test a consume span deferred on a reused executor thread is finished by a later close"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps("sender", "false", embeddedKafka) + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + def executor = Executors.newSingleThreadExecutor() + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "greeting")).get() + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + + DDSpan lingering = executor.submit({ + -> + for (record in pollResult) { + } + (DDSpan) activeSpan() + } as Callable).get() + + consumer.close() + + then: + pollResult.count() == 1 + lingering != null + lingering.operationName.toString() == operationForConsumer() + TEST_WRITER.waitUntilReported(lingering, 10, TimeUnit.SECONDS) + lingering.isFinished() + + cleanup: + executor?.shutdownNow() + producer?.close() + } + + // legacy-manager only -- the context-swap manager has no iteration-keep-alive backstop (the + // deferred span is finished on the next poll/close/unsubscribe instead) + @IgnoreIf({ !instance.deferConsumerScope() || !instance.legacyContextManager() }) + def "test a lingering consume span is eventually finished with no further trigger"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps("sender", "false", embeddedKafka) + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "greeting")).get() + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + for (record in pollResult) { + // no-op: iterating to the end is what defers the last record's consume scope + } + DDSpan lingering = (DDSpan) activeSpan() + + then: + pollResult.count() == 1 + lingering != null + !lingering.isFinished() + // no further poll()/close()/unsubscribe() ever arrives -- the native iteration-scope cleaner + // (keep-alive) is what finishes this + TEST_WRITER.waitUntilReported(lingering, 10, TimeUnit.SECONDS) + lingering.isFinished() + + cleanup: + consumer?.close() + producer?.close() + } + + // committing offsets must NOT be treated as a close trigger -- the lingering consume span must + // still be open immediately after a commitSync() call + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test committing offsets does not finish a lingering consume span"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps("sender", "false", embeddedKafka) + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "greeting")).get() + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + for (record in pollResult) { + // no-op: iterating to the end is what arms the deferred close for the last record + } + DDSpan lingering = (DDSpan) activeSpan() + consumer.commitSync() + + then: + pollResult.count() == 1 + lingering != null + !lingering.isFinished() + + cleanup: + consumer?.close() + producer?.close() + } + def "test spring kafka template produce and batch consume"() { setup: def senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString()) @@ -1192,6 +1608,10 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { recs.next().value() == "test-dsm-consume-transaction" !recs.hasNext() + // finish the (possibly deferred) consume span so its trace flushes deterministically, without + // relying on the iteration-keep-alive timer (absent under the context-swap manager) + consumer.close() + // The consume span is created by TracingIterator when iterating over records // Find the consumer span with the DSM transaction tags TEST_WRITER.waitForTraces(2) @@ -1548,6 +1968,44 @@ class KafkaClientContextSwapForkedTest extends KafkaClientV0ForkedTest { super.configurePreAgent() injectSysConfig(TraceInstrumentationConfig.LEGACY_CONTEXT_MANAGER_ENABLED, "false") } + + @Override + boolean legacyContextManager() { + false + } +} + +// exercises DD_TRACE_KAFKA_CREATE_CONSUMER_SCOPE_ENABLED=true in LEGACY context-manager mode (the +// only mode the flag defers in). A short scope-iteration keep-alive is injected so the native +// RootIterationScopeCleaner reaps a lingering consume span quickly when no further trigger arrives. +class KafkaClientCreateConsumerScopeForkedTest extends KafkaClientV0ForkedTest { + @Override + void configurePreAgent() { + super.configurePreAgent() + injectSysConfig(TraceInstrumentationConfig.KAFKA_CREATE_CONSUMER_SCOPE_ENABLED, "true") + injectSysConfig(TracerConfig.SCOPE_ITERATION_KEEP_ALIVE, "1") + } + + @Override + boolean deferConsumerScope() { + true + } +} + +// the flag on under the NEW context-swap manager: the consumer scope is deferred here too (via the +// owner-aware handle + a context swap-back on the next poll/close/unsubscribe), so deferConsumerScope() +// stays true. legacyContextManager() is false because this manager has no iteration-keep-alive backstop. +class KafkaClientCreateConsumerScopeContextSwapForkedTest extends KafkaClientCreateConsumerScopeForkedTest { + @Override + void configurePreAgent() { + super.configurePreAgent() + injectSysConfig(TraceInstrumentationConfig.LEGACY_CONTEXT_MANAGER_ENABLED, "false") + } + + @Override + boolean legacyContextManager() { + false + } } class KafkaClientBadBase64HeaderForkedTest extends InstrumentationSpecification { diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfoInstrumentation.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfoInstrumentation.java index c94c49369ed..e5afd49d8d3 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfoInstrumentation.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfoInstrumentation.java @@ -100,5 +100,12 @@ public void methodAdvice(MethodTransformer transformer) { .and(takesArguments(1)) .and(returns(named("org.apache.kafka.clients.consumer.ConsumerRecords"))), packageName + ".RecordsAdvice"); + // close()/unsubscribe() end the poll loop for good, so a consume span deferred past the last + // poll must be finished here rather than left dangling. + transformer.applyAdvice( + isMethod().and(isPublic()).and(named("close")), packageName + ".ConsumerCloseAdvice"); + transformer.applyAdvice( + isMethod().and(isPublic()).and(named("unsubscribe")).and(takesArguments(0)), + packageName + ".ConsumerCloseAdvice"); } } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/LegacyKafkaConsumerInfoInstrumentation.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/LegacyKafkaConsumerInfoInstrumentation.java index dd36ff1d934..2c35a04767c 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/LegacyKafkaConsumerInfoInstrumentation.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/LegacyKafkaConsumerInfoInstrumentation.java @@ -100,5 +100,12 @@ public void methodAdvice(MethodTransformer transformer) { .and(takesArguments(1)) .and(returns(named("org.apache.kafka.clients.consumer.ConsumerRecords"))), packageName + ".RecordsAdvice"); + // close()/unsubscribe() end the poll loop for good, so a consume span deferred past the last + // poll must be finished here rather than left dangling. + transformer.applyAdvice( + isMethod().and(isPublic()).and(named("close")), packageName + ".ConsumerCloseAdvice"); + transformer.applyAdvice( + isMethod().and(isPublic()).and(named("unsubscribe")).and(takesArguments(0)), + packageName + ".ConsumerCloseAdvice"); } } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ConsumerCloseAdvice.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ConsumerCloseAdvice.java new file mode 100644 index 00000000000..eb3903f7b33 --- /dev/null +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ConsumerCloseAdvice.java @@ -0,0 +1,21 @@ +package datadog.trace.instrumentation.kafka_clients38; + +import datadog.trace.api.Config; +import datadog.trace.bootstrap.InstrumentationContext; +import net.bytebuddy.asm.Advice; +import org.apache.kafka.clients.consumer.internals.ConsumerDelegate; + +/** + * Finishes a {@code kafka.consume} span left active past the last poll when {@code close()}/{@code + * unsubscribe()} ends the poll loop (see {@link TracingIterator}). + */ +public class ConsumerCloseAdvice { + @Advice.OnMethodEnter(suppress = Throwable.class) + public static void onEnter(@Advice.This ConsumerDelegate consumer) { + if (Config.get().isKafkaCreateConsumerScopeEnabled()) { + KafkaConsumerInfo kafkaConsumerInfo = + InstrumentationContext.get(ConsumerDelegate.class, KafkaConsumerInfo.class).get(consumer); + KafkaConsumerInstrumentationHelper.closeLingeringConsumeScope(kafkaConsumerInfo); + } + } +} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IterableAdvice.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IterableAdvice.java index c6cc52a3d9e..d52d6d9599e 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IterableAdvice.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IterableAdvice.java @@ -29,7 +29,8 @@ public static void wrap( KafkaDecorator.CONSUMER_DECORATE, group, clusterId, - bootstrapServers); + bootstrapServers, + kafkaConsumerInfo); } } } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IteratorAdvice.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IteratorAdvice.java index 39859c20e19..f808f45f449 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IteratorAdvice.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/IteratorAdvice.java @@ -30,7 +30,8 @@ public static void wrap( KafkaDecorator.CONSUMER_DECORATE, group, clusterId, - bootstrapServers); + bootstrapServers, + kafkaConsumerInfo); } } } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfo.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfo.java index 947bbd623ae..26530613a3f 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfo.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfo.java @@ -1,7 +1,9 @@ package datadog.trace.instrumentation.kafka_clients38; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import java.util.Objects; import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.Metadata; public class KafkaConsumerInfo { @@ -9,6 +11,10 @@ public class KafkaConsumerInfo { private final Metadata metadata; private final String bootstrapServers; + // handle to the consume span this consumer left lingering past its last poll loop; not part of + // consumer identity, so excluded from equals/hashCode + private final AtomicReference deferredConsumeSpan = new AtomicReference<>(); + public KafkaConsumerInfo(String consumerGroup, Metadata metadata, String bootstrapServers) { this.consumerGroup = consumerGroup; this.metadata = metadata; @@ -21,6 +27,14 @@ public KafkaConsumerInfo(String consumerGroup, String bootstrapServers) { this.bootstrapServers = bootstrapServers; } + public void setDeferredConsumeSpan(AgentSpan span) { + deferredConsumeSpan.set(span); + } + + public AgentSpan getAndClearDeferredConsumeSpan() { + return deferredConsumeSpan.getAndSet(null); + } + public Optional getConsumerGroup() { return Optional.ofNullable(consumerGroup); } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInstrumentationHelper.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInstrumentationHelper.java index 11c020e78e3..83a00012ac3 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInstrumentationHelper.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInstrumentationHelper.java @@ -1,10 +1,21 @@ package datadog.trace.instrumentation.kafka_clients38; +import datadog.context.Context; +import datadog.trace.api.Config; +import datadog.trace.api.InstrumenterConfig; import datadog.trace.bootstrap.ContextStore; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.AgentTracer; import datadog.trace.instrumentation.kafka_common.MetadataState; import org.apache.kafka.clients.Metadata; public class KafkaConsumerInstrumentationHelper { + + /** Whether the last record's consume span should be kept active past the poll loop. */ + public static boolean shouldDeferConsumerScope() { + return Config.get().isKafkaCreateConsumerScopeEnabled(); + } + public static String extractGroup(KafkaConsumerInfo kafkaConsumerInfo) { if (kafkaConsumerInfo != null) { return kafkaConsumerInfo.getConsumerGroup().get(); @@ -28,4 +39,35 @@ public static String extractClusterId( public static String extractBootstrapServers(KafkaConsumerInfo kafkaConsumerInfo) { return kafkaConsumerInfo == null ? null : kafkaConsumerInfo.getBootstrapServers().get(); } + + /** + * Finishes the {@code kafka.consume} span deliberately left active past the poll loop when + * consumer-scope deferral is enabled, restoring the caller's context. Safe to call from any + * consumer entry point (poll, close, unsubscribe). + */ + public static void closeLingeringConsumeScope(KafkaConsumerInfo info) { + if (!shouldDeferConsumerScope()) { + return; + } + // restore the caller's context when the lingering consume span is still active on this thread + AgentSpan active = AgentTracer.activeSpan(); + if (active != null && KafkaDecorator.KAFKA_CONSUME.equals(active.getOperationName())) { + if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { + AgentTracer.closeLingeringIterationScope(); + } else { + AgentSpan swappedOut = AgentSpan.fromContext(Context.root().swap()); + if (swappedOut != null) { + swappedOut.finishWithEndToEnd(); + } + } + } + // owner-aware: finish the exact deferred span regardless of thread/context (CAS-safe if already + // finished) + if (info != null) { + AgentSpan deferred = info.getAndClearDeferredConsumeSpan(); + if (deferred != null) { + deferred.finishWithEndToEnd(); + } + } + } } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ListAdvice.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ListAdvice.java index 1c65b8a2ccd..2b7d5cea9a6 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ListAdvice.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ListAdvice.java @@ -28,7 +28,13 @@ public static void wrap( KafkaConsumerInstrumentationHelper.extractBootstrapServers(kafkaConsumerInfo); iterable = new TracingList( - iterable, KAFKA_CONSUME, CONSUMER_DECORATE, group, clusterId, bootstrapServers); + iterable, + KAFKA_CONSUME, + CONSUMER_DECORATE, + group, + clusterId, + bootstrapServers, + kafkaConsumerInfo); } } } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/RecordsAdvice.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/RecordsAdvice.java index b8f3dff049a..6b6001c31a2 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/RecordsAdvice.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/RecordsAdvice.java @@ -27,9 +27,12 @@ public class RecordsAdvice { @Advice.OnMethodEnter(suppress = Throwable.class) public static AgentScope onEnter(@Advice.This ConsumerDelegate consumer) { - // Set cluster ID in ClusterIdHolder for Schema Registry instrumentation KafkaConsumerInfo kafkaConsumerInfo = InstrumentationContext.get(ConsumerDelegate.class, KafkaConsumerInfo.class).get(consumer); + // a poll() means any span deferred past the previous poll loop is done -- close it now (no-op + // unless consumer-scope deferral is enabled) + KafkaConsumerInstrumentationHelper.closeLingeringConsumeScope(kafkaConsumerInfo); + // Set cluster ID in ClusterIdHolder for Schema Registry instrumentation if (kafkaConsumerInfo != null && Config.get().isDataStreamsEnabled()) { String clusterId = KafkaConsumerInstrumentationHelper.extractClusterId( diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterable.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterable.java index 59b53b1c363..92c58f4fb83 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterable.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterable.java @@ -10,6 +10,7 @@ public class TracingIterable implements Iterable>, TracingI private final String group; private final String clusterId; private final String bootstrapServers; + private final KafkaConsumerInfo kafkaConsumerInfo; public TracingIterable( final Iterable> delegate, @@ -17,20 +18,28 @@ public TracingIterable( final KafkaDecorator decorator, String group, String clusterId, - String bootstrapServers) { + String bootstrapServers, + KafkaConsumerInfo kafkaConsumerInfo) { this.delegate = delegate; this.operationName = operationName; this.decorator = decorator; this.group = group; this.clusterId = clusterId; this.bootstrapServers = bootstrapServers; + this.kafkaConsumerInfo = kafkaConsumerInfo; } @Override public Iterator> iterator() { // every iteration will add spans. Not only the very first one return new TracingIterator( - delegate.iterator(), operationName, decorator, group, clusterId, bootstrapServers); + delegate.iterator(), + operationName, + decorator, + group, + clusterId, + bootstrapServers, + kafkaConsumerInfo); } @Override diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterator.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterator.java index 7beba848473..74e952f8b9c 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterator.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterator.java @@ -43,6 +43,7 @@ public class TracingIterator implements Iterator> { private final String group; private final String clusterId; private final String bootstrapServers; + private final KafkaConsumerInfo kafkaConsumerInfo; public TracingIterator( final Iterator> delegateIterator, @@ -50,32 +51,53 @@ public TracingIterator( final KafkaDecorator decorator, String group, String clusterId, - String bootstrapServers) { + String bootstrapServers, + KafkaConsumerInfo kafkaConsumerInfo) { this.delegateIterator = delegateIterator; this.operationName = operationName; this.decorator = decorator; this.group = group; this.clusterId = clusterId; this.bootstrapServers = bootstrapServers; + this.kafkaConsumerInfo = kafkaConsumerInfo; } @Override public boolean hasNext() { boolean moreRecords = delegateIterator.hasNext(); if (!moreRecords) { - // no more records, use this as a signal to close the last iteration scope - if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { - closePrevious(true); - } else { - final AgentSpan previousSpan = AgentSpan.fromContext(Context.root().swap()); - if (previousSpan != null) { - previousSpan.finishWithEndToEnd(); + if (!KafkaConsumerInstrumentationHelper.shouldDeferConsumerScope()) { + // no more records, use this as a signal to close the last iteration scope + if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { + closePrevious(true); + } else { + final AgentSpan previousSpan = AgentSpan.fromContext(Context.root().swap()); + if (previousSpan != null) { + previousSpan.finishWithEndToEnd(); + } } + } else { + // deferring: leave the last record's scope active past the loop; record a handle to it + captureDeferredConsumeSpan(); } } return moreRecords; } + /** + * Stashes the active {@code kafka.consume} span on the per-consumer {@link KafkaConsumerInfo} so + * a later poll()/close()/unsubscribe() can finish that precise span regardless of thread or + * stack. + */ + protected void captureDeferredConsumeSpan() { + if (kafkaConsumerInfo != null) { + AgentSpan span = AgentTracer.activeSpan(); + if (span != null && KafkaDecorator.KAFKA_CONSUME.equals(span.getOperationName())) { + kafkaConsumerInfo.setDeferredConsumeSpan(span); + } + } + } + @Override public ConsumerRecord next() { final ConsumerRecord next = delegateIterator.next(); @@ -86,7 +108,15 @@ public boolean hasNext() { protected void startNewRecordSpan(ConsumerRecord val) { try { if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { - closePrevious(true); + // real records always close the previous scope; the terminal close is deferred when the + // flag is on -- see hasNext(). + if (val != null || !KafkaConsumerInstrumentationHelper.shouldDeferConsumerScope()) { + closePrevious(true); + } else { + // terminal null record while deferring: leave the last consume span active, record a + // handle so a later poll()/close()/unsubscribe() finishes it + captureDeferredConsumeSpan(); + } } else if (val == null) { // previous message span was the last final AgentSpan previousSpan = AgentSpan.fromContext(Context.root().swap()); if (previousSpan != null) { diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingList.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingList.java index 24cd318d175..4ab11d91c41 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingList.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingList.java @@ -14,6 +14,7 @@ public class TracingList implements List>, TracingIterableD private final String group; private final String clusterId; private final String bootstrapServers; + private final KafkaConsumerInfo kafkaConsumerInfo; public TracingList( final List> delegate, @@ -21,13 +22,15 @@ public TracingList( final KafkaDecorator decorator, String group, String clusterId, - String bootstrapServers) { + String bootstrapServers, + KafkaConsumerInfo kafkaConsumerInfo) { this.operationName = operationName; this.decorator = decorator; this.delegate = delegate; this.group = group; this.clusterId = clusterId; this.bootstrapServers = bootstrapServers; + this.kafkaConsumerInfo = kafkaConsumerInfo; } @Override @@ -140,7 +143,13 @@ public int lastIndexOf(final Object o) { public ListIterator> listIterator(final int index) { // every iteration will add spans. Not only the very first one return new TracingListIterator( - delegate.listIterator(index), operationName, decorator, group, clusterId, bootstrapServers); + delegate.listIterator(index), + operationName, + decorator, + group, + clusterId, + bootstrapServers, + kafkaConsumerInfo); } @Override @@ -151,7 +160,8 @@ public int lastIndexOf(final Object o) { decorator, group, clusterId, - bootstrapServers); + bootstrapServers, + kafkaConsumerInfo); } @Override diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingListIterator.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingListIterator.java index 3b17006d562..148246f161a 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingListIterator.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingListIterator.java @@ -19,8 +19,16 @@ public TracingListIterator( KafkaDecorator decorator, String group, String clusterId, - String bootstrapServers) { - super(delegateIterator, operationName, decorator, group, clusterId, bootstrapServers); + String bootstrapServers, + KafkaConsumerInfo kafkaConsumerInfo) { + super( + delegateIterator, + operationName, + decorator, + group, + clusterId, + bootstrapServers, + kafkaConsumerInfo); this.delegateIterator = delegateIterator; } @@ -28,14 +36,19 @@ public TracingListIterator( public boolean hasPrevious() { boolean moreRecords = delegateIterator.hasPrevious(); if (!moreRecords) { - // no more records, use this as a signal to close the last iteration scope - if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { - closePrevious(true); - } else { - final AgentSpan previousSpan = AgentSpan.fromContext(Context.root().swap()); - if (previousSpan != null) { - previousSpan.finishWithEndToEnd(); + if (!KafkaConsumerInstrumentationHelper.shouldDeferConsumerScope()) { + // no more records, use this as a signal to close the last iteration scope + if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { + closePrevious(true); + } else { + final AgentSpan previousSpan = AgentSpan.fromContext(Context.root().swap()); + if (previousSpan != null) { + previousSpan.finishWithEndToEnd(); + } } + } else { + // deferring: leave the last record's scope active past the loop; record a handle to it + captureDeferredConsumeSpan(); } } return moreRecords; diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaClientTestBase.groovy b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaClientTestBase.groovy index a3f451f126d..3a7b115aa25 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaClientTestBase.groovy +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaClientTestBase.groovy @@ -36,8 +36,11 @@ import org.springframework.kafka.test.EmbeddedKafkaBroker import org.springframework.kafka.test.EmbeddedKafkaKraftBroker import org.springframework.kafka.test.utils.ContainerTestUtils import org.springframework.kafka.test.utils.KafkaTestUtils +import spock.lang.IgnoreIf +import java.util.concurrent.Callable import java.util.concurrent.ExecutionException +import java.util.concurrent.Executors import java.util.concurrent.Future import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.TimeUnit @@ -147,6 +150,18 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { "kafka" } + // whether the consumer-scope-deferring flag is on for this variant, i.e. the last record's + // consume span is deferred past the poll loop instead of closed immediately + boolean deferConsumerScope() { + false + } + + // whether this variant runs under the legacy context manager (which has the iteration-keep-alive + // timer backstop); false for the context-swap manager, which has no such backstop + boolean legacyContextManager() { + true + } + abstract boolean hasQueueSpan() abstract boolean splitByDestination() @@ -521,6 +536,10 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { container?.stop() } + // when the scope is deferred, the single (and therefore last) record's consume span is left + // active past the loop instead of flushing immediately, so this test's immediate-flush + // assumption does not hold; the dedicated deferred-close cases cover the ON behavior + @IgnoreIf({ instance.deferConsumerScope() }) def "test records(TopicPartition) kafka consume"() { setup: // set up the Kafka consumer properties @@ -575,6 +594,8 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { producer.close() } + // see the deferred-scope note on "test records(TopicPartition) kafka consume" above + @IgnoreIf({ instance.deferConsumerScope() }) def "test records(TopicPartition).subList kafka consume"() { setup: @@ -632,6 +653,8 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { producer.close() } + // see the deferred-scope note on "test records(TopicPartition) kafka consume" above + @IgnoreIf({ instance.deferConsumerScope() }) def "test records(TopicPartition).forEach kafka consume"() { setup: @@ -689,6 +712,9 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { producer.close() } + // the final forward and final backward record are each "last" for their direction, so both leave + // a deferred, still-active consume span behind when the flag is on + @IgnoreIf({ instance.deferConsumerScope() }) def "test iteration backwards over ConsumerRecords"() { setup: @@ -889,6 +915,452 @@ abstract class KafkaClientTestBase extends VersionedNamingTestBase { producer?.close() } + // when a consumer buffers records and writes to the database *after* the per-record iterator + // loop, the write span is disconnected from the consume span when deferConsumerScope() is off, + // and reparented under it when a flag-ON forked variant overrides deferConsumerScope() to true + def "test work done after the consume loop is disconnected from the consume span"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps(embeddedKafka.getBrokersAsString(), "sender", "false") + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + def greeting = "Hello from MockConsumer!" + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, greeting)) + + then: + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + + // No DB work inside the iterator loop -- payloads are only buffered here. + def buffer = [] + for (record in pollResult) { + buffer.add(record.value()) + } + // "db.write" stands in for the application's database write, run after the consume loop. + buffer.each { + runUnderTrace("db.write") {} + } + // When deferConsumerScope() is on, the last record's consume span is intentionally left + // open past the loop; closing the consumer is one of the triggers that finishes it (see the + // dedicated close()/unsubscribe()/next-poll cases below) so the trace can flush. + // When off, the consume span already finished during the loop and this is a no-op for it. + consumer.close() + + then: + // the produced record was actually consumed (fail clearly here instead of timing out below) + buffer.size() == 1 + // Off: producer trace + consume trace (+ queue span, same trace) + the db.write root trace, + // written out as 3 separate batches since db.write is disconnected (THE BUG). + // On: producer trace + (consume trace + db.write), the latter now written out as a single + // batch because db.write is nested under -- and flushes together with -- the consume span. + TEST_WRITER.waitForTraces(deferConsumerScope() ? 2 : 3) + List spans = TEST_WRITER.flatten() + def consumeSpan = spans.find { + it.operationName.toString() == operationForConsumer() + } + def dbWriteSpan = spans.find { + it.operationName.toString() == "db.write" + } + + consumeSpan != null + dbWriteSpan != null + + if (deferConsumerScope()) { + // with the deferred-close flag on, the post-loop db.write nests under the last record's + // still-active consume span (assert required: conditions nested in if/else are not treated + // as implicit Spock conditions) + assert dbWriteSpan.traceId == consumeSpan.traceId + assert dbWriteSpan.parentId == consumeSpan.spanId + } else { + // bug: the DB write is its own root trace, not a child of the consume span, and the consume + // span is left childless + assert dbWriteSpan.parentId == 0 + assert dbWriteSpan.traceId != consumeSpan.traceId + assert spans.every { + it.parentId != consumeSpan.spanId + } + } + + cleanup: + consumer.close() + producer.close() + } + + // The following cases only apply when the deferred-close flag is on; the flag-off behavior + // above (and everywhere else in this suite) is unaffected. + + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test only the last of multiple records keeps its consume span active past the loop"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps(embeddedKafka.getBrokersAsString(), "sender", "false") + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + def greetings = ["msg 1", "msg 2", "msg 3"] + greetings.each { + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, it)) + } + + then: + TEST_WRITER.waitForTraces(3) + def pollResult = KafkaTestUtils.getRecords(consumer) + def records = pollResult.records(new TopicPartition(SHARED_TOPIC, kafkaPartition)) + records.size() == 3 + def lastRecord = records.get(records.size() - 1) + + for (record in pollResult) { + // no-op -- the loop itself drives TracingIterator#hasNext()/startNewRecordSpan() + } + def lastSpan = activeSpan() + + // only the LAST record's consume span survives the loop; earlier records still close in-loop. + lastSpan != null + lastSpan.operationName.toString() == operationForConsumer() + (lastSpan.getTag(InstrumentationTags.OFFSET) as long) == lastRecord.offset() + + consumer.close() + + then: + // 3 producer traces + 3 consume traces (a queue span, if any, shares the consume trace) + TEST_WRITER.waitForTraces(6) + + cleanup: + consumer.close() + producer.close() + } + + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test a lingering consume span is closed by the next poll, not leaked onto it"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps(embeddedKafka.getBrokersAsString(), "sender", "false") + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "msg 1")) + + then: + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + def receivedRecords = 0 + for (record in pollResult) { + // no-op -- driving the real (instrumented) iterator is what creates/activates the + // consume span; ConsumerRecords#count() reads an internal map size and never touches + // the wrapped iterator, so it must not be used here. + receivedRecords++ + } + receivedRecords == 1 + activeSpan() != null + activeSpan().operationName.toString() == operationForConsumer() + + // a second, empty poll is itself a close trigger for the span left lingering by the first + // poll -- see RecordsAdvice#onEnter, which closes it before this poll's own work starts. + def secondPoll = consumer.poll(java.time.Duration.ofMillis(500)) + secondPoll.count() == 0 + + then: + // the first record's consume trace (producer trace already counted) has now flushed. + TEST_WRITER.waitForTraces(2) + List spans = TEST_WRITER.flatten() + def consumeSpan = spans.find { + it.operationName.toString() == operationForConsumer() + } + consumeSpan != null + // the second poll's span (if DSM created one at all) must not be nested under the stale + // consume span from the first poll. + spans.every { + it.resourceName.toString() != "kafka.poll" || it.parentId != consumeSpan.spanId + } + + cleanup: + consumer.close() + producer.close() + } + + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test close() finishes a lingering consume span"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps(embeddedKafka.getBrokersAsString(), "sender", "false") + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "msg 1")) + + then: + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + def receivedRecords = 0 + for (record in pollResult) { + // no-op -- driving the real (instrumented) iterator is what creates/activates the + // consume span; ConsumerRecords#count() reads an internal map size and never touches + // the wrapped iterator, so it must not be used here. + receivedRecords++ + } + receivedRecords == 1 + activeSpan() != null + activeSpan().operationName.toString() == operationForConsumer() + + consumer.close() + + then: + TEST_WRITER.waitForTraces(2) + + cleanup: + producer.close() + } + + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test a consume span deferred on a worker thread is finished by a close on another thread"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps(embeddedKafka.getBrokersAsString(), "sender", "false") + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "msg 1")) + + then: + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + + // iterate on a worker thread so the deferred consume span is left active there, never + // top-of-stack on the main thread that closes the consumer + def received = new java.util.concurrent.atomic.AtomicInteger(0) + DDSpan lingering = null + Thread.start { + for (record in pollResult) { + // no-op -- driving the real (instrumented) iterator creates the consume span here + received.incrementAndGet() + } + lingering = (DDSpan) activeSpan() + }.join() + + received.get() == 1 + lingering != null + lingering.operationName.toString() == operationForConsumer() + + consumer.close() + + then: + // only the per-consumer owner-aware handle can reach the worker thread's span from here + TEST_WRITER.waitForTraces(2) + + cleanup: + producer.close() + } + + // owner-aware cleanup across a *reused* worker thread: even when records are iterated on a pooled + // executor thread, a later close() finishes the exact deferred span via the per-consumer handle. + // (The executor thread's own context is not restored cross-thread -- a documented context-swap + // limitation -- so only the finish guarantee is asserted.) + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test a consume span deferred on a reused executor thread is finished by a later close"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps(embeddedKafka.getBrokersAsString(), "sender", "false") + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + def executor = Executors.newSingleThreadExecutor() + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "msg 1")) + + then: + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + + def received = new java.util.concurrent.atomic.AtomicInteger(0) + DDSpan lingering = executor.submit({ + -> + for (record in pollResult) { + received.incrementAndGet() + } + (DDSpan) activeSpan() + } as Callable).get() + + received.get() == 1 + lingering != null + lingering.operationName.toString() == operationForConsumer() + + consumer.close() + + then: + // only the per-consumer owner-aware handle can reach the executor thread's span from here + TEST_WRITER.waitForTraces(2) + + cleanup: + executor?.shutdownNow() + producer.close() + } + + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test unsubscribe() finishes a lingering consume span"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps(embeddedKafka.getBrokersAsString(), "sender", "false") + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "msg 1")) + + then: + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + def receivedRecords = 0 + for (record in pollResult) { + // no-op -- driving the real (instrumented) iterator is what creates/activates the + // consume span; ConsumerRecords#count() reads an internal map size and never touches + // the wrapped iterator, so it must not be used here. + receivedRecords++ + } + receivedRecords == 1 + activeSpan() != null + activeSpan().operationName.toString() == operationForConsumer() + + consumer.unsubscribe() + + then: + TEST_WRITER.waitForTraces(2) + + cleanup: + consumer.close() + producer.close() + } + + // legacy-manager only -- the context-swap manager has no iteration-keep-alive backstop (the + // deferred span is finished on the next poll/close/unsubscribe instead) + @IgnoreIf({ !instance.deferConsumerScope() || !instance.legacyContextManager() }) + def "test a lingering consume span is eventually finished with no further trigger"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps(embeddedKafka.getBrokersAsString(), "sender", "false") + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "msg 1")) + + then: + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + def receivedRecords = 0 + for (record in pollResult) { + // no-op -- driving the real (instrumented) iterator is what creates/activates the + // consume span; ConsumerRecords#count() reads an internal map size and never touches + // the wrapped iterator, so it must not be used here. + receivedRecords++ + } + receivedRecords == 1 + activeSpan() != null + activeSpan().operationName.toString() == operationForConsumer() + + // no further poll, no close()/unsubscribe(), no commit -- the lingering ITERATION-sourced + // consume span is reaped by the scope manager's native root-iteration-scope keep-alive + // (RootIterationScopeCleaner), so it must eventually flush without any explicit trigger. + TEST_WRITER.waitForTraces(2) + + cleanup: + consumer.close() + producer.close() + } + + @IgnoreIf({ !instance.deferConsumerScope() }) + def "test commitSync() does not finish a lingering consume span"() { + setup: + def kafkaPartition = 0 + def consumerProperties = KafkaTestUtils.consumerProps(embeddedKafka.getBrokersAsString(), "sender", "false") + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + def consumer = new KafkaConsumer(consumerProperties) + + def senderProps = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + def producer = new KafkaProducer(senderProps) + + consumer.assign(Arrays.asList(new TopicPartition(SHARED_TOPIC, kafkaPartition))) + + when: + producer.send(new ProducerRecord(SHARED_TOPIC, kafkaPartition, null, "msg 1")) + + then: + TEST_WRITER.waitForTraces(1) + def pollResult = KafkaTestUtils.getRecords(consumer) + def receivedRecords = 0 + for (record in pollResult) { + // no-op -- driving the real (instrumented) iterator is what creates/activates the + // consume span; ConsumerRecords#count() reads an internal map size and never touches + // the wrapped iterator, so it must not be used here. + receivedRecords++ + } + receivedRecords == 1 + def lingeringSpan = activeSpan() + lingeringSpan != null + lingeringSpan.operationName.toString() == operationForConsumer() + + consumer.commitSync() + + then: + // commit is explicitly NOT a close trigger: the same span must still be active + // and unfinished immediately after the commit call. This is checked synchronously (rather + // than via a bounded waitForTraces window) because the tracer's generic root-iteration-scope + // keep-alive (dd.trace.scope.iteration.keep-alive, defaulted to 1s in tests) independently + // reaps this same lingering ITERATION-sourced scope around the same time -- a timing-based + // "did it flush yet" check would be racing that unrelated cleanup, not testing commit. + activeSpan() == lingeringSpan + !(lingeringSpan as DDSpan).isFinished() + + cleanup: + consumer.close() + producer.close() + } + def producerSpan( TraceAssert trace, Map config, @@ -1215,6 +1687,45 @@ class KafkaClientContextSwapForkedTest extends KafkaClientV0ForkedTest { super.configurePreAgent() injectSysConfig(TraceInstrumentationConfig.LEGACY_CONTEXT_MANAGER_ENABLED, "false") } + + @Override + boolean legacyContextManager() { + false + } +} + +// consumer-scope deferral turned on in LEGACY context-manager mode (the only mode the flag defers +// in) -- the last record's consume span is deferred past the poll loop instead of closed in-loop. A +// short scope-iteration keep-alive lets the native RootIterationScopeCleaner reap a lingering span +// quickly when no explicit trigger arrives. +class KafkaClientCreateConsumerScopeForkedTest extends KafkaClientV0ForkedTest { + @Override + void configurePreAgent() { + super.configurePreAgent() + injectSysConfig(TraceInstrumentationConfig.KAFKA_CREATE_CONSUMER_SCOPE_ENABLED, "true") + injectSysConfig(TracerConfig.SCOPE_ITERATION_KEEP_ALIVE, "1") + } + + @Override + boolean deferConsumerScope() { + true + } +} + +// the flag on under the NEW context-swap manager: the consumer scope is deferred here too (via the +// owner-aware handle + a context swap-back on the next poll/close/unsubscribe), so deferConsumerScope() +// stays true. legacyContextManager() is false because this manager has no iteration-keep-alive backstop. +class KafkaClientCreateConsumerScopeContextSwapForkedTest extends KafkaClientCreateConsumerScopeForkedTest { + @Override + void configurePreAgent() { + super.configurePreAgent() + injectSysConfig(TraceInstrumentationConfig.LEGACY_CONTEXT_MANAGER_ENABLED, "false") + } + + @Override + boolean legacyContextManager() { + false + } } class KafkaClientBadBase64HeaderForkedTest extends InstrumentationSpecification { diff --git a/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java b/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java index a57e5d37882..6c1be44f78a 100644 --- a/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java +++ b/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java @@ -302,6 +302,8 @@ public final class ConfigDefaults { static final boolean DEFAULT_TRACE_128_BIT_TRACEID_LOGGING_ENABLED = true; static final boolean DEFAULT_SECURE_RANDOM = false; + static final boolean DEFAULT_TRACE_KAFKA_CREATE_CONSUMER_SCOPE_ENABLED = false; + public static final int DEFAULT_TRACE_X_DATADOG_TAGS_MAX_LENGTH = 512; static final boolean DEFAULT_TRACE_HTTP_RESOURCE_REMOVE_TRAILING_SLASH = false; diff --git a/dd-trace-api/src/main/java/datadog/trace/api/config/TraceInstrumentationConfig.java b/dd-trace-api/src/main/java/datadog/trace/api/config/TraceInstrumentationConfig.java index 6d44cf8b249..47ddc795e30 100644 --- a/dd-trace-api/src/main/java/datadog/trace/api/config/TraceInstrumentationConfig.java +++ b/dd-trace-api/src/main/java/datadog/trace/api/config/TraceInstrumentationConfig.java @@ -110,6 +110,8 @@ public final class TraceInstrumentationConfig { "kafka.client.propagation.disabled.topics"; public static final String KAFKA_CLIENT_BASE64_DECODING_ENABLED = "kafka.client.base64.decoding.enabled"; + public static final String KAFKA_CREATE_CONSUMER_SCOPE_ENABLED = + "trace.kafka.create-consumer-scope.enabled"; public static final String JMS_PROPAGATION_DISABLED_TOPICS = "jms.propagation.disabled.topics"; public static final String JMS_PROPAGATION_DISABLED_QUEUES = "jms.propagation.disabled.queues"; diff --git a/dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java b/dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java index d8b3fef89ce..6d4ed7cffbe 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java @@ -1191,6 +1191,11 @@ public void closePrevious(boolean finishSpan) { scopeManager.closePrevious(finishSpan); } + @Override + public void closeLingeringIterationScope() { + scopeManager.closeLingeringIterationScope(); + } + @Override public AgentScope activateNext(AgentSpan span) { if (!InstrumenterConfig.get().isLegacyContextManagerEnabled()) { diff --git a/dd-trace-core/src/main/java/datadog/trace/core/scopemanager/ContinuableScopeManager.java b/dd-trace-core/src/main/java/datadog/trace/core/scopemanager/ContinuableScopeManager.java index 4cc16cd8a95..a478f7355f7 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/scopemanager/ContinuableScopeManager.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/scopemanager/ContinuableScopeManager.java @@ -267,6 +267,29 @@ public void closePrevious(final boolean finishSpan) { } } + /** + * Silently closes a lingering iteration scope (same cleanup as {@link #closePrevious(boolean)}'s + * iteration branch), but a quiet no-op with no telemetry when the top of stack is not an + * iteration scope, since it is called speculatively on every poll. + */ + public void closeLingeringIterationScope() { + ScopeStack scopeStack = scopeStack(); + + final ContinuableScope top = scopeStack.top; + if (top == null || top.source() != ITERATION) { + return; + } + if (iterationKeepAlive > 0) { // skip depth check because cancelling is cheap + cancelRootIterationScopeCleanup(scopeStack, top); + } + top.close(); + scopeStack.cleanup(); + AgentSpan span = top.span(); + if (span != null) { + span.finishWithEndToEnd(); + } + } + public AgentScope activateNext(final AgentSpan span) { ScopeStack scopeStack = scopeStack(); diff --git a/internal-api/src/main/java/datadog/trace/api/Config.java b/internal-api/src/main/java/datadog/trace/api/Config.java index c74f276cff7..b71b4851421 100644 --- a/internal-api/src/main/java/datadog/trace/api/Config.java +++ b/internal-api/src/main/java/datadog/trace/api/Config.java @@ -607,6 +607,7 @@ import static datadog.trace.api.config.TraceInstrumentationConfig.JMS_UNACKNOWLEDGED_MAX_AGE; import static datadog.trace.api.config.TraceInstrumentationConfig.KAFKA_CLIENT_BASE64_DECODING_ENABLED; import static datadog.trace.api.config.TraceInstrumentationConfig.KAFKA_CLIENT_PROPAGATION_DISABLED_TOPICS; +import static datadog.trace.api.config.TraceInstrumentationConfig.KAFKA_CREATE_CONSUMER_SCOPE_ENABLED; import static datadog.trace.api.config.TraceInstrumentationConfig.LOGS_INJECTION; import static datadog.trace.api.config.TraceInstrumentationConfig.LOGS_INJECTION_ENABLED; import static datadog.trace.api.config.TraceInstrumentationConfig.MESSAGE_BROKER_SPLIT_BY_DESTINATION; @@ -1269,6 +1270,7 @@ public static String getHostName() { private final boolean kafkaClientPropagationEnabled; private final Set kafkaClientPropagationDisabledTopics; private final boolean kafkaClientBase64DecodingEnabled; + private final boolean kafkaCreateConsumerScopeEnabled; private final boolean jmsPropagationEnabled; private final Set jmsPropagationDisabledTopics; @@ -2985,6 +2987,10 @@ PROFILING_DATADOG_PROFILER_ENABLED, isDatadogProfilerSafeInCurrentEnvironment()) tryMakeImmutableSet(configProvider.getList(KAFKA_CLIENT_PROPAGATION_DISABLED_TOPICS)); kafkaClientBase64DecodingEnabled = configProvider.getBoolean(KAFKA_CLIENT_BASE64_DECODING_ENABLED, false); + kafkaCreateConsumerScopeEnabled = + configProvider.getBoolean( + KAFKA_CREATE_CONSUMER_SCOPE_ENABLED, + ConfigDefaults.DEFAULT_TRACE_KAFKA_CREATE_CONSUMER_SCOPE_ENABLED); jmsPropagationEnabled = isPropagationEnabled(true, "jms"); jmsPropagationDisabledTopics = tryMakeImmutableSet(configProvider.getList(JMS_PROPAGATION_DISABLED_TOPICS)); @@ -4885,6 +4891,10 @@ public boolean isKafkaClientBase64DecodingEnabled() { return kafkaClientBase64DecodingEnabled; } + public boolean isKafkaCreateConsumerScopeEnabled() { + return kafkaCreateConsumerScopeEnabled; + } + public boolean isRabbitPropagationEnabled() { return rabbitPropagationEnabled; } @@ -6558,6 +6568,8 @@ public String toString() { + kafkaClientPropagationDisabledTopics + ", kafkaClientBase64DecodingEnabled=" + kafkaClientBase64DecodingEnabled + + ", kafkaCreateConsumerScopeEnabled=" + + kafkaCreateConsumerScopeEnabled + ", jmsPropagationEnabled=" + jmsPropagationEnabled + ", jmsPropagationDisabledTopics=" diff --git a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentTracer.java b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentTracer.java index a945dddbda6..ef312ca53b1 100644 --- a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentTracer.java +++ b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentTracer.java @@ -152,6 +152,14 @@ public static void closePrevious(final boolean finishSpan) { get().closePrevious(finishSpan); } + /** + * Silently closes a lingering iteration scope, if one is at the top of the stack. Safe to call + * speculatively (e.g. on every poll) since it is a quiet no-op when there is nothing to close. + */ + public static void closeLingeringIterationScope() { + get().closeLingeringIterationScope(); + } + /** * Activates a new iteration scope; closes automatically after a fixed period. * @@ -345,6 +353,8 @@ AgentSpan startSpan( void closePrevious(boolean finishSpan); + void closeLingeringIterationScope(); + AgentScope activateNext(AgentSpan span); AgentSpan activeSpan(); @@ -530,6 +540,9 @@ public void closeActive() {} @Override public void closePrevious(final boolean finishSpan) {} + @Override + public void closeLingeringIterationScope() {} + @Override public AgentScope activateNext(final AgentSpan span) { return NoopScope.INSTANCE; diff --git a/internal-api/src/test/groovy/datadog/trace/api/ConfigTest.groovy b/internal-api/src/test/groovy/datadog/trace/api/ConfigTest.groovy index 7db955e49ae..64089c01dc5 100644 --- a/internal-api/src/test/groovy/datadog/trace/api/ConfigTest.groovy +++ b/internal-api/src/test/groovy/datadog/trace/api/ConfigTest.groovy @@ -94,6 +94,7 @@ import static datadog.trace.api.config.TraceInstrumentationConfig.DB_CLIENT_HOST import static datadog.trace.api.config.TraceInstrumentationConfig.DB_CLIENT_HOST_SPLIT_BY_INSTANCE import static datadog.trace.api.config.TraceInstrumentationConfig.DB_CLIENT_HOST_SPLIT_BY_INSTANCE_TYPE_SUFFIX import static datadog.trace.api.config.TraceInstrumentationConfig.HTTP_CLIENT_HOST_SPLIT_BY_DOMAIN +import static datadog.trace.api.config.TraceInstrumentationConfig.KAFKA_CREATE_CONSUMER_SCOPE_ENABLED import static datadog.trace.api.config.TraceInstrumentationConfig.RUNTIME_CONTEXT_FIELD_INJECTION import static datadog.trace.api.config.TraceInstrumentationConfig.TRACE_ENABLED import static datadog.trace.api.config.TracerConfig.AGENT_HOST @@ -3478,4 +3479,17 @@ class ConfigTest extends DDSpecification { "1" | true "0" | false } + + def "kafka create consumer scope enabled defaults to false"() { + expect: + !Config.get().isKafkaCreateConsumerScopeEnabled() + } + + def "kafka create consumer scope enabled can be enabled via system property"() { + setup: + injectSysConfig(KAFKA_CREATE_CONSUMER_SCOPE_ENABLED, "true") + + expect: + Config.get().isKafkaCreateConsumerScopeEnabled() + } } diff --git a/metadata/supported-configurations.json b/metadata/supported-configurations.json index 6241dc3e6af..04aba7ec372 100644 --- a/metadata/supported-configurations.json +++ b/metadata/supported-configurations.json @@ -7729,6 +7729,14 @@ "aliases": ["DD_TRACE_INTEGRATION_KAFKA_CONNECT_ENABLED", "DD_INTEGRATION_KAFKA_CONNECT_ENABLED"] } ], + "DD_TRACE_KAFKA_CREATE_CONSUMER_SCOPE_ENABLED": [ + { + "version": "A", + "type": "boolean", + "default": "false", + "aliases": [] + } + ], "DD_TRACE_KAFKA_E2E_DURATION_ENABLED": [ { "version": "A",