Skip to content

Commit 3bf87d6

Browse files
kr-igorkr-igor
committed
Fix KStreams with dynamic queue names
1 parent 08d7b46 commit 3bf87d6

4 files changed

Lines changed: 11 additions & 7 deletions

File tree

‎dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaProducerInstrumentation.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -132,7 +132,8 @@ public static AgentScope onEnter(
132132
sortedTags.put(TYPE_TAG, "kafka");
133133
try {
134134
propagate().inject(span, record.headers(), SETTER);
135-
if (STREAMING_CONTEXT.empty() || STREAMING_CONTEXT.isSinkTopic(record.topic())) {
135+
if (STREAMING_CONTEXT.isDisabledForTopic(record.topic()) ||
136+
STREAMING_CONTEXT.isSinkTopic(record.topic())) {
136137
// inject the context in the headers, but delay sending the stats until we know the
137138
// message size.
138139
// The stats are saved in the pathway context and sent in PayloadSizeAdvice.
@@ -153,7 +154,8 @@ record =
153154
record.headers());
154155

155156
propagate().inject(span, record.headers(), SETTER);
156-
if (STREAMING_CONTEXT.empty() || STREAMING_CONTEXT.isSinkTopic(record.topic())) {
157+
if (STREAMING_CONTEXT.isDisabledForTopic(record.topic()) ||
158+
STREAMING_CONTEXT.isSinkTopic(record.topic())) {
157159
propagate()
158160
.injectPathwayContextWithoutSendingStats(
159161
span, record.headers(), SETTER, sortedTags);

‎dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,7 @@ protected void startNewRecordSpan(ConsumerRecord<?, ?> val) {
105105

106106
final long payloadSize =
107107
span.traceConfig().isDataStreamsEnabled() ? computePayloadSizeBytes(val) : 0;
108-
if (STREAMING_CONTEXT.empty()) {
108+
if (STREAMING_CONTEXT.isDisabledForTopic(val.topic())) {
109109
AgentTracer.get()
110110
.getDataStreamsMonitoring()
111111
.setCheckpoint(span, sortedTags, val.timestamp(), payloadSize);

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -64,8 +64,10 @@ public boolean isSourceTopic(final String topic) {
6464
return Objects.equals(topics.getOrDefault(topic, UNKNOWN_TOPIC), SOURCE_TOPIC);
6565
}
6666

67-
public boolean empty() {
68-
return topics.isEmpty();
67+
// Checks if this topic is a part of a streaming topology
68+
public boolean isDisabledForTopic(final String topic) {
69+
return topics.isEmpty() ||
70+
Objects.equals(topics.getOrDefault(topic, UNKNOWN_TOPIC), UNKNOWN_TOPIC);
6971
}
7072

7173
private final Set<String> allSourceTopics = ConcurrentHashMap.newKeySet();

‎dd-java-agent/instrumentation/kafka-streams-0.11/src/main/java/datadog/trace/instrumentation/kafka_streams/KafkaStreamTaskInstrumentation.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -255,7 +255,7 @@ public static void start(
255255

256256
final long payloadSize =
257257
span.traceConfig().isDataStreamsEnabled() ? computePayloadSizeBytes(record.value) : 0;
258-
if (STREAMING_CONTEXT.empty()) {
258+
if (STREAMING_CONTEXT.isDisabledForTopic(record.topic())) {
259259
AgentTracer.get()
260260
.getDataStreamsMonitoring()
261261
.setCheckpoint(span, sortedTags, record.timestamp, payloadSize);
@@ -337,7 +337,7 @@ public static void start(
337337
payloadSize = metadata.serializedKeySize() + metadata.serializedValueSize();
338338
}
339339

340-
if (STREAMING_CONTEXT.empty()) {
340+
if (STREAMING_CONTEXT.isDisabledForTopic(record.topic())) {
341341
AgentTracer.get()
342342
.getDataStreamsMonitoring()
343343
.setCheckpoint(span, sortedTags, record.timestamp(), payloadSize);

0 commit comments

Comments
 (0)