Skip to content

Commit fb55014

Browse files
piochelepiotrclaude
piochelepiotr
and
claude
committed
Fix review issues: type values, debug logging, dedup map, and null checks
- Use "kafka_producer"/"kafka_consumer" consistently in KafkaConfigHelper - Remove log.warn debug logging from MsgPackDatastreamsPayloadWriter - Remove unbounded reportedKafkaConfigs dedup map; configs flow through StatsBucket like other stats points - Make MASKED_VALUE and ALLOWED_KEYS public in KafkaConfigHelper - Remove redundant null checks in writeKafkaConfigs (constructor normalizes) - Update tests to reflect removal of dedup behavior Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent c0cff1d commit fb55014

File tree

5 files changed

+58
-114
lines changed

5 files changed

+58
-114
lines changed

‎dd-java-agent/instrumentation/kafka/kafka-common/src/main/java/datadog/trace/instrumentation/kafka_common/KafkaConfigHelper.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,13 @@
1616
public class KafkaConfigHelper {
1717
private static final Logger log = LoggerFactory.getLogger(KafkaConfigHelper.class);
1818

19-
static final String MASKED_VALUE = "****";
19+
public static final String MASKED_VALUE = "****";
2020

2121
/**
2222
* Config keys that are safe to capture with their values. Other keys are captured with masked
2323
* values.
2424
*/
25-
static final Set<String> ALLOWED_KEYS =
25+
public static final Set<String> ALLOWED_KEYS =
2626
new HashSet<>(
2727
Arrays.asList(
2828
// Common client configs
@@ -93,15 +93,15 @@ public class KafkaConfigHelper {
9393

9494
/** Store a producer config to be reported once the cluster ID is known from metadata. */
9595
public static void storePendingProducerConfig(MetadataState state, Map<String, String> config) {
96-
state.setPendingConfig(new PendingConfig("producer", "", config));
96+
state.setPendingConfig(new PendingConfig("kafka_producer", "", config));
9797
log.debug("Stored pending producer config (cluster ID not yet known)");
9898
}
9999

100100
/** Store a consumer config to be reported once the cluster ID is known from metadata. */
101101
public static void storePendingConsumerConfig(
102102
MetadataState state, String consumerGroup, Map<String, String> config) {
103103
state.setPendingConfig(
104-
new PendingConfig("consumer", consumerGroup != null ? consumerGroup : "", config));
104+
new PendingConfig("kafka_consumer", consumerGroup != null ? consumerGroup : "", config));
105105
log.debug("Stored pending consumer config (cluster ID not yet known)");
106106
}
107107

‎dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoring.java‎

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -82,8 +82,6 @@ public class DefaultDataStreamsMonitoring implements DataStreamsMonitoring, Even
8282
private volatile boolean agentSupportsDataStreams = false;
8383
private volatile boolean configSupportsDataStreams = false;
8484
private final ConcurrentHashMap<String, SchemaSampler> schemaSamplers;
85-
private final ConcurrentHashMap<KafkaConfigReport, Boolean> reportedKafkaConfigs =
86-
new ConcurrentHashMap<>();
8785
private static final ThreadLocal<String> serviceNameOverride = new ThreadLocal<>();
8886

8987
// contains a list of active extractors by type. Thread-safe via volatile with immutable
@@ -296,17 +294,14 @@ public void reportSchemaRegistryUsage(
296294
@Override
297295
public void reportKafkaConfig(
298296
String type, String kafkaClusterId, String consumerGroup, Map<String, String> config) {
299-
KafkaConfigReport report =
297+
inbox.offer(
300298
new KafkaConfigReport(
301299
type,
302300
kafkaClusterId,
303301
consumerGroup,
304302
config,
305303
timeSource.getCurrentTimeNanos(),
306-
getThreadServiceName());
307-
if (reportedKafkaConfigs.putIfAbsent(report, Boolean.TRUE) == null) {
308-
inbox.offer(report);
309-
}
304+
getThreadServiceName()));
310305
}
311306

312307
@Override
@@ -523,7 +518,6 @@ private void flush(long timestampNanos) {
523518
public void clear() {
524519
timeToBucket.clear();
525520
schemaSamplers.clear();
526-
reportedKafkaConfigs.clear();
527521
}
528522

529523
void report() {

‎dd-trace-core/src/main/java/datadog/trace/core/datastreams/MsgPackDatastreamsPayloadWriter.java‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,11 @@
1717
import java.util.Collection;
1818
import java.util.List;
1919
import java.util.Map;
20+
import org.slf4j.Logger;
21+
import org.slf4j.LoggerFactory;
2022

2123
public class MsgPackDatastreamsPayloadWriter implements DatastreamsPayloadWriter {
24+
private static final Logger log = LoggerFactory.getLogger(MsgPackDatastreamsPayloadWriter.class);
2225
private static final byte[] ENV = "Env".getBytes(ISO_8859_1);
2326
private static final byte[] VERSION = "Version".getBytes(ISO_8859_1);
2427
private static final byte[] PRIMARY_TAG = "PrimaryTag".getBytes(ISO_8859_1);
@@ -290,14 +293,13 @@ private void writeKafkaConfigs(List<KafkaConfigReport> configs, Writable packer)
290293
packer.startMap(4); // Type, KafkaClusterId, ConsumerGroup, Config
291294

292295
packer.writeUTF8(CONFIG_TYPE);
293-
packer.writeString(config.getType() != null ? config.getType() : "", null);
296+
packer.writeString(config.getType(), null);
294297

295298
packer.writeUTF8(CONFIG_KAFKA_CLUSTER_ID);
296-
packer.writeString(
297-
config.getKafkaClusterId() != null ? config.getKafkaClusterId() : "", null);
299+
packer.writeString(config.getKafkaClusterId(), null);
298300

299301
packer.writeUTF8(CONFIG_CONSUMER_GROUP);
300-
packer.writeString(config.getConsumerGroup() != null ? config.getConsumerGroup() : "", null);
302+
packer.writeString(config.getConsumerGroup(), null);
301303

302304
packer.writeUTF8(CONFIG_ENTRIES);
303305
Map<String, String> entries = config.getConfig();

‎dd-trace-core/src/test/groovy/datadog/trace/core/datastreams/DataStreamsWritingTest.groovy‎

Lines changed: 25 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -219,7 +219,7 @@ class DataStreamsWritingTest extends DDCoreSpecification {
219219
validateKafkaConfigMessage(requestBodies[0])
220220
}
221221

222-
def "Duplicate Kafka configs are not serialized twice"() {
222+
def "Duplicate Kafka configs are each serialized in the payload"() {
223223
given:
224224
def conditions = new PollingConditions(timeout: 2)
225225

@@ -252,7 +252,7 @@ class DataStreamsWritingTest extends DDCoreSpecification {
252252
def dataStreams = new DefaultDataStreamsMonitoring(fakeConfig, sharedCommObjects, timeSource, { traceConfig })
253253
dataStreams.start()
254254

255-
// Report the same producer config twice
255+
// Report the same producer config twice — both should be serialized
256256
dataStreams.reportKafkaConfig("kafka_producer", "", "", ["bootstrap.servers": "localhost:9092", "acks": "all"])
257257
dataStreams.reportKafkaConfig("kafka_producer", "", "", ["bootstrap.servers": "localhost:9092", "acks": "all"])
258258

@@ -267,7 +267,7 @@ class DataStreamsWritingTest extends DDCoreSpecification {
267267
assert requestBodies.size() == 1
268268
}
269269

270-
validateDedupedKafkaConfigMessage(requestBodies[0])
270+
validateDuplicateKafkaConfigMessage(requestBodies[0])
271271
}
272272

273273
def validateKafkaConfigMessage(byte[] message) {
@@ -346,7 +346,7 @@ class DataStreamsWritingTest extends DDCoreSpecification {
346346
return true
347347
}
348348

349-
def validateDedupedKafkaConfigMessage(byte[] message) {
349+
def validateDuplicateKafkaConfigMessage(byte[] message) {
350350
GzipSource gzipSource = new GzipSource(Okio.source(new ByteArrayInputStream(message)))
351351
BufferedSource bufferedSource = Okio.buffer(gzipSource)
352352
MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(bufferedSource.inputStream())
@@ -368,26 +368,28 @@ class DataStreamsWritingTest extends DDCoreSpecification {
368368
if (bucketKey == "Configs") {
369369
foundConfigs = true
370370
def numConfigs = unpacker.unpackArrayHeader()
371-
// Only 1 config should be present (deduplication)
372-
assert numConfigs == 1
373-
374-
assert unpacker.unpackMapHeader() == 4
375-
assert unpacker.unpackString() == "Type"
376-
assert unpacker.unpackString() == "kafka_producer"
377-
assert unpacker.unpackString() == "KafkaClusterId"
378-
unpacker.unpackString() // skip cluster id value
379-
assert unpacker.unpackString() == "ConsumerGroup"
380-
unpacker.unpackString() // skip consumer group value
381-
assert unpacker.unpackString() == "Config"
382-
def configSize = unpacker.unpackMapHeader()
383-
Map<String, String> configEntries = [:]
384-
configSize.times {
385-
def ck = unpacker.unpackString()
386-
def cv = unpacker.unpackString()
387-
configEntries[ck] = cv
371+
// Both configs should be present (no deduplication)
372+
assert numConfigs == 2
373+
374+
numConfigs.times {
375+
assert unpacker.unpackMapHeader() == 4
376+
assert unpacker.unpackString() == "Type"
377+
assert unpacker.unpackString() == "kafka_producer"
378+
assert unpacker.unpackString() == "KafkaClusterId"
379+
unpacker.unpackString() // skip cluster id value
380+
assert unpacker.unpackString() == "ConsumerGroup"
381+
unpacker.unpackString() // skip consumer group value
382+
assert unpacker.unpackString() == "Config"
383+
def configSize = unpacker.unpackMapHeader()
384+
Map<String, String> configEntries = [:]
385+
configSize.times {
386+
def ck = unpacker.unpackString()
387+
def cv = unpacker.unpackString()
388+
configEntries[ck] = cv
389+
}
390+
assert configEntries["bootstrap.servers"] == "localhost:9092"
391+
assert configEntries["acks"] == "all"
388392
}
389-
assert configEntries["bootstrap.servers"] == "localhost:9092"
390-
assert configEntries["acks"] == "all"
391393
} else {
392394
unpacker.skipValue()
393395
}

‎dd-trace-core/src/test/groovy/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTest.groovy‎

Lines changed: 21 additions & 75 deletions
Original file line numberDiff line numberDiff line change
@@ -1105,7 +1105,7 @@ class DefaultDataStreamsMonitoringTest extends DDCoreSpecification {
11051105
dataStreams.close()
11061106
}
11071107

1108-
def "Duplicate Kafka configs are deduplicated and only sent once"() {
1108+
def "Duplicate Kafka configs are each reported in the bucket"() {
11091109
given:
11101110
def conditions = new PollingConditions(timeout: 1)
11111111
def features = Stub(DDAgentFeaturesDiscovery) {
@@ -1125,32 +1125,30 @@ class DefaultDataStreamsMonitoringTest extends DDCoreSpecification {
11251125
def config1 = ["bootstrap.servers": "localhost:9092", "acks": "all"]
11261126
def config2 = ["bootstrap.servers": "localhost:9092", "acks": "all"]
11271127
dataStreams.reportKafkaConfig("kafka_producer", "", "", config1)
1128-
dataStreams.reportKafkaConfig("kafka_producer", "", "", config2) // duplicate, should be ignored
1128+
dataStreams.reportKafkaConfig("kafka_producer", "", "", config2)
11291129
timeSource.advance(DEFAULT_BUCKET_DURATION_NANOS)
11301130
dataStreams.report()
11311131

1132-
then: "only one config is reported in the bucket"
1132+
then: "both configs are reported in the bucket"
11331133
conditions.eventually {
11341134
assert dataStreams.inbox.isEmpty()
11351135
assert dataStreams.thread.state != Thread.State.RUNNABLE
11361136
assert payloadWriter.buckets.size() == 1
11371137
}
11381138

11391139
with(payloadWriter.buckets.get(0)) {
1140-
kafkaConfigs.size() == 1
1141-
with(kafkaConfigs.get(0)) {
1142-
type == "kafka_producer"
1143-
config["bootstrap.servers"] == "localhost:9092"
1144-
config["acks"] == "all"
1145-
}
1140+
kafkaConfigs.size() == 2
1141+
kafkaConfigs.every { it.type == "kafka_producer" }
1142+
kafkaConfigs.every { it.config["bootstrap.servers"] == "localhost:9092" }
1143+
kafkaConfigs.every { it.config["acks"] == "all" }
11461144
}
11471145

11481146
cleanup:
11491147
payloadWriter.close()
11501148
dataStreams.close()
11511149
}
11521150

1153-
def "Duplicate Kafka configs across buckets are deduplicated"() {
1151+
def "Kafka configs reported in separate buckets appear in each bucket"() {
11541152
given:
11551153
def conditions = new PollingConditions(timeout: 1)
11561154
def features = Stub(DDAgentFeaturesDiscovery) {
@@ -1164,7 +1162,7 @@ class DefaultDataStreamsMonitoringTest extends DDCoreSpecification {
11641162
isDataStreamsEnabled() >> true
11651163
}
11661164

1167-
when: "reporting the same config in two different bucket windows"
1165+
when: "reporting a config in the first bucket"
11681166
def dataStreams = new DefaultDataStreamsMonitoring(sink, features, timeSource, { traceConfig }, payloadWriter, DEFAULT_BUCKET_DURATION_NANOS)
11691167
dataStreams.start()
11701168
def config = ["bootstrap.servers": "localhost:9092", "acks": "all"]
@@ -1185,17 +1183,25 @@ class DefaultDataStreamsMonitoringTest extends DDCoreSpecification {
11851183

11861184
when: "reporting the same config again in a new bucket"
11871185
payloadWriter.buckets.clear()
1188-
dataStreams.reportKafkaConfig("kafka_producer", "", "", config) // duplicate, should be ignored globally
1186+
dataStreams.reportKafkaConfig("kafka_producer", "", "", config)
11891187
timeSource.advance(DEFAULT_BUCKET_DURATION_NANOS)
11901188
dataStreams.report()
11911189

1192-
then: "second bucket has no configs because the duplicate was filtered"
1190+
then: "second bucket also has the config"
11931191
conditions.eventually {
11941192
assert dataStreams.inbox.isEmpty()
11951193
assert dataStreams.thread.state != Thread.State.RUNNABLE
1194+
assert payloadWriter.buckets.size() == 1
1195+
}
1196+
1197+
with(payloadWriter.buckets.get(0)) {
1198+
kafkaConfigs.size() == 1
1199+
with(kafkaConfigs.get(0)) {
1200+
type == "kafka_producer"
1201+
config["bootstrap.servers"] == "localhost:9092"
1202+
config["acks"] == "all"
1203+
}
11961204
}
1197-
// Either no buckets at all, or buckets with empty kafkaConfigs
1198-
payloadWriter.buckets.every { it.kafkaConfigs.isEmpty() }
11991205

12001206
cleanup:
12011207
payloadWriter.close()
@@ -1418,66 +1424,6 @@ class DefaultDataStreamsMonitoringTest extends DDCoreSpecification {
14181424
payloadWriter.close()
14191425
}
14201426

1421-
def "clear() resets Kafka config dedup cache"() {
1422-
given:
1423-
def conditions = new PollingConditions(timeout: 1)
1424-
def features = Stub(DDAgentFeaturesDiscovery) {
1425-
supportsDataStreams() >> true
1426-
}
1427-
def timeSource = new ControllableTimeSource()
1428-
def sink = Mock(Sink)
1429-
def payloadWriter = new CapturingPayloadWriter()
1430-
1431-
def traceConfig = Mock(TraceConfig) {
1432-
isDataStreamsEnabled() >> true
1433-
}
1434-
1435-
when: "reporting config, flushing, clearing, then reporting the same config again"
1436-
def dataStreams = new DefaultDataStreamsMonitoring(sink, features, timeSource, { traceConfig }, payloadWriter, DEFAULT_BUCKET_DURATION_NANOS)
1437-
dataStreams.start()
1438-
def config = ["bootstrap.servers": "localhost:9092", "acks": "all"]
1439-
dataStreams.reportKafkaConfig("kafka_producer", "", "", config)
1440-
timeSource.advance(DEFAULT_BUCKET_DURATION_NANOS)
1441-
dataStreams.report()
1442-
1443-
then: "first config is reported"
1444-
conditions.eventually {
1445-
assert dataStreams.inbox.isEmpty()
1446-
assert dataStreams.thread.state != Thread.State.RUNNABLE
1447-
assert payloadWriter.buckets.size() == 1
1448-
}
1449-
1450-
with(payloadWriter.buckets.get(0)) {
1451-
kafkaConfigs.size() == 1
1452-
}
1453-
1454-
when: "clearing the state and reporting the same config"
1455-
payloadWriter.buckets.clear()
1456-
dataStreams.clear()
1457-
dataStreams.reportKafkaConfig("kafka_producer", "", "", config)
1458-
timeSource.advance(DEFAULT_BUCKET_DURATION_NANOS)
1459-
dataStreams.report()
1460-
1461-
then: "the config is reported again because the dedup cache was cleared"
1462-
conditions.eventually {
1463-
assert dataStreams.inbox.isEmpty()
1464-
assert dataStreams.thread.state != Thread.State.RUNNABLE
1465-
assert payloadWriter.buckets.size() == 1
1466-
}
1467-
1468-
with(payloadWriter.buckets.get(0)) {
1469-
kafkaConfigs.size() == 1
1470-
with(kafkaConfigs.get(0)) {
1471-
type == "kafka_producer"
1472-
config["bootstrap.servers"] == "localhost:9092"
1473-
}
1474-
}
1475-
1476-
cleanup:
1477-
payloadWriter.close()
1478-
dataStreams.close()
1479-
}
1480-
14811427
def "KafkaConfigReport equals and hashCode work correctly"() {
14821428
given:
14831429
def config1 = new KafkaConfigReport("kafka_producer", "", "", ["bootstrap.servers": "localhost:9092", "acks": "all"], 1000L, null)

0 commit comments

Comments
 (0)