Wait for kafka to start
MODIFY
instrumentation/kafka/kafka-clients/kafka-clients-0.11/testing/src/main/groovy/io/opentelemetry/instrumentation/kafkaclients/KafkaClientBaseTest.groovy
instrumentation/kafka/kafka-clients/kafka-clients-0.11/testing/src/main/groovy/io/opentelemetry/instrumentation/kafkaclients/KafkaClientBaseTest.groovy
package io.opentelemetry.instrumentation.kafkaclients
import io.opentelemetry.instrumentation.test.InstrumentationSpecification
+import java.time.Duration
import org.apache.kafka.clients.admin.AdminClient
import org.apache.kafka.clients.admin.NewTopic
import org.apache.kafka.clients.consumer.Consumer
@@ -18,13 +19,18 @@ import org.apache.kafka.common.serialization.IntegerDeserializer
import org.apache.kafka.common.serialization.IntegerSerializer
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.kafka.common.serialization.StringSerializer
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
import org.testcontainers.containers.KafkaContainer
+import org.testcontainers.containers.output.Slf4jLogConsumer
+import org.testcontainers.containers.wait.strategy.Wait
import spock.lang.Shared
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
abstract class KafkaClientBaseTest extends InstrumentationSpecification {
+ private static final Logger logger = LoggerFactory.getLogger(KafkaClientBaseTest)
protected static final SHARED_TOPIC = "shared.topic"
@@ -41,6 +47,9 @@ abstract class KafkaClientBaseTest extends InstrumentationSpecification {
def setupSpec() {
kafka = new KafkaContainer()
+ .withLogConsumer(new Slf4jLogConsumer(logger))
+ .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
+ .withStartupTimeout(Duration.ofMinutes(1))
kafka.start()
// create test topic

MODIFY
instrumentation/kafka/kafka-streams-0.11/javaagent/src/test/groovy/KafkaStreamsBaseTest.groovy
instrumentation/kafka/kafka-streams-0.11/javaagent/src/test/groovy/KafkaStreamsBaseTest.groovy
*/
import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
+import java.time.Duration
import org.apache.kafka.clients.admin.AdminClient
import org.apache.kafka.clients.admin.NewTopic
import org.apache.kafka.clients.consumer.Consumer
@@ -16,13 +17,18 @@ import org.apache.kafka.common.serialization.IntegerDeserializer
import org.apache.kafka.common.serialization.IntegerSerializer
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.kafka.common.serialization.StringSerializer
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
import org.testcontainers.containers.KafkaContainer
+import org.testcontainers.containers.output.Slf4jLogConsumer
+import org.testcontainers.containers.wait.strategy.Wait
import spock.lang.Shared
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
class KafkaStreamsBaseTest extends AgentInstrumentationSpecification {
+ private static final Logger logger = LoggerFactory.getLogger("io.opentelemetry.KafkaStreamsBaseTest")
protected static final STREAM_PENDING = "test.pending"
protected static final STREAM_PROCESSED = "test.processed"
@@ -38,6 +44,9 @@ class KafkaStreamsBaseTest extends AgentInstrumentationSpecification {
def setupSpec() {
kafka = new KafkaContainer()
+ .withLogConsumer(new Slf4jLogConsumer(logger))
+ .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
+ .withStartupTimeout(Duration.ofMinutes(1))
kafka.start()
// create test topic

MODIFY
instrumentation/spring/spring-kafka-2.7/javaagent/src/test/groovy/SpringKafkaInstrumentationTest.groovy
instrumentation/spring/spring-kafka-2.7/javaagent/src/test/groovy/SpringKafkaInstrumentationTest.groovy
@@ -7,6 +7,7 @@ import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
import io.opentelemetry.instrumentation.testing.GlobalTraceUtil
import io.opentelemetry.sdk.trace.data.SpanData
import io.opentelemetry.semconv.trace.attributes.SemanticAttributes
+import java.time.Duration
import org.apache.kafka.clients.admin.NewTopic
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.springframework.boot.SpringApplication
@@ -20,6 +21,7 @@ import org.springframework.kafka.config.TopicBuilder
import org.springframework.kafka.core.ConsumerFactory
import org.springframework.kafka.core.KafkaTemplate
import org.testcontainers.containers.KafkaContainer
+import org.testcontainers.containers.wait.strategy.Wait
import spock.lang.Shared
import static io.opentelemetry.api.trace.SpanKind.CONSUMER
@@ -34,6 +36,8 @@ class SpringKafkaInstrumentationTest extends AgentInstrumentationSpecification {
def setupSpec() {
kafka = new KafkaContainer()
+ .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
+ .withStartupTimeout(Duration.ofMinutes(1))
kafka.start()
def app = new SpringApplication(ConsumerConfig)

Wait for kafka to start (#5341)