From 488eb2f67608f8ef3b55dd372e03884805f6db11 Mon Sep 17 00:00:00 2001 From: Nelson Osacky Date: Tue, 20 Oct 2026 19:52:15 +0200 Subject: [PATCH] test(kafka): Add an end-to-end test of the raw Kafka client's tracing SentryKafkaEndToEndTest publishes a message through a wrapped KafkaProducer to an in-JVM KRaft broker, consumes it with SentryKafkaConsumerTracing, and checks what reaches MockSentryServer: the queue.publish span in the producing transaction, and a queue.process transaction with the consumer origin and the Kafka system that continues the producer's trace. It ports the Kafka part of the console sample's system test, which needed a Redpanda container. The broker comes from spring-kafka-test 3, which needs Java 17, so the test classpaths resolve Java 17 variants. The tests run on it anyway. With its core, profiling and Kafka parts ported, the console sample has no system test left, so it comes off the system-test runner and the backend workflow, whose base row moves to the console OTLP sample. Co-Authored-By: Claude Opus 5.5 (1M context) --- .github/workflows/system-tests-backend.yml | 5 +- sentry-kafka/build.gradle.kts | 10 ++ .../sentry/kafka/SentryKafkaEndToEndTest.kt | 103 +++++++++++++ .../sentry-samples-console/build.gradle.kts | 41 ------ .../src/test/kotlin/io/sentry/DummyTest.kt | 12 -- .../ConsoleApplicationSystemTest.kt | 137 ------------------ test/system-test-runner.py | 5 +- 7 files changed, 115 insertions(+), 198 deletions(-) create mode 100644 sentry-kafka/src/test/kotlin/io/sentry/kafka/SentryKafkaEndToEndTest.kt delete mode 100644 sentry-samples/sentry-samples-console/src/test/kotlin/io/sentry/DummyTest.kt delete mode 100644 sentry-samples/sentry-samples-console/src/test/kotlin/io/sentry/systemtest/ConsoleApplicationSystemTest.kt diff --git a/.github/workflows/system-tests-backend.yml b/.github/workflows/system-tests-backend.yml index e8a43cba0ee..eb29d4c92da 100644 --- a/.github/workflows/system-tests-backend.yml +++ b/.github/workflows/system-tests-backend.yml @@ -24,13 +24,10 @@ jobs: strategy: fail-fast: false matrix: - sample: [ "sentry-samples-console" ] + sample: [ "sentry-samples-console-otlp" ] agent: [ "false" ] agent-auto-init: [ "true" ] include: - - sample: "sentry-samples-console-otlp" - agent: "false" - agent-auto-init: "true" - sample: "sentry-samples-logback" agent: "false" agent-auto-init: "true" diff --git a/sentry-kafka/build.gradle.kts b/sentry-kafka/build.gradle.kts index 0d543bad270..469e1ee576f 100644 --- a/sentry-kafka/build.gradle.kts +++ b/sentry-kafka/build.gradle.kts @@ -28,10 +28,13 @@ dependencies { // tests testImplementation(projects.sentryTestSupport) testImplementation(kotlin(Config.kotlinStdLib)) + testImplementation(libs.google.truth) testImplementation(libs.kotlin.test.junit) testImplementation(libs.mockito.kotlin) testImplementation(libs.mockito.inline) testImplementation(libs.kafka.clients) + // For an in-JVM broker. It brings no Spring application into the tests. + testImplementation(libs.spring.kafka3.test) } tasks.withType().configureEach { @@ -60,3 +63,10 @@ tasks.jar { ) } } + +// The tests' in-JVM Kafka broker needs Java 17, which the tests run on. +listOf("testCompileClasspath", "testRuntimeClasspath").forEach { + configurations.named(it) { + attributes { attribute(TargetJvmVersion.TARGET_JVM_VERSION_ATTRIBUTE, 17) } + } +} diff --git a/sentry-kafka/src/test/kotlin/io/sentry/kafka/SentryKafkaEndToEndTest.kt b/sentry-kafka/src/test/kotlin/io/sentry/kafka/SentryKafkaEndToEndTest.kt new file mode 100644 index 00000000000..2c42817a4ec --- /dev/null +++ b/sentry-kafka/src/test/kotlin/io/sentry/kafka/SentryKafkaEndToEndTest.kt @@ -0,0 +1,103 @@ +package io.sentry.kafka + +import com.google.common.truth.Truth.assertThat +import io.sentry.Sentry +import io.sentry.test.MockSentryServer +import io.sentry.test.initForTest +import io.sentry.test.transaction +import java.time.Duration +import java.util.Properties +import kotlin.test.AfterTest +import kotlin.test.Test +import org.apache.kafka.clients.consumer.ConsumerConfig +import org.apache.kafka.clients.consumer.ConsumerRecord +import org.apache.kafka.clients.consumer.KafkaConsumer +import org.apache.kafka.clients.producer.KafkaProducer +import org.apache.kafka.clients.producer.ProducerConfig +import org.apache.kafka.clients.producer.ProducerRecord +import org.apache.kafka.common.serialization.StringDeserializer +import org.apache.kafka.common.serialization.StringSerializer +import org.springframework.kafka.test.EmbeddedKafkaKraftBroker + +/** + * Sends a message through the plain Kafka client to an in-JVM KRaft broker, so no Docker is needed, + * and checks that both sides reach Sentry over the real HTTP transport. + */ +class SentryKafkaEndToEndTest { + private val server = MockSentryServer() + private val kafka = EmbeddedKafkaKraftBroker(1, 1, TOPIC).apply { afterPropertiesSet() } + + @AfterTest + fun teardown() { + Sentry.close() + kafka.destroy() + server.close() + } + + @Test + fun `a message is traced where it's published and where it's processed`() { + initForTest { + it.dsn = server.dsn + it.tracesSampleRate = 1.0 + it.isEnableQueueTracing = true + } + + val transaction = Sentry.startTransaction("kafka-demo", "demo") + transaction.makeCurrent().use { + SentryKafkaProducer.wrap(KafkaProducer(producerProperties())).use { producer + -> + producer.send(ProducerRecord(TOPIC, "sentry-kafka message")).get() + } + } + transaction.finish() + + KafkaConsumer(consumerProperties()).use { consumer -> + consumer.subscribe(listOf(TOPIC)) + val record = pollOne(consumer) + SentryKafkaConsumerTracing.withTracing(record, Runnable {}) + } + + val producer = + server.awaitEnvelope { it.transaction()?.transaction == "kafka-demo" }.transaction()!! + assertThat(producer.spans.map { it.op }).contains("queue.publish") + + val consumer = + server + .awaitEnvelope { it.transaction()?.contexts?.trace?.operation == "queue.process" } + .transaction()!! + val trace = consumer.contexts.trace!! + assertThat(trace.origin).isEqualTo(SentryKafkaConsumerTracing.TRACE_ORIGIN) + assertThat(trace.data).containsEntry("messaging.system", "kafka") + // The producer's span puts the trace in the message's headers, and the consumer continues it. + assertThat(trace.traceId).isEqualTo(producer.contexts.trace!!.traceId) + } + + private fun pollOne(consumer: KafkaConsumer): ConsumerRecord { + repeat(20) { + consumer.poll(Duration.ofMillis(500)).firstOrNull()?.let { + return it + } + } + error("No message arrived") + } + + private fun producerProperties(): Properties = + Properties().apply { + put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.brokersAsString) + put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name) + put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name) + } + + private fun consumerProperties(): Properties = + Properties().apply { + put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.brokersAsString) + put(ConsumerConfig.GROUP_ID_CONFIG, "sentry-kafka-end-to-end-test") + put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name) + put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name) + } + + private companion object { + const val TOPIC = "sentry-topic" + } +} diff --git a/sentry-samples/sentry-samples-console/build.gradle.kts b/sentry-samples/sentry-samples-console/build.gradle.kts index d714a0ed541..9d14cdb56ae 100644 --- a/sentry-samples/sentry-samples-console/build.gradle.kts +++ b/sentry-samples/sentry-samples-console/build.gradle.kts @@ -1,12 +1,8 @@ -import org.jetbrains.kotlin.gradle.tasks.KotlinCompile - plugins { java application - alias(libs.plugins.kotlin.jvm) alias(libs.plugins.gradle.versions) alias(libs.plugins.shadow) - id("io.sentry.systemtest") } application { mainClass.set("io.sentry.samples.console.Main") } @@ -20,17 +16,6 @@ configure { targetCompatibility = JavaVersion.VERSION_17 } -tasks.withType().configureEach { - compilerOptions.jvmTarget = org.jetbrains.kotlin.gradle.dsl.JvmTarget.JVM_17 -} - -tasks.withType().configureEach { - kotlin { - compilerOptions.freeCompilerArgs = listOf("-Xjsr305=strict") - compilerOptions.jvmTarget = org.jetbrains.kotlin.gradle.dsl.JvmTarget.JVM_17 - } -} - dependencies { implementation(projects.sentry) implementation(projects.sentryAsyncProfiler) @@ -39,13 +24,6 @@ dependencies { implementation(libs.jcache) implementation(libs.caffeine.jcache) implementation(libs.kafka.clients) - - testImplementation(kotlin(Config.kotlinStdLib)) - testImplementation(projects.sentry) - testImplementation(projects.sentrySystemTestSupport) - testImplementation(libs.kotlin.test.junit) - testImplementation(libs.slf4j.api) - testImplementation(libs.slf4j.jdk14) } // Configure the Shadow JAR (executable JAR with all dependencies) @@ -63,22 +41,3 @@ tasks.jar { // Fix the startScripts task dependency tasks.startScripts { dependsOn(tasks.shadowJar) } - -tasks.register("systemTest").configure { - group = "verification" - description = "Runs the System tests" - - val test = project.extensions.getByType()["test"] - testClassesDirs = test.output.classesDirs - classpath = test.runtimeClasspath - - maxParallelForks = 1 - - filter { includeTestsMatching("io.sentry.systemtest*") } -} - -tasks.named("test").configure { - require(this is Test) - - filter { excludeTestsMatching("io.sentry.systemtest.*") } -} diff --git a/sentry-samples/sentry-samples-console/src/test/kotlin/io/sentry/DummyTest.kt b/sentry-samples/sentry-samples-console/src/test/kotlin/io/sentry/DummyTest.kt deleted file mode 100644 index 6f762b7e453..00000000000 --- a/sentry-samples/sentry-samples-console/src/test/kotlin/io/sentry/DummyTest.kt +++ /dev/null @@ -1,12 +0,0 @@ -package io.sentry - -import kotlin.test.Test -import kotlin.test.assertTrue - -class DummyTest { - @Test - fun `the only test`() { - // only needed to have more than 0 tests and not fail the build - assertTrue(true) - } -} diff --git a/sentry-samples/sentry-samples-console/src/test/kotlin/io/sentry/systemtest/ConsoleApplicationSystemTest.kt b/sentry-samples/sentry-samples-console/src/test/kotlin/io/sentry/systemtest/ConsoleApplicationSystemTest.kt deleted file mode 100644 index db6f54a616b..00000000000 --- a/sentry-samples/sentry-samples-console/src/test/kotlin/io/sentry/systemtest/ConsoleApplicationSystemTest.kt +++ /dev/null @@ -1,137 +0,0 @@ -package io.sentry.systemtest - -import io.sentry.protocol.SentryId -import io.sentry.systemtest.util.TestHelper -import java.util.concurrent.TimeUnit -import org.junit.Assert.assertEquals -import org.junit.Assert.assertTrue -import org.junit.Before -import org.junit.Test - -class ConsoleApplicationSystemTest { - lateinit var testHelper: TestHelper - - @Before - fun setup() { - testHelper = TestHelper("http://localhost:8000") - testHelper.reset() - } - - @Test - fun `console application sends expected events when run as JAR`() { - val process = launchConsoleProcess() - - process.waitFor(30, TimeUnit.SECONDS) - assertEquals(0, process.exitValue()) - - // Verify that we received the expected events - verifyExpectedEvents() - } - - @Test - fun `console application sends kafka producer and consumer tracing when kafka is enabled`() { - val process = - launchConsoleProcess(mapOf("SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS" to "localhost:9092")) - - process.waitFor(30, TimeUnit.SECONDS) - assertEquals(0, process.exitValue()) - - testHelper.ensureTransactionReceived { transaction, _ -> - transaction.transaction == "kafka-demo" && - testHelper.doesTransactionContainSpanWithOp(transaction, "queue.publish") - } - - testHelper.ensureTransactionReceived { transaction, _ -> - testHelper.doesTransactionHaveOp(transaction, "queue.process") && - transaction.contexts.trace?.origin == "manual.queue.kafka.consumer" && - transaction.contexts.trace?.data?.get("messaging.system") == "kafka" - } - } - - private fun launchConsoleProcess(overrides: Map = emptyMap()): Process { - val jarFile = testHelper.findJar("sentry-samples-console") - val env = - mutableMapOf( - "SENTRY_DSN" to testHelper.dsn, - "SENTRY_TRACES_SAMPLE_RATE" to "1.0", - "SENTRY_ENABLE_PRETTY_SERIALIZATION_OUTPUT" to "false", - "SENTRY_DEBUG" to "true", - "SENTRY_PROFILE_SESSION_SAMPLE_RATE" to "1.0", - "SENTRY_PROFILE_LIFECYCLE" to "TRACE", - ) - env.putAll(overrides) - return testHelper.launch(jarFile, env) - } - - private fun verifyExpectedEvents() { - var profilerId: SentryId? = null - // Verify we received a "Fatal message!" event - testHelper.ensureErrorReceived { event -> - event.message?.formatted == "Fatal message!" && event.level?.name == "FATAL" - } - - // Verify we received a "Some warning!" event - testHelper.ensureErrorReceived { event -> - event.message?.formatted == "Some warning!" && event.level?.name == "WARNING" - } - - // Verify we received the RuntimeException - testHelper.ensureErrorReceived { event -> - event.exceptions?.any { ex -> ex.type == "RuntimeException" && ex.value == "Some error!" } == - true && testHelper.doesEventHaveFlag(event, "my-feature-flag", true) - } - - // Verify we received the detailed event with fingerprint - testHelper.ensureErrorReceived { event -> - event.message?.message == "Detailed event" && - event.fingerprints?.contains("NewClientDebug") == true && - event.level?.name == "DEBUG" - } - - // Verify we received transaction events - testHelper.ensureTransactionReceived { transaction, _ -> - profilerId = transaction.contexts.profile?.profilerId - transaction.transaction == "transaction name" && - transaction.spans?.any { span -> span.op == "child" } == true - } - - testHelper.ensureProfileChunkReceived { profileChunk, envelopeHeader -> - profileChunk.profilerId == profilerId - } - - // Verify we received the loop messages (should be 10 of them) - var loopMessageCount = 0 - try { - for (i in 0..9) { - testHelper.ensureErrorReceived { event -> - val matches = - event.message?.message?.contains("items we'll wait to flush to Sentry!") == true - if (matches) loopMessageCount++ - matches - } - } - } catch (e: Exception) { - // Some loop messages might be missing, but we should have at least some - } - - assertTrue( - "Should receive at least 5 loop messages, got $loopMessageCount", - loopMessageCount >= 5, - ) - - // Verify we have breadcrumbs - testHelper.ensureErrorReceived { event -> - event.breadcrumbs?.any { breadcrumb -> - breadcrumb.message?.contains("Processed by") == true - } == true - } - - testHelper.ensureMetricsReceived { metricsEvents, sentryEnvelopeHeader -> - testHelper.doesContainMetric(metricsEvents, "countMetric", "counter", 1.0) && - testHelper.doesContainMetric(metricsEvents, "gaugeMetric", "gauge", 5.0) && - testHelper.doesContainMetric(metricsEvents, "distributionMetric", "distribution", 7.0) && - testHelper.doesMetricHaveAttribute(metricsEvents, "countMetric", "user.type", "admin") && - testHelper.doesMetricHaveAttribute(metricsEvents, "countMetric", "feature.version", 2) - } - } -} diff --git a/test/system-test-runner.py b/test/system-test-runner.py index 58821315853..c790caac082 100644 --- a/test/system-test-runner.py +++ b/test/system-test-runner.py @@ -68,9 +68,7 @@ KAFKA_CONTAINER_NAME = "sentry-java-system-test-kafka" KAFKA_BOOTSTRAP_SERVERS = "localhost:9092" -KAFKA_BROKER_REQUIRED_MODULES = { - "sentry-samples-console", -} +KAFKA_BROKER_REQUIRED_MODULES = set() KAFKA_PROFILE_REQUIRED_MODULES = set() class ServerType(Enum): @@ -828,7 +826,6 @@ def get_available_modules(self) -> List[ModuleConfig]: """Get list of all available test modules.""" return [ ModuleConfig("sentry-samples-spring-boot-4-otlp", "false", "true", "false"), - ModuleConfig("sentry-samples-console", "false", "true", "false"), ModuleConfig("sentry-samples-console-otlp", "false", "true", "false"), ModuleConfig("sentry-samples-console-opentelemetry-noagent", "false", "true", "false"), ModuleConfig("sentry-samples-logback", "false", "true", "false"),