Skip DSM produce backlog for failed Kafka sends

[johannbotha]

What's wrong

This affects Java services with Data Streams Monitoring enabled (DD_DATA_STREAMS_ENABLED=true) that produce to Kafka with kafka-clients 1.1 or newer. dd-trace-java has behaved this way since at least v1.8.0.

Sometimes a producer's sends to a partition fail for more than a minute and no other send to that partition succeeds in the meantime. Causes include an ACL or auth error, delivery timeouts, records that are too large, or a broker outage. The tracer reports produce offset -1 for each failed send. If consumers keep committing during that time, DSM shows their consumer lag as 0 messages and 0 seconds until the producer recovers, even when the consumer is far behind. Monitors on data_streams.kafka.lag_messages or data_streams.kafka.lag_seconds can resolve or stay quiet during exactly the incident they should catch. After the producer recovers, lag in messages is correct again. Lag in seconds stays far too low for messages produced before the failure.

The consumers can be in any language, except Go consumers on confluent-kafka-go. Those report the partition's high watermark, which DSM uses for their lag instead of produce offsets.

Shorter failures, and failures mixed with successful sends, do not affect lag.

Impact

• Who: Java services on dd-trace-java v1.8.0 through v1.67.1 with Data Streams Monitoring enabled and a Kafka producer on kafka-clients 2.0+ (including Kafka Streams and Connect producers), whenever a send fails (broker unreachable, delivery timeout, auth error, unknown topic, record too large, max.block.ms metadata timeout).
• What they see: a phantom partition:-1 series at value -1 on data_streams.kafka.producer_offset while failures last. On a real partition, a failed send arriving more than a minute after the last accepted offset wipes that partition's offset history, so lag_messages clamps to 0 until the next successful ack and lag_seconds is understated after recovery (lag effect inferred from code).
• Since when: introduced in 905f42323b (first release v1.8.0). The bogus partition:-1 series appear in production DSM data, though no customer reports were found. A similar gap in dd-trace-dotnet ProduceAsync (partition -1, offset -1001) is not covered by this fix.

Before / After

The numbers come from a simulation of the DSM lag writer's offset logic. A consumer is stuck at offset 500 while the producer had reached offset 1000. The producer's sends then fail for 100 s, and it recovers at t=210 s.

What the customer looks at	Before	After
Consumer lag after the producer has failed for over a minute	0 messages, 0 s	501 messages, and lag in seconds keeps rising
Lag monitor on that consumer during the failure	Can resolve or never fire	Sees the real lag
Lag in seconds right after the producer recovers	5 s (true lag: 2 m 40 s)	2 m 40 s
data_streams.kafka.producer_offset in Metrics Explorer or notebooks	Drops to -1 after a minute of failures. Failed sends with no partition assigned yet add a partition:-1 series. DSM pages do not query this metric.	No -1 values, no partition:-1 series
Successful sends	Real partition and offset	Unchanged

Cause: from kafka-clients 1.1 on, a failed send passes placeholder metadata (offset -1, and on newer clients partition -1 if no partition was assigned) to the send callback. The tracer only skipped null metadata, so it reported -1 as the produce offset. The backend treats a lower produce offset that persists for over a minute as a topic reset. It discards that partition's offset history and computes lag against -1.

What Does This Do

Skips DSM kafka_produce backlog tracking in KafkaProducerCallback when the send failed. The guard if (metadata == null) becomes if (metadata == null || exception != null || metadata.partition() < 0) in both kafka-clients-0.11 and kafka-clients-3.8. Successful acks are unchanged. The user callback and span handling still run before the return.

Motivation

Failed sends should not count as produce progress. Today, when consumers keep committing, a producer-side failure longer than a minute hides their consumer lag. The Python, .NET and Go tracers already skip errored produces.

Test

• KafkaProducerFailureDataStreamsForkedTest (JUnit) in kafka-clients-0.11/src/latestDepTest/java/ and kafka-clients-3.8/src/test/java/. Each has two tests. Both produce one good record to ok.topic. One also sends a record that the broker rejects (topic with max.message.bytes=512), and the other one that the client rejects (max.request.size=10). Both then assert that the only kafka_produce backlog is the one for ok.topic. With the fix reverted they fail with, for example:
  expected: <[topic:ok.topic partition:0 offset:0]> but was: <[topic:ok.topic partition:0 offset:0, topic:rejecting.topic partition:0 offset:-1]>
  With kafka-clients 3.8.0, the client-side rejection shows topic:rejecting.topic partition:-1 offset:-1. They pass with this change.
• kafka-clients-0.11/build.gradle adds addForkedTestTask('latestDepTest') so the new class runs in its own JVM. KafkaDecorator captures some config in static fields at class load, so sharing a JVM would let this class's config leak into KafkaClientLatestTest, depending on class order.
• CI's check and instrumentation test jobs (test_inst, test_inst_latest) pass on the branch head.

How to verify manually

1. On the branch head, put back the old guard while keeping the tests: git revert --no-commit 8a42c465d5. Then run:
   ./gradlew :dd-java-agent:instrumentation:kafka:kafka-clients-0.11:latestDepTestForkedTest :dd-java-agent:instrumentation:kafka:kafka-clients-3.8:forkedTest :dd-java-agent:instrumentation:kafka:kafka-clients-3.8:latestDepForkedTest --tests KafkaProducerFailureDataStreamsForkedTest
   The tests fail, and the backlog list contains a topic:rejecting.topic entry with offset:-1.
2. Run git revert --abort and rerun the same command. The tests pass, and only the ok.topic backlog with its real partition and offset is reported.
3. Optional live check: create a topic with max.message.bytes=512, produce a 2 KB record with retries=0 from an app with DD_DATA_STREAMS_ENABLED=true, and inspect the DSM payload sent to the agent. Before: it contains a kafka_produce backlog with offset -1 for that topic. After: no backlog for the failed send.

Additional Notes

Only records that Kafka marks as failed are dropped. The change does not affect acks=0 producers, because it adds no offset < 0 guard.

🤖 Generated with Claude Code

[dd-octo-sts]

🟢 Java Benchmark SLOs — All performance SLOs passed

Suite	Status
Startup	🟢 pass

SLO thresholds are defined here based on automatically generated metrics. A warning is raised when results are within 5% of the threshold.

PR vs. master results

Scenario	Candidate	master	Δ (95% CI of mean)
startup:insecure-bank:iast:Agent	14.73 s	14.61 s	[+0.1%; +1.5%] (maybe worse)
startup:insecure-bank:tracing:Agent	13.50 s	13.72 s	[-2.5%; -0.8%] (maybe better)
startup:petclinic:appsec:Agent	17.55 s	17.48 s	[-0.5%; +1.3%] (no difference)
startup:petclinic:iast:Agent	17.32 s	17.54 s	[-2.1%; -0.3%] (maybe better)
startup:petclinic:profiling:Agent	17.13 s	17.29 s	[-2.1%; +0.3%] (no difference)
startup:petclinic:sca:Agent	17.56 s	17.46 s	[-0.3%; +1.5%] (no difference)
startup:petclinic:tracing:Agent	16.59 s	16.64 s	[-1.3%; +0.7%] (no difference)

Commit: 3585db63 · CI Pipeline · Benchmarking Platform UI

---

Load and DaCapo benchmarks can be triggered manually in the GitLab pipeline. Results will appear in the Benchmarking Platform UI after completion.

[pr-commenter]

Kafka / producer-benchmark

Parameters

	Baseline	Candidate
baseline_or_candidate	baseline	candidate
git_branch	master	jj.botha/dsm-failed-kafka-produce-reports-backlog-offset
git_commit_date	1792170226	1792185672
git_commit_sha	6dbd635	3585db6

See matching parameters

	Baseline	Candidate
ci_job_date	1792186788	1792186788
ci_job_id	2148797639	2148797639
ci_pipeline_id	145427330	145427330
cpu_model	Intel(R) Xeon(R) Platinum 8259CL CPU @ 2.50GHz	Intel(R) Xeon(R) Platinum 8259CL CPU @ 2.50GHz
jdkVersion	11.0.31	11.0.31
jmhVersion	1.36	1.36
jvm	/usr/lib/jvm/java-11-openjdk-amd64/bin/java	/usr/lib/jvm/java-11-openjdk-amd64/bin/java
jvmArgs	-Dhttp.proxyHost=127.0.0.1 -Dhttp.proxyPort=15002 -Dhttps.proxyHost=127.0.0.1 -Dhttps.proxyPort=15002 -Dhttp.nonProxyHosts=localhost	*.localhost
kernel_version	Linux runner-zfyrx7zua-project-304-concurrent-0-x5xloai8 6.8.0-1031-aws #33~22.04.1-Ubuntu SMP Thu Jun 26 14:22:30 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux	Linux runner-zfyrx7zua-project-304-concurrent-0-x5xloai8 6.8.0-1031-aws #33~22.04.1-Ubuntu SMP Thu Jun 26 14:22:30 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
vmName	OpenJDK 64-Bit Server VM	OpenJDK 64-Bit Server VM
vmVersion	11.0.31+11-post-1ubuntu1-22.04.2-Ubuntu	11.0.31+11-post-1ubuntu1-22.04.2-Ubuntu

Summary

Found 0 performance improvements and 0 performance regressions! Performance is the same for 3 metrics, 0 unstable metrics.

See unchanged results

scenario	Δ mean throughput
scenario:not-instrumented/KafkaProduceBenchmark.benchProduce	same
scenario:only-tracing-dsm-disabled-benchmarks/KafkaProduceBenchmark.benchProduce	same
scenario:only-tracing-dsm-enabled-benchmarks/KafkaProduceBenchmark.benchProduce	unsure [+185.162op/s; +5443.331op/s] or [+0.147%; +4.323%]