From 601f77fa409b8c4e55b114a18dc765e0161c361a Mon Sep 17 00:00:00 2001 From: Johann Date: Fri, 16 Oct 2026 13:55:07 -0400 Subject: [PATCH 1/3] test: reproduce kafka_produce backlog with offset -1 on failed produce A produce rejected by the broker (record larger than the topic's max.message.bytes) or by the client (record larger than max.request.size) still records a kafka_produce backlog, with offset -1 and, on kafka-clients 3.8+, partition -1. The kafka-clients-0.11 test runs as a latestDep forked test so that its config does not leak into KafkaClientLatestTest through KafkaDecorator's statically captured settings. Co-Authored-By: Claude Opus 5.5 --- .../kafka/kafka-clients-0.11/build.gradle | 1 + ...roducerFailureDataStreamsForkedTest.groovy | 116 ++++++++++++++++++ ...roducerFailureDataStreamsForkedTest.groovy | 116 ++++++++++++++++++ 3 files changed, 233 insertions(+) create mode 100644 dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy create mode 100644 dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/build.gradle b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/build.gradle index 10c111c9aed..5db01a9b595 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/build.gradle +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/build.gradle @@ -22,6 +22,7 @@ muzzle { } addTestSuite('latestDepTest') +addForkedTestTask('latestDepTest') addTestSuite('iastLatestDepTest3') dependencies { diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy new file mode 100644 index 00000000000..faa1b44ca83 --- /dev/null +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy @@ -0,0 +1,116 @@ +import datadog.trace.agent.test.naming.VersionedNamingTestBase +import datadog.trace.api.Config +import org.apache.kafka.clients.admin.NewTopic +import org.apache.kafka.clients.producer.KafkaProducer +import org.apache.kafka.clients.producer.ProducerConfig +import org.apache.kafka.clients.producer.ProducerRecord +import org.apache.kafka.clients.producer.RecordMetadata +import org.apache.kafka.common.serialization.StringSerializer +import org.springframework.kafka.test.EmbeddedKafkaBroker +import org.springframework.kafka.test.utils.KafkaTestUtils +import spock.util.concurrent.PollingConditions + +import java.util.concurrent.ExecutionException + +class KafkaProducerFailureDataStreamsForkedTest extends VersionedNamingTestBase { + static final String OK_TOPIC = "ok.topic" + static final String REJECTING_TOPIC = "rejecting.topic" + + EmbeddedKafkaBroker embeddedKafka + + def setup() { + embeddedKafka = new EmbeddedKafkaBroker(1, true, OK_TOPIC) + embeddedKafka.afterPropertiesSet() + embeddedKafka.addTopics(new NewTopic(REJECTING_TOPIC, 1, (short) 1).configs(["max.message.bytes": "512"])) + } + + def cleanup() { + embeddedKafka?.destroy() + } + + @Override + int version() { + 0 + } + + @Override + String operation() { + null + } + + @Override + String service() { + Config.get().getServiceName() + } + + @Override + protected boolean isDataStreamsEnabled() { + true + } + + def "produce rejected by the broker does not report a kafka_produce backlog"() { + setup: + def producer = newProducer([:]) + + when: + producer.send(new ProducerRecord(REJECTING_TOPIC, "x" * 2048)).get() + + then: + thrown(ExecutionException) + + when: + RecordMetadata ok = producer.send(new ProducerRecord(OK_TOPIC, "hello")).get() + waitForProduceBacklog(OK_TOPIC) + + then: + produceBacklogs() == ["topic:$OK_TOPIC partition:${ok.partition()} offset:${ok.offset()}".toString()] + + cleanup: + producer?.close() + } + + def "produce rejected by the client does not report a kafka_produce backlog"() { + setup: + def failingProducer = newProducer([(ProducerConfig.MAX_REQUEST_SIZE_CONFIG): 10]) + def producer = newProducer([:]) + + when: + failingProducer.send(new ProducerRecord(REJECTING_TOPIC, "hello")).get() + + then: + thrown(ExecutionException) + + when: + RecordMetadata ok = producer.send(new ProducerRecord(OK_TOPIC, "hello")).get() + waitForProduceBacklog(OK_TOPIC) + + then: + produceBacklogs() == ["topic:$OK_TOPIC partition:${ok.partition()} offset:${ok.offset()}".toString()] + + cleanup: + failingProducer?.close() + producer?.close() + } + + KafkaProducer newProducer(Map overrides) { + def props = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + props.put(ProducerConfig.RETRIES_CONFIG, 0) + props.putAll(overrides) + new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()) + } + + List produceBacklogs() { + TEST_DATA_STREAMS_WRITER.payloads + .collectMany { it.backlogs ?: [] } + .findAll { it.key.hasAllTags("type:kafka_produce") } + .collect { "${it.key.topic} ${it.key.partition} offset:${it.value}".toString() } + .sort() + } + + // Backlogs recorded before this one sit in the same or an earlier bucket, so they are flushed by then. + void waitForProduceBacklog(String topic) { + new PollingConditions(timeout: 10).eventually { + assert produceBacklogs().any { it.startsWith("topic:$topic ") } + } + } +} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy new file mode 100644 index 00000000000..0828f1f89b7 --- /dev/null +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy @@ -0,0 +1,116 @@ +import datadog.trace.agent.test.naming.VersionedNamingTestBase +import org.apache.kafka.clients.admin.NewTopic +import org.apache.kafka.clients.producer.KafkaProducer +import org.apache.kafka.clients.producer.ProducerConfig +import org.apache.kafka.clients.producer.ProducerRecord +import org.apache.kafka.clients.producer.RecordMetadata +import org.apache.kafka.common.serialization.StringSerializer +import org.springframework.kafka.test.EmbeddedKafkaBroker +import org.springframework.kafka.test.EmbeddedKafkaKraftBroker +import org.springframework.kafka.test.utils.KafkaTestUtils +import spock.util.concurrent.PollingConditions + +import java.util.concurrent.ExecutionException + +class KafkaProducerFailureDataStreamsForkedTest extends VersionedNamingTestBase { + static final String OK_TOPIC = "ok.topic" + static final String REJECTING_TOPIC = "rejecting.topic" + + EmbeddedKafkaBroker embeddedKafka + + def setup() { + embeddedKafka = new EmbeddedKafkaKraftBroker(1, 1, OK_TOPIC) + embeddedKafka.afterPropertiesSet() + embeddedKafka.addTopics(new NewTopic(REJECTING_TOPIC, 1, (short) 1).configs(["max.message.bytes": "512"])) + } + + def cleanup() { + embeddedKafka?.destroy() + } + + @Override + int version() { + 0 + } + + @Override + String operation() { + null + } + + @Override + String service() { + "KafkaProducerFailureDataStreamsForkedTest" + } + + @Override + protected boolean isDataStreamsEnabled() { + true + } + + def "produce rejected by the broker does not report a kafka_produce backlog"() { + setup: + def producer = newProducer([:]) + + when: + producer.send(new ProducerRecord(REJECTING_TOPIC, "x" * 2048)).get() + + then: + thrown(ExecutionException) + + when: + RecordMetadata ok = producer.send(new ProducerRecord(OK_TOPIC, "hello")).get() + waitForProduceBacklog(OK_TOPIC) + + then: + produceBacklogs() == ["topic:$OK_TOPIC partition:${ok.partition()} offset:${ok.offset()}".toString()] + + cleanup: + producer?.close() + } + + def "produce rejected by the client does not report a kafka_produce backlog"() { + setup: + def failingProducer = newProducer([(ProducerConfig.MAX_REQUEST_SIZE_CONFIG): 10]) + def producer = newProducer([:]) + + when: + failingProducer.send(new ProducerRecord(REJECTING_TOPIC, "hello")).get() + + then: + thrown(ExecutionException) + + when: + RecordMetadata ok = producer.send(new ProducerRecord(OK_TOPIC, "hello")).get() + waitForProduceBacklog(OK_TOPIC) + + then: + produceBacklogs() == ["topic:$OK_TOPIC partition:${ok.partition()} offset:${ok.offset()}".toString()] + + cleanup: + failingProducer?.close() + producer?.close() + } + + KafkaProducer newProducer(Map overrides) { + def props = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) + props.put(ProducerConfig.RETRIES_CONFIG, 0) + props.putAll(overrides) + new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()) + } + + List produceBacklogs() { + TEST_DATA_STREAMS_WRITER.payloads + .collectMany { it.backlogs ?: [] } + .findAll { it.key.hasAllTags("type:kafka_produce") } + .collect { "${it.key.topic} ${it.key.partition} offset:${it.value}".toString() } + .sort() + } + + // Backlogs recorded before this one sit in the same or an earlier bucket, so they are flushed by then. + void waitForProduceBacklog(String topic) { + new PollingConditions(timeout: 10).eventually { + assert produceBacklogs().any { it.startsWith("topic:$topic ") } + } + } +} From 8a42c465d5e96e58fec09f60a9366b0a1faad09b Mon Sep 17 00:00:00 2001 From: Johann Date: Fri, 16 Oct 2026 14:08:31 -0400 Subject: [PATCH 2/3] fix: skip kafka_produce backlog when the produce failed KafkaProducerCallback only skipped DSM backlog tracking when metadata was null. Since kafka-clients 2.0 a failed send passes placeholder metadata with offset -1 (and partition -1 when no partition was assigned yet), so failures were recorded as produce offsets of -1 and could create a bogus partition:-1 series. Skip tracking when the send reported an exception or carries no partition, matching the Python, .NET and Go tracers. Co-Authored-By: Claude Opus 5.5 --- .../instrumentation/kafka_clients/KafkaProducerCallback.java | 2 +- .../instrumentation/kafka_clients38/KafkaProducerCallback.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaProducerCallback.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaProducerCallback.java index c536e147e77..e2cd0f8d3a9 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaProducerCallback.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaProducerCallback.java @@ -48,7 +48,7 @@ public void onCompletion(final RecordMetadata metadata, final Exception exceptio callback.onCompletion(metadata, exception); } } - if (metadata == null) { + if (metadata == null || exception != null || metadata.partition() < 0) { return; } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaProducerCallback.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaProducerCallback.java index dd7bb52989c..c5d4e8b95f9 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaProducerCallback.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaProducerCallback.java @@ -47,7 +47,7 @@ public void onCompletion(final RecordMetadata metadata, final Exception exceptio callback.onCompletion(metadata, exception); } } - if (metadata == null) { + if (metadata == null || exception != null || metadata.partition() < 0) { return; } DataStreamsTags tags = From 3585db63bf62662b52d940b496279c21a30fb044 Mon Sep 17 00:00:00 2001 From: Johann Date: Fri, 16 Oct 2026 17:21:12 -0400 Subject: [PATCH 3/3] test: rewrite kafka produce failure DSM tests in Java The repo blocks new Groovy files, so KafkaProducerFailureDataStreamsForkedTest moves from Spock to JUnit for both kafka-clients-0.11 (latestDepTest) and kafka-clients-3.8. AbstractInstrumentationTest has no data streams support, so the test installs a tracer whose DefaultDataStreamsMonitoring records flushed payloads. Assertions are unchanged: a failed produce must not report a kafka_produce backlog. Co-Authored-By: Claude Opus 5.5 --- ...roducerFailureDataStreamsForkedTest.groovy | 116 -------- ...aProducerFailureDataStreamsForkedTest.java | 262 +++++++++++++++++ ...roducerFailureDataStreamsForkedTest.groovy | 116 -------- ...aProducerFailureDataStreamsForkedTest.java | 263 ++++++++++++++++++ 4 files changed, 525 insertions(+), 232 deletions(-) delete mode 100644 dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy create mode 100644 dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java delete mode 100644 dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy create mode 100644 dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy deleted file mode 100644 index faa1b44ca83..00000000000 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy +++ /dev/null @@ -1,116 +0,0 @@ -import datadog.trace.agent.test.naming.VersionedNamingTestBase -import datadog.trace.api.Config -import org.apache.kafka.clients.admin.NewTopic -import org.apache.kafka.clients.producer.KafkaProducer -import org.apache.kafka.clients.producer.ProducerConfig -import org.apache.kafka.clients.producer.ProducerRecord -import org.apache.kafka.clients.producer.RecordMetadata -import org.apache.kafka.common.serialization.StringSerializer -import org.springframework.kafka.test.EmbeddedKafkaBroker -import org.springframework.kafka.test.utils.KafkaTestUtils -import spock.util.concurrent.PollingConditions - -import java.util.concurrent.ExecutionException - -class KafkaProducerFailureDataStreamsForkedTest extends VersionedNamingTestBase { - static final String OK_TOPIC = "ok.topic" - static final String REJECTING_TOPIC = "rejecting.topic" - - EmbeddedKafkaBroker embeddedKafka - - def setup() { - embeddedKafka = new EmbeddedKafkaBroker(1, true, OK_TOPIC) - embeddedKafka.afterPropertiesSet() - embeddedKafka.addTopics(new NewTopic(REJECTING_TOPIC, 1, (short) 1).configs(["max.message.bytes": "512"])) - } - - def cleanup() { - embeddedKafka?.destroy() - } - - @Override - int version() { - 0 - } - - @Override - String operation() { - null - } - - @Override - String service() { - Config.get().getServiceName() - } - - @Override - protected boolean isDataStreamsEnabled() { - true - } - - def "produce rejected by the broker does not report a kafka_produce backlog"() { - setup: - def producer = newProducer([:]) - - when: - producer.send(new ProducerRecord(REJECTING_TOPIC, "x" * 2048)).get() - - then: - thrown(ExecutionException) - - when: - RecordMetadata ok = producer.send(new ProducerRecord(OK_TOPIC, "hello")).get() - waitForProduceBacklog(OK_TOPIC) - - then: - produceBacklogs() == ["topic:$OK_TOPIC partition:${ok.partition()} offset:${ok.offset()}".toString()] - - cleanup: - producer?.close() - } - - def "produce rejected by the client does not report a kafka_produce backlog"() { - setup: - def failingProducer = newProducer([(ProducerConfig.MAX_REQUEST_SIZE_CONFIG): 10]) - def producer = newProducer([:]) - - when: - failingProducer.send(new ProducerRecord(REJECTING_TOPIC, "hello")).get() - - then: - thrown(ExecutionException) - - when: - RecordMetadata ok = producer.send(new ProducerRecord(OK_TOPIC, "hello")).get() - waitForProduceBacklog(OK_TOPIC) - - then: - produceBacklogs() == ["topic:$OK_TOPIC partition:${ok.partition()} offset:${ok.offset()}".toString()] - - cleanup: - failingProducer?.close() - producer?.close() - } - - KafkaProducer newProducer(Map overrides) { - def props = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) - props.put(ProducerConfig.RETRIES_CONFIG, 0) - props.putAll(overrides) - new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()) - } - - List produceBacklogs() { - TEST_DATA_STREAMS_WRITER.payloads - .collectMany { it.backlogs ?: [] } - .findAll { it.key.hasAllTags("type:kafka_produce") } - .collect { "${it.key.topic} ${it.key.partition} offset:${it.value}".toString() } - .sort() - } - - // Backlogs recorded before this one sit in the same or an earlier bucket, so they are flushed by then. - void waitForProduceBacklog(String topic) { - new PollingConditions(timeout: 10).eventually { - assert produceBacklogs().any { it.startsWith("topic:$topic ") } - } - } -} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java new file mode 100644 index 00000000000..2ab9e0b7763 --- /dev/null +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java @@ -0,0 +1,262 @@ +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.agent.test.datastreams.MockFeaturesDiscovery; +import datadog.trace.agent.tooling.TracerInstaller; +import datadog.trace.api.IdGenerationStrategy; +import datadog.trace.api.Pair; +import datadog.trace.api.TraceConfig; +import datadog.trace.api.datastreams.DataStreamsTags; +import datadog.trace.api.datastreams.DataStreamsTransactionExtractor; +import datadog.trace.api.sampling.SamplingRule.SpanSamplingRule; +import datadog.trace.api.sampling.SamplingRule.TraceSamplingRule; +import datadog.trace.api.time.SystemTimeSource; +import datadog.trace.common.metrics.EventListener; +import datadog.trace.common.metrics.Sink; +import datadog.trace.core.CoreTracer; +import datadog.trace.core.datastreams.DefaultDataStreamsMonitoring; +import datadog.trace.core.datastreams.StatsBucket; +import datadog.trace.test.junit.utils.config.WithConfig; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.utils.KafkaTestUtils; + +@WithConfig(key = "data.streams.enabled", value = "true") +class KafkaProducerFailureDataStreamsForkedTest extends AbstractInstrumentationTest { + private static final String OK_TOPIC = "ok.topic"; + private static final String REJECTING_TOPIC = "rejecting.topic"; + + private static final List dataStreamsPayloads = new CopyOnWriteArrayList<>(); + + private EmbeddedKafkaBroker embeddedKafka; + + // AbstractInstrumentationTest has no data streams support, so swap in a tracer whose data + // streams monitoring records the payloads it flushes. + @BeforeAll + static void installDataStreamsTracer() { + DefaultDataStreamsMonitoring dataStreamsMonitoring = + new DefaultDataStreamsMonitoring( + new NoopSink(), + new MockFeaturesDiscovery(true), + SystemTimeSource.INSTANCE, + new DataStreamsEnabledTraceConfig(), + (data, serviceNameOverride) -> dataStreamsPayloads.addAll(data), + TimeUnit.MILLISECONDS.toNanos(50)); + CoreTracer coreTracer = + CoreTracer.builder() + .writer(writer) + .idGenerationStrategy(IdGenerationStrategy.fromName("SEQUENTIAL")) + .strictTraceWrites(true) + .dataStreamsMonitoring(dataStreamsMonitoring) + .build(); + TracerInstaller.forceInstallGlobalTracer(coreTracer); + tracer = coreTracer; + } + + @BeforeEach + void startBroker() { + dataStreamsPayloads.clear(); + embeddedKafka = new EmbeddedKafkaBroker(1, true, OK_TOPIC); + embeddedKafka.afterPropertiesSet(); + embeddedKafka.addTopics( + new NewTopic(REJECTING_TOPIC, 1, (short) 1) + .configs(Collections.singletonMap("max.message.bytes", "512"))); + } + + @AfterEach + void stopBroker() { + if (embeddedKafka != null) { + embeddedKafka.destroy(); + } + } + + @Test + void produceRejectedByTheBrokerDoesNotReportAKafkaProduceBacklog() throws Exception { + try (KafkaProducer producer = newProducer(Collections.emptyMap())) { + assertThrows( + ExecutionException.class, + () -> producer.send(new ProducerRecord<>(REJECTING_TOPIC, repeat('x', 2048))).get()); + + RecordMetadata ok = producer.send(new ProducerRecord<>(OK_TOPIC, "hello")).get(); + waitForProduceBacklog(OK_TOPIC); + + assertEquals( + Collections.singletonList( + "topic:" + OK_TOPIC + " partition:" + ok.partition() + " offset:" + ok.offset()), + produceBacklogs()); + } + } + + @Test + void produceRejectedByTheClientDoesNotReportAKafkaProduceBacklog() throws Exception { + try (KafkaProducer failingProducer = + newProducer(Collections.singletonMap(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10)); + KafkaProducer producer = newProducer(Collections.emptyMap())) { + assertThrows( + ExecutionException.class, + () -> failingProducer.send(new ProducerRecord<>(REJECTING_TOPIC, "hello")).get()); + + RecordMetadata ok = producer.send(new ProducerRecord<>(OK_TOPIC, "hello")).get(); + waitForProduceBacklog(OK_TOPIC); + + assertEquals( + Collections.singletonList( + "topic:" + OK_TOPIC + " partition:" + ok.partition() + " offset:" + ok.offset()), + produceBacklogs()); + } + } + + private KafkaProducer newProducer(Map overrides) { + Map props = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()); + props.put(ProducerConfig.RETRIES_CONFIG, 0); + props.putAll(overrides); + return new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()); + } + + private static List produceBacklogs() { + List backlogs = new ArrayList<>(); + for (StatsBucket bucket : dataStreamsPayloads) { + Collection> bucketBacklogs = bucket.getBacklogs(); + if (bucketBacklogs == null) { + continue; + } + for (Map.Entry backlog : bucketBacklogs) { + DataStreamsTags tags = backlog.getKey(); + if (tags.hasAllTags(new String[] {"type:kafka_produce"})) { + backlogs.add( + tags.getTopic() + " " + tags.getPartition() + " offset:" + backlog.getValue()); + } + } + } + Collections.sort(backlogs); + return backlogs; + } + + // Backlogs recorded before this one sit in the same or an earlier bucket, so they are flushed by + // then. + private static void waitForProduceBacklog(String topic) throws InterruptedException { + long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(10); + while (System.currentTimeMillis() < deadline) { + for (String backlog : produceBacklogs()) { + if (backlog.startsWith("topic:" + topic + " ")) { + return; + } + } + Thread.sleep(20); + } + throw new AssertionError("No kafka_produce backlog reported for " + topic); + } + + private static String repeat(char c, int count) { + char[] chars = new char[count]; + Arrays.fill(chars, c); + return new String(chars); + } + + private static final class NoopSink implements Sink { + @Override + public void accept(int messageCount, ByteBuffer buffer) {} + + @Override + public void register(EventListener listener) {} + } + + private static final class DataStreamsEnabledTraceConfig + implements TraceConfig, Supplier { + @Override + public TraceConfig get() { + return this; + } + + @Override + public boolean isTraceEnabled() { + return true; + } + + @Override + public boolean isRuntimeMetricsEnabled() { + return true; + } + + @Override + public boolean isLogsInjectionEnabled() { + return true; + } + + @Override + public boolean isDataStreamsEnabled() { + return true; + } + + @Override + public Map getServiceMapping() { + return null; + } + + @Override + public Map getRequestHeaderTags() { + return null; + } + + @Override + public Map getResponseHeaderTags() { + return null; + } + + @Override + public Map getBaggageMapping() { + return null; + } + + @Override + public Double getTraceSampleRate() { + return null; + } + + @Override + public Map getTracingTags() { + return null; + } + + @Override + public Pair getPreferredServiceNameAndSource() { + return null; + } + + @Override + public List getSpanSamplingRules() { + return null; + } + + @Override + public List getTraceSamplingRules() { + return null; + } + + @Override + public List getDataStreamsTransactionExtractors() { + return null; + } + } +} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy deleted file mode 100644 index 0828f1f89b7..00000000000 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/groovy/KafkaProducerFailureDataStreamsForkedTest.groovy +++ /dev/null @@ -1,116 +0,0 @@ -import datadog.trace.agent.test.naming.VersionedNamingTestBase -import org.apache.kafka.clients.admin.NewTopic -import org.apache.kafka.clients.producer.KafkaProducer -import org.apache.kafka.clients.producer.ProducerConfig -import org.apache.kafka.clients.producer.ProducerRecord -import org.apache.kafka.clients.producer.RecordMetadata -import org.apache.kafka.common.serialization.StringSerializer -import org.springframework.kafka.test.EmbeddedKafkaBroker -import org.springframework.kafka.test.EmbeddedKafkaKraftBroker -import org.springframework.kafka.test.utils.KafkaTestUtils -import spock.util.concurrent.PollingConditions - -import java.util.concurrent.ExecutionException - -class KafkaProducerFailureDataStreamsForkedTest extends VersionedNamingTestBase { - static final String OK_TOPIC = "ok.topic" - static final String REJECTING_TOPIC = "rejecting.topic" - - EmbeddedKafkaBroker embeddedKafka - - def setup() { - embeddedKafka = new EmbeddedKafkaKraftBroker(1, 1, OK_TOPIC) - embeddedKafka.afterPropertiesSet() - embeddedKafka.addTopics(new NewTopic(REJECTING_TOPIC, 1, (short) 1).configs(["max.message.bytes": "512"])) - } - - def cleanup() { - embeddedKafka?.destroy() - } - - @Override - int version() { - 0 - } - - @Override - String operation() { - null - } - - @Override - String service() { - "KafkaProducerFailureDataStreamsForkedTest" - } - - @Override - protected boolean isDataStreamsEnabled() { - true - } - - def "produce rejected by the broker does not report a kafka_produce backlog"() { - setup: - def producer = newProducer([:]) - - when: - producer.send(new ProducerRecord(REJECTING_TOPIC, "x" * 2048)).get() - - then: - thrown(ExecutionException) - - when: - RecordMetadata ok = producer.send(new ProducerRecord(OK_TOPIC, "hello")).get() - waitForProduceBacklog(OK_TOPIC) - - then: - produceBacklogs() == ["topic:$OK_TOPIC partition:${ok.partition()} offset:${ok.offset()}".toString()] - - cleanup: - producer?.close() - } - - def "produce rejected by the client does not report a kafka_produce backlog"() { - setup: - def failingProducer = newProducer([(ProducerConfig.MAX_REQUEST_SIZE_CONFIG): 10]) - def producer = newProducer([:]) - - when: - failingProducer.send(new ProducerRecord(REJECTING_TOPIC, "hello")).get() - - then: - thrown(ExecutionException) - - when: - RecordMetadata ok = producer.send(new ProducerRecord(OK_TOPIC, "hello")).get() - waitForProduceBacklog(OK_TOPIC) - - then: - produceBacklogs() == ["topic:$OK_TOPIC partition:${ok.partition()} offset:${ok.offset()}".toString()] - - cleanup: - failingProducer?.close() - producer?.close() - } - - KafkaProducer newProducer(Map overrides) { - def props = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()) - props.put(ProducerConfig.RETRIES_CONFIG, 0) - props.putAll(overrides) - new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()) - } - - List produceBacklogs() { - TEST_DATA_STREAMS_WRITER.payloads - .collectMany { it.backlogs ?: [] } - .findAll { it.key.hasAllTags("type:kafka_produce") } - .collect { "${it.key.topic} ${it.key.partition} offset:${it.value}".toString() } - .sort() - } - - // Backlogs recorded before this one sit in the same or an earlier bucket, so they are flushed by then. - void waitForProduceBacklog(String topic) { - new PollingConditions(timeout: 10).eventually { - assert produceBacklogs().any { it.startsWith("topic:$topic ") } - } - } -} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java new file mode 100644 index 00000000000..b56ed2328fa --- /dev/null +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java @@ -0,0 +1,263 @@ +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.agent.test.datastreams.MockFeaturesDiscovery; +import datadog.trace.agent.tooling.TracerInstaller; +import datadog.trace.api.IdGenerationStrategy; +import datadog.trace.api.Pair; +import datadog.trace.api.TraceConfig; +import datadog.trace.api.datastreams.DataStreamsTags; +import datadog.trace.api.datastreams.DataStreamsTransactionExtractor; +import datadog.trace.api.sampling.SamplingRule.SpanSamplingRule; +import datadog.trace.api.sampling.SamplingRule.TraceSamplingRule; +import datadog.trace.api.time.SystemTimeSource; +import datadog.trace.common.metrics.EventListener; +import datadog.trace.common.metrics.Sink; +import datadog.trace.core.CoreTracer; +import datadog.trace.core.datastreams.DefaultDataStreamsMonitoring; +import datadog.trace.core.datastreams.StatsBucket; +import datadog.trace.test.junit.utils.config.WithConfig; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.EmbeddedKafkaKraftBroker; +import org.springframework.kafka.test.utils.KafkaTestUtils; + +@WithConfig(key = "data.streams.enabled", value = "true") +class KafkaProducerFailureDataStreamsForkedTest extends AbstractInstrumentationTest { + private static final String OK_TOPIC = "ok.topic"; + private static final String REJECTING_TOPIC = "rejecting.topic"; + + private static final List dataStreamsPayloads = new CopyOnWriteArrayList<>(); + + private EmbeddedKafkaBroker embeddedKafka; + + // AbstractInstrumentationTest has no data streams support, so swap in a tracer whose data + // streams monitoring records the payloads it flushes. + @BeforeAll + static void installDataStreamsTracer() { + DefaultDataStreamsMonitoring dataStreamsMonitoring = + new DefaultDataStreamsMonitoring( + new NoopSink(), + new MockFeaturesDiscovery(true), + SystemTimeSource.INSTANCE, + new DataStreamsEnabledTraceConfig(), + (data, serviceNameOverride) -> dataStreamsPayloads.addAll(data), + TimeUnit.MILLISECONDS.toNanos(50)); + CoreTracer coreTracer = + CoreTracer.builder() + .writer(writer) + .idGenerationStrategy(IdGenerationStrategy.fromName("SEQUENTIAL")) + .strictTraceWrites(true) + .dataStreamsMonitoring(dataStreamsMonitoring) + .build(); + TracerInstaller.forceInstallGlobalTracer(coreTracer); + tracer = coreTracer; + } + + @BeforeEach + void startBroker() { + dataStreamsPayloads.clear(); + embeddedKafka = new EmbeddedKafkaKraftBroker(1, 1, OK_TOPIC); + embeddedKafka.afterPropertiesSet(); + embeddedKafka.addTopics( + new NewTopic(REJECTING_TOPIC, 1, (short) 1) + .configs(Collections.singletonMap("max.message.bytes", "512"))); + } + + @AfterEach + void stopBroker() { + if (embeddedKafka != null) { + embeddedKafka.destroy(); + } + } + + @Test + void produceRejectedByTheBrokerDoesNotReportAKafkaProduceBacklog() throws Exception { + try (KafkaProducer producer = newProducer(Collections.emptyMap())) { + assertThrows( + ExecutionException.class, + () -> producer.send(new ProducerRecord<>(REJECTING_TOPIC, repeat('x', 2048))).get()); + + RecordMetadata ok = producer.send(new ProducerRecord<>(OK_TOPIC, "hello")).get(); + waitForProduceBacklog(OK_TOPIC); + + assertEquals( + Collections.singletonList( + "topic:" + OK_TOPIC + " partition:" + ok.partition() + " offset:" + ok.offset()), + produceBacklogs()); + } + } + + @Test + void produceRejectedByTheClientDoesNotReportAKafkaProduceBacklog() throws Exception { + try (KafkaProducer failingProducer = + newProducer(Collections.singletonMap(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10)); + KafkaProducer producer = newProducer(Collections.emptyMap())) { + assertThrows( + ExecutionException.class, + () -> failingProducer.send(new ProducerRecord<>(REJECTING_TOPIC, "hello")).get()); + + RecordMetadata ok = producer.send(new ProducerRecord<>(OK_TOPIC, "hello")).get(); + waitForProduceBacklog(OK_TOPIC); + + assertEquals( + Collections.singletonList( + "topic:" + OK_TOPIC + " partition:" + ok.partition() + " offset:" + ok.offset()), + produceBacklogs()); + } + } + + private KafkaProducer newProducer(Map overrides) { + Map props = KafkaTestUtils.producerProps(embeddedKafka.getBrokersAsString()); + props.put(ProducerConfig.RETRIES_CONFIG, 0); + props.putAll(overrides); + return new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()); + } + + private static List produceBacklogs() { + List backlogs = new ArrayList<>(); + for (StatsBucket bucket : dataStreamsPayloads) { + Collection> bucketBacklogs = bucket.getBacklogs(); + if (bucketBacklogs == null) { + continue; + } + for (Map.Entry backlog : bucketBacklogs) { + DataStreamsTags tags = backlog.getKey(); + if (tags.hasAllTags(new String[] {"type:kafka_produce"})) { + backlogs.add( + tags.getTopic() + " " + tags.getPartition() + " offset:" + backlog.getValue()); + } + } + } + Collections.sort(backlogs); + return backlogs; + } + + // Backlogs recorded before this one sit in the same or an earlier bucket, so they are flushed by + // then. + private static void waitForProduceBacklog(String topic) throws InterruptedException { + long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(10); + while (System.currentTimeMillis() < deadline) { + for (String backlog : produceBacklogs()) { + if (backlog.startsWith("topic:" + topic + " ")) { + return; + } + } + Thread.sleep(20); + } + throw new AssertionError("No kafka_produce backlog reported for " + topic); + } + + private static String repeat(char c, int count) { + char[] chars = new char[count]; + Arrays.fill(chars, c); + return new String(chars); + } + + private static final class NoopSink implements Sink { + @Override + public void accept(int messageCount, ByteBuffer buffer) {} + + @Override + public void register(EventListener listener) {} + } + + private static final class DataStreamsEnabledTraceConfig + implements TraceConfig, Supplier { + @Override + public TraceConfig get() { + return this; + } + + @Override + public boolean isTraceEnabled() { + return true; + } + + @Override + public boolean isRuntimeMetricsEnabled() { + return true; + } + + @Override + public boolean isLogsInjectionEnabled() { + return true; + } + + @Override + public boolean isDataStreamsEnabled() { + return true; + } + + @Override + public Map getServiceMapping() { + return null; + } + + @Override + public Map getRequestHeaderTags() { + return null; + } + + @Override + public Map getResponseHeaderTags() { + return null; + } + + @Override + public Map getBaggageMapping() { + return null; + } + + @Override + public Double getTraceSampleRate() { + return null; + } + + @Override + public Map getTracingTags() { + return null; + } + + @Override + public Pair getPreferredServiceNameAndSource() { + return null; + } + + @Override + public List getSpanSamplingRules() { + return null; + } + + @Override + public List getTraceSamplingRules() { + return null; + } + + @Override + public List getDataStreamsTransactionExtractors() { + return null; + } + } +}