Skip to content

Fix KStreams with dynamic queue names #6857

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Mar 29, 2024
Merged

Fix KStreams with dynamic queue names #6857

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3bf87d6
Fix KStreams with dynamic queue names
kr-igor Mar 29, 2024
f8acfb7
Spotless apply
kr-igor Mar 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions ...c/main/java/datadog/trace/instrumentation/kafka_clients/KafkaProducerInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,8 @@ public static AgentScope onEnter(
sortedTags.put(TYPE_TAG, "kafka");
try {
propagate().inject(span, record.headers(), SETTER);
if (STREAMING_CONTEXT.empty() || STREAMING_CONTEXT.isSinkTopic(record.topic())) {
if (STREAMING_CONTEXT.isDisabledForTopic(record.topic())
|| STREAMING_CONTEXT.isSinkTopic(record.topic())) {
// inject the context in the headers, but delay sending the stats until we know the
// message size.
// The stats are saved in the pathway context and sent in PayloadSizeAdvice.
Expand All @@ -153,7 +154,8 @@ record =
record.headers());

propagate().inject(span, record.headers(), SETTER);
if (STREAMING_CONTEXT.empty() || STREAMING_CONTEXT.isSinkTopic(record.topic())) {
if (STREAMING_CONTEXT.isDisabledForTopic(record.topic())
|| STREAMING_CONTEXT.isSinkTopic(record.topic())) {
propagate()
.injectPathwayContextWithoutSendingStats(
span, record.headers(), SETTER, sortedTags);
Expand Down
2 changes: 1 addition & 1 deletion ...ients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ protected void startNewRecordSpan(ConsumerRecord<?, ?> val) {

final long payloadSize =
span.traceConfig().isDataStreamsEnabled() ? computePayloadSizeBytes(val) : 0;
if (STREAMING_CONTEXT.empty()) {
if (STREAMING_CONTEXT.isDisabledForTopic(val.topic())) {
AgentTracer.get()
.getDataStreamsMonitoring()
.setCheckpoint(span, sortedTags, val.timestamp(), payloadSize);
Expand Down
6 changes: 4 additions & 2 deletions ...fka-common/src/main/java/datadog/trace/instrumentation/kafka_common/StreamingContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,8 +64,10 @@ public boolean isSourceTopic(final String topic) {
return Objects.equals(topics.getOrDefault(topic, UNKNOWN_TOPIC), SOURCE_TOPIC);
}

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

private final Set<String> allSourceTopics = ConcurrentHashMap.newKeySet();
Expand Down
4 changes: 2 additions & 2 deletions ...main/java/datadog/trace/instrumentation/kafka_streams/KafkaStreamTaskInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,7 +255,7 @@ public static void start(

final long payloadSize =
span.traceConfig().isDataStreamsEnabled() ? computePayloadSizeBytes(record.value) : 0;
if (STREAMING_CONTEXT.empty()) {
if (STREAMING_CONTEXT.isDisabledForTopic(record.topic())) {
AgentTracer.get()
.getDataStreamsMonitoring()
.setCheckpoint(span, sortedTags, record.timestamp, payloadSize);
Expand Down Expand Up @@ -337,7 +337,7 @@ public static void start(
payloadSize = metadata.serializedKeySize() + metadata.serializedValueSize();
}

if (STREAMING_CONTEXT.empty()) {
if (STREAMING_CONTEXT.isDisabledForTopic(record.topic())) {
AgentTracer.get()
.getDataStreamsMonitoring()
.setCheckpoint(span, sortedTags, record.timestamp(), payloadSize);
Expand Down