Skip to content

Fix KafkaStreamsTest #8638

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

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Fix KafkaStreamsTest #8638

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5809db7
Fix KafkaStreamsTest
smola Mar 28, 2025
e17d650
Remove Flaky mark from KafkaStreamsTest
smola Apr 7, 2025
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
33 changes: 24 additions & 9 deletions ...agent/instrumentation/kafka-streams-0.11/src/latestDepTest/groovy/KafkaStreamsTest.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ import datadog.trace.api.DDTags
import datadog.trace.bootstrap.instrumentation.api.InstrumentationTags
import datadog.trace.bootstrap.instrumentation.api.Tags
import datadog.trace.core.datastreams.StatsGroup
import datadog.trace.test.util.Flaky
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.streams.KafkaStreams
Expand All @@ -27,8 +26,8 @@ import spock.lang.Shared

import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.TimeUnit
import java.util.regex.Pattern

@Flaky("https://github.com/DataDog/dd-trace-java/issues/3865")
class KafkaStreamsTest extends AgentTestRunner {
static final STREAM_PENDING = "test.pending"
static final STREAM_PROCESSED = "test.processed"
Expand All @@ -39,6 +38,11 @@ class KafkaStreamsTest extends AgentTestRunner {
@Shared
EmbeddedKafkaBroker embeddedKafka = kafkaRule.embeddedKafka

def setup() {
// Filter out additional traces for kafka.poll operation, otherwise, there will be more traces than expected.
TEST_WRITER.setFilter { trace -> trace[0].operationName.toString() != 'kafka.poll' }
}

@Override
protected boolean isDataStreamsEnabled() {
return true
Expand Down Expand Up @@ -135,6 +139,7 @@ class KafkaStreamsTest extends AgentTestRunner {
if ({ isDataStreamsEnabled()}) {
"$DDTags.PATHWAY_HASH" { String }
}
"$InstrumentationTags.KAFKA_BOOTSTRAP_SERVERS" Pattern.compile("127.0.0.1:[0-9]+")
defaultTagsNoPeerService()
}
}
Expand All @@ -159,7 +164,6 @@ class KafkaStreamsTest extends AgentTestRunner {
"$InstrumentationTags.PARTITION" { it >= 0 }
"$InstrumentationTags.OFFSET" 0
"$InstrumentationTags.PROCESSOR_NAME" "KSTREAM-SOURCE-0000000000"
"$InstrumentationTags.MESSAGING_DESTINATION_NAME" "$STREAM_PENDING"
"asdf" "testing"
if ({ isDataStreamsEnabled()}) {
"$DDTags.PATHWAY_HASH" { String }
Expand All @@ -186,6 +190,7 @@ class KafkaStreamsTest extends AgentTestRunner {
if ({ isDataStreamsEnabled()}) {
"$DDTags.PATHWAY_HASH" { String }
}
"$InstrumentationTags.KAFKA_BOOTSTRAP_SERVERS" Pattern.compile("127.0.0.1:[0-9]+")
defaultTagsNoPeerService()
}
}
Expand All @@ -212,6 +217,7 @@ class KafkaStreamsTest extends AgentTestRunner {
if ({ isDataStreamsEnabled()}) {
"$DDTags.PATHWAY_HASH" { String }
}
"$InstrumentationTags.KAFKA_BOOTSTRAP_SERVERS" Pattern.compile("127.0.0.1:[0-9]+")
defaultTags(true)
}
}
Expand All @@ -226,8 +232,11 @@ class KafkaStreamsTest extends AgentTestRunner {
if (isDataStreamsEnabled()) {
StatsGroup originProducerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
verifyAll(originProducerPoint) {
edgeTags == ["direction:out", "topic:$STREAM_PENDING", "type:kafka"]
edgeTags.size() == 3
edgeTags.any { it.startsWith("kafka_cluster_id:") }
for (String tag : ["direction:out", "topic:$STREAM_PENDING", "type:kafka"]) {
assert edgeTags.contains(tag)
}
edgeTags.size() == 4
}

StatsGroup kafkaStreamsConsumerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == originProducerPoint.hash }
Expand All @@ -243,14 +252,20 @@ class KafkaStreamsTest extends AgentTestRunner {

StatsGroup kafkaStreamsProducerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == kafkaStreamsConsumerPoint.hash }
verifyAll(kafkaStreamsProducerPoint) {
edgeTags == ["direction:out", "topic:$STREAM_PROCESSED", "type:kafka"]
edgeTags.size() == 3
edgeTags.any { it.startsWith("kafka_cluster_id:") }
for (String tag : ["direction:out", "topic:$STREAM_PROCESSED", "type:kafka"]) {
assert edgeTags.contains(tag)
}
edgeTags.size() == 4
}

StatsGroup finalConsumerPoint = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == kafkaStreamsProducerPoint.hash }
verifyAll(finalConsumerPoint) {
edgeTags == ["direction:in", "group:sender", "topic:$STREAM_PROCESSED".toString(), "type:kafka"]
edgeTags.size() == 4
edgeTags.any { it.startsWith("kafka_cluster_id:") }
for (String tag : ["direction:in", "group:sender", "topic:$STREAM_PROCESSED".toString(), "type:kafka"]) {
assert edgeTags.contains(tag)
}
edgeTags.size() == 5
}
}

Expand Down