Rework pulsar tests (#8000)

instrumentation/apache-pulsar/apache-pulsar-2.8/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/telemetry/PulsarSingletons.java

@@ -24,7 +24,8 @@ import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
 
 public final class PulsarSingletons {
-  private static final String INSTRUMENTATION_NAME = "io.opentelemetry.pulsar-client-2.8";
+  private static final String INSTRUMENTATION_NAME = "io.opentelemetry.apache-pulsar-2.8";
+
   private static final OpenTelemetry TELEMETRY = GlobalOpenTelemetry.get();
   private static final TextMapPropagator PROPAGATOR =
       TELEMETRY.getPropagators().getTextMapPropagator();

instrumentation/apache-pulsar/apache-pulsar-2.8/javaagent/src/test/groovy/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/PulsarClientTest.groovy

@@ -6,7 +6,6 @@
 package io.opentelemetry.javaagent.instrumentation.pulsar.v2_8
 
 import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
-import io.opentelemetry.instrumentation.test.asserts.SpanAssert
 import io.opentelemetry.semconv.trace.attributes.SemanticAttributes
 import org.apache.pulsar.client.admin.PulsarAdmin
 import org.apache.pulsar.client.api.Consumer
@@ -16,11 +15,14 @@ import org.apache.pulsar.client.api.Producer
 import org.apache.pulsar.client.api.PulsarClient
 import org.apache.pulsar.client.api.Schema
 import org.apache.pulsar.client.api.SubscriptionInitialPosition
-import org.junit.Assert
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
 import org.testcontainers.containers.PulsarContainer
+import org.testcontainers.containers.output.Slf4jLogConsumer
 import org.testcontainers.utility.DockerImageName
 import spock.lang.Shared
 
+import java.time.Duration
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 
@@ -29,6 +31,7 @@ import static io.opentelemetry.api.trace.SpanKind.INTERNAL
 import static io.opentelemetry.api.trace.SpanKind.PRODUCER
 
 class PulsarClientTest extends AgentInstrumentationSpecification {
+  private static final Logger logger = LoggerFactory.getLogger(PulsarClientTest)
 
   private static final DockerImageName DEFAULT_IMAGE_NAME =
     DockerImageName.parse("apachepulsar/pulsar:2.8.0")
@@ -52,6 +55,9 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
   @Override
   def setupSpec() {
     pulsar = new PulsarContainer(DEFAULT_IMAGE_NAME)
+      .withEnv("PULSAR_MEM", "-Xmx128m")
+      .withLogConsumer(new Slf4jLogConsumer(logger))
+      .withStartupTimeout(Duration.ofMinutes(2))
     pulsar.start()
 
     brokerUrl = pulsar.pulsarBrokerUrl
@@ -71,44 +77,30 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
 
   def "test send non-partitioned topic"() {
     setup:
-    def topic = "persistent://public/default/testNonPartitionedTopic_" + UUID.randomUUID()
+    def topic = "persistent://public/default/testSendNonPartitionedTopic"
     admin.topics().createNonPartitionedTopic(topic)
     producer =
       client.newProducer(Schema.STRING).topic(topic)
         .enableBatching(false).create()
 
-    String msg = UUID.randomUUID().toString()
+    when:
-
+    String msg = "test"
-    def msgId
+    def msgId = runWithSpan("parent") {
-    runWithSpan("parent") {
+      producer.send(msg)
-      msgId = producer.send(msg)
     }
 
-    def traces = waitForTraces(1)
+    then:
-    Assert.assertEquals(traces.size(), 1)
+    assertTraces(1) {
-    def spans = traces[0]
+      trace(0, 2) {
-    Assert.assertEquals(spans.size(), 2)
+        span(0) {
-    def parent = spans.find {
+          name "parent"
-      it0 ->
+          kind INTERNAL
-        it0.name.equalsIgnoreCase("parent")
-    }
-    def producer = spans.find {
-      it0 ->
-        it0.name.equalsIgnoreCase("PRODUCER/SEND")
-    }
-    Assert.assertNotNull(parent)
-    Assert.assertNotNull(producer)
-
-    SpanAssert.assertSpan(parent) {
-      name("parent")
-      kind(INTERNAL)
           hasNoParent()
         }
-
+        span(1) {
-    SpanAssert.assertSpan(producer) {
+          name "PRODUCER/SEND"
-      name("PRODUCER/SEND")
+          kind PRODUCER
-      kind(PRODUCER)
+          childOf span(0)
-      childOf parent
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
             "$SemanticAttributes.NET_SOCK_PEER_ADDR" brokerUrl
@@ -120,10 +112,12 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
           }
         }
       }
+    }
+  }
 
   def "test consume non-partitioned topic"() {
     setup:
-    def topic = "persistent://public/default/testNonPartitionedTopic_" + UUID.randomUUID()
+    def topic = "persistent://public/default/testConsumeNonPartitionedTopic"
     def latch = new CountDownLatch(1)
     admin.topics().createNonPartitionedTopic(topic)
     consumer = client.newConsumer(Schema.STRING)
@@ -144,47 +138,26 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
       .enableBatching(false)
       .create()
 
-    def msgId
+    when:
-    def msg = UUID.randomUUID().toString()
+    def msg = "test"
-    runWithSpan("parent") {
+    def msgId = runWithSpan("parent") {
-      msgId = producer.send(msg)
+      producer.send(msg)
     }
 
     latch.await(1, TimeUnit.MINUTES)
-    // Wait until all the spans finished.
-    Thread.sleep(TimeUnit.SECONDS.toMillis(20))
 
-    def traces = waitForTraces(1)
+    then:
-    def spans = traces[0]
+    assertTraces(1) {
-    Assert.assertEquals(spans.size(), 4)
+      trace(0, 4) {
-    def parent = spans.find {
+        span(0) {
-      it0 ->
+          name "parent"
-        it0.name.equalsIgnoreCase("parent")
+          kind INTERNAL
-    }
-    def send = spans.find {
-      it0 ->
-        it0.name.equalsIgnoreCase("PRODUCER/SEND")
-    }
-    def receive = spans.find {
-      it0 ->
-        it0.name.equalsIgnoreCase("CONSUMER/RECEIVE")
-    }
-
-    def process = spans.find {
-      it0 ->
-        it0.name.equalsIgnoreCase("CONSUMER/PROCESS")
-    }
-
-    SpanAssert.assertSpan(parent) {
-      name("parent")
-      kind(INTERNAL)
           hasNoParent()
         }
-
+        span(1) {
-    SpanAssert.assertSpan(send) {
+          name "PRODUCER/SEND"
-      name("PRODUCER/SEND")
+          kind PRODUCER
-      kind(PRODUCER)
+          childOf span(0)
-      childOf parent
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
             "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
@@ -195,11 +168,10 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
             "$SemanticAttributes.MESSAGE_TYPE" "NORMAL"
           }
         }
-
+        span(2) {
-    SpanAssert.assertSpan(receive) {
+          name "CONSUMER/RECEIVE"
-      name("CONSUMER/RECEIVE")
+          kind CONSUMER
-      kind(CONSUMER)
+          childOf span(1)
-      childOf(send)
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
             "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
@@ -210,11 +182,10 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
             "$SemanticAttributes.MESSAGING_OPERATION" "receive"
           }
         }
-
+        span(3) {
-    SpanAssert.assertSpan(process) {
+          name "CONSUMER/PROCESS"
-      name("CONSUMER/PROCESS")
+          kind INTERNAL
-      kind(INTERNAL)
+          childOf span(2)
-      childOf(receive)
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
             "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
@@ -225,46 +196,35 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
           }
         }
       }
-
+    }
+  }
 
   def "test send partitioned topic"() {
     setup:
-    def topic = "persistent://public/default/testPartitionedTopic_" + UUID.randomUUID()
+    def topic = "persistent://public/default/testSendPartitionedTopic"
     admin.topics().createPartitionedTopic(topic, 2)
     producer =
       client.newProducer(Schema.STRING).topic(topic)
         .enableBatching(false).create()
 
-    String msg = UUID.randomUUID().toString()
+    when:
-
+    String msg = "test"
-    def msgId
+    def msgId = runWithSpan("parent") {
-    runWithSpan("parent") {
+      producer.send(msg)
-      msgId = producer.send(msg)
     }
 
-    def traces = waitForTraces(1)
+    then:
-    def spans = traces[0]
+    assertTraces(1) {
-    Assert.assertEquals(spans.size(), 2)
+      trace(0, 2) {
-
+        span(0) {
-    def parent = spans.find {
+          name "parent"
-      it0 ->
+          kind INTERNAL
-        it0.name.equalsIgnoreCase("parent")
-    }
-    def send = spans.find {
-      it0 ->
-        it0.name.equalsIgnoreCase("PRODUCER/SEND")
-    }
-
-    SpanAssert.assertSpan(parent) {
-      name("parent")
-      kind(INTERNAL)
           hasNoParent()
         }
-
+        span(1) {
-    SpanAssert.assertSpan(send) {
+          name "PRODUCER/SEND"
-      name("PRODUCER/SEND")
+          kind PRODUCER
-      kind(PRODUCER)
+          childOf span(0)
-      childOf parent
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
             "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
@@ -279,10 +239,12 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
           }
         }
       }
+    }
+  }
 
   def "test consume partitioned topic"() {
     setup:
-    def topic = "persistent://public/default/testPartitionedTopic_" + UUID.randomUUID()
+    def topic = "persistent://public/default/testConsumePartitionedTopic"
     admin.topics().createPartitionedTopic(topic, 2)
 
     def latch = new CountDownLatch(1)
@@ -304,105 +266,73 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
       .enableBatching(false)
       .create()
 
-    def msgId
+    when:
-    def msg = UUID.randomUUID().toString()
+    def msg = "test"
-    runWithSpan("parent") {
+    def msgId = runWithSpan("parent") {
-      msgId = producer.send(msg)
+      producer.send(msg)
     }
 
     latch.await(1, TimeUnit.MINUTES)
-    // Wait until all the spans finished.
-    Thread.sleep(TimeUnit.SECONDS.toMillis(20))
 
-    def traces = waitForTraces(1)
+    then:
-    Assert.assertEquals(traces.size(), 1)
+    assertTraces(1) {
-    def spans = traces[0]
+      trace(0, 4) {
-    Assert.assertEquals(spans.size(), 4)
+        span(0) {
-
+          name "parent"
-    def parent = spans.find {
+          kind INTERNAL
-      it0 ->
-        it0.name.equalsIgnoreCase("parent")
-    }
-    def send = spans.find {
-      it0 ->
-        it0.name.equalsIgnoreCase("PRODUCER/SEND")
-    }
-    def receive = spans.find {
-      it0 ->
-        it0.name.equalsIgnoreCase("CONSUMER/RECEIVE")
-    }
-
-    def process = spans.find {
-      it0 ->
-        it0.name.equalsIgnoreCase("CONSUMER/PROCESS")
-    }
-
-    SpanAssert.assertSpan(parent) {
-      name("parent")
-      kind(INTERNAL)
           hasNoParent()
         }
-
+        span(1) {
-    SpanAssert.assertSpan(send) {
+          name "PRODUCER/SEND"
-      name("PRODUCER/SEND")
+          kind PRODUCER
-      kind(PRODUCER)
+          childOf span(0)
-      childOf parent
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
             "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
             "$SemanticAttributes.NET_SOCK_PEER_ADDR" brokerUrl
-        "$SemanticAttributes.MESSAGING_DESTINATION_NAME" {
+            "$SemanticAttributes.MESSAGING_DESTINATION_NAME" { it.startsWith(topic) }
-          v ->
-            return v.toString().contains(topic)
-        }
             "$SemanticAttributes.MESSAGING_MESSAGE_ID" msgId.toString()
             "$SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES" Long
             "$SemanticAttributes.MESSAGE_TYPE" "NORMAL"
           }
         }
-
+        span(2) {
-    SpanAssert.assertSpan(receive) {
+          name "CONSUMER/RECEIVE"
-      name("CONSUMER/RECEIVE")
+          kind CONSUMER
-      kind(CONSUMER)
+          childOf span(1)
-      childOf(send)
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
             "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
             "$SemanticAttributes.NET_SOCK_PEER_ADDR" brokerUrl
-        "$SemanticAttributes.MESSAGING_DESTINATION_NAME" {
+            "$SemanticAttributes.MESSAGING_DESTINATION_NAME" { it.startsWith(topic) }
-          v ->
-            return v.toString().contains(topic)
-        }
             "$SemanticAttributes.MESSAGING_MESSAGE_ID" msgId.toString()
             "$SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES" Long
             "$SemanticAttributes.MESSAGING_OPERATION" "receive"
           }
         }
-
+        span(3) {
-    SpanAssert.assertSpan(process) {
+          name "CONSUMER/PROCESS"
-      name("CONSUMER/PROCESS")
+          kind INTERNAL
-      kind(INTERNAL)
+          childOf span(2)
-      childOf(receive)
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
             "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
-        "$SemanticAttributes.MESSAGING_DESTINATION_NAME" {
+            "$SemanticAttributes.MESSAGING_DESTINATION_NAME" { it.startsWith(topic) }
-          v ->
-            return v.toString().contains(topic)
-        }
             "$SemanticAttributes.MESSAGING_MESSAGE_ID" msgId.toString()
             "$SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES" Long
             "$SemanticAttributes.MESSAGING_OPERATION" "process"
           }
         }
       }
-
+    }
+  }
 
   def "test consume multi-topics"() {
     setup:
 
-    def topic = "persistent://public/default/testNonPartitionedTopic_" + UUID.randomUUID()
+    def topicNamePrefix = "persistent://public/default/testConsumeMulti_"
-    def topic1 = "persistent://public/default/testNonPartitionedTopic_" + UUID.randomUUID()
+    def topic = topicNamePrefix + "1"
+    def topic1 = topicNamePrefix + "2"
 
     def latch = new CountDownLatch(2)
     producer = client.newProducer(Schema.STRING)
@@ -414,9 +344,12 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
       .enableBatching(false)
       .create()
 
-    runWithSpan("parent") {
+    when:
-      producer.send(UUID.randomUUID().toString())
+    runWithSpan("parent1") {
-      producer1.send(UUID.randomUUID().toString())
+      producer.send("test1")
+    }
+    runWithSpan("parent2") {
+      producer1.send("test2")
     }
 
     consumer = client.newConsumer(Schema.STRING)
@@ -433,73 +366,73 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
       .subscribe()
 
     latch.await(1, TimeUnit.MINUTES)
-    // Wait until all the spans finished.
-    Thread.sleep(TimeUnit.SECONDS.toMillis(20))
 
-    def traces = waitForTraces(1)
+    then:
-    Assert.assertEquals(traces.size(), 1)
+    assertTraces(2) {
-    def spans = traces[0]
+      traces.sort(orderByRootSpanName("parent1", "parent2"))
-    Assert.assertEquals(spans.size(), 7)
+      for (int i in 1..2) {
-
+        trace(i - 1, 4) {
-    def parent = spans.find {
+          span(0) {
-      it0 ->
+            name "parent" + i
-        it0.name.equalsIgnoreCase("parent")
+            kind INTERNAL
-    }
-
-    SpanAssert.assertSpan(parent) {
             hasNoParent()
-      kind(INTERNAL)
           }
-
+          span(1) {
-
+            name "PRODUCER/SEND"
-    def sendSpans = spans.findAll {
+            kind PRODUCER
-      it0 ->
+            childOf span(0)
-        it0.name.equalsIgnoreCase("PRODUCER/SEND")
+            attributes {
-    }
+              "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
-
+              "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
-    sendSpans.forEach {
+              "$SemanticAttributes.NET_SOCK_PEER_ADDR" brokerUrl
-      it0 ->
+              "$SemanticAttributes.MESSAGING_DESTINATION_NAME" { it.startsWith(topicNamePrefix) }
-        SpanAssert.assertSpan(it0) {
+              "$SemanticAttributes.MESSAGING_MESSAGE_ID" String
-          kind(PRODUCER)
+              "$SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES" Long
-          childOf(parent)
+              "$SemanticAttributes.MESSAGE_TYPE" "NORMAL"
             }
           }
-
+          span(1) {
-    def receiveSpans = spans.findAll {
+            name "PRODUCER/SEND"
-      it0 ->
+            kind PRODUCER
-        it0.name.equalsIgnoreCase("CONSUMER/RECEIVE")
+            childOf span(0)
-    }
+            attributes {
-
+              "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
-    def processSpans = spans.findAll {
+              "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
-      it0 ->
+              "$SemanticAttributes.NET_SOCK_PEER_ADDR" brokerUrl
-        it0.name.equalsIgnoreCase("CONSUMER/PROCESS")
+              "$SemanticAttributes.MESSAGING_DESTINATION_NAME" { it.startsWith(topicNamePrefix) }
-    }
+              "$SemanticAttributes.MESSAGING_MESSAGE_ID" String
-
+              "$SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES" Long
-    receiveSpans.forEach {
+              "$SemanticAttributes.MESSAGE_TYPE" "NORMAL"
-      it0 ->
+            }
-        def parentSpanId = it0.getParentSpanId()
+          }
-        def parent0 = sendSpans.find {
+          span(2) {
-          v ->
+            name "CONSUMER/RECEIVE"
-            (v.spanId == parentSpanId)
+            kind CONSUMER
-        }
+            childOf span(1)
-
+            attributes {
-        SpanAssert.assertSpan(it0) {
+              "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
-          kind(CONSUMER)
+              "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
-          childOf(parent0)
+              "$SemanticAttributes.NET_SOCK_PEER_ADDR" brokerUrl
+              "$SemanticAttributes.MESSAGING_DESTINATION_NAME" { it.startsWith(topicNamePrefix) }
+              "$SemanticAttributes.MESSAGING_MESSAGE_ID" String
+              "$SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES" Long
+              "$SemanticAttributes.MESSAGING_OPERATION" "receive"
+            }
+          }
+          span(3) {
+            name "CONSUMER/PROCESS"
+            kind INTERNAL
+            childOf span(2)
+            attributes {
+              "$SemanticAttributes.MESSAGING_SYSTEM" "pulsar"
+              "$SemanticAttributes.MESSAGING_DESTINATION_KIND" "topic"
+              "$SemanticAttributes.MESSAGING_DESTINATION_NAME" { it.startsWith(topicNamePrefix) }
+              "$SemanticAttributes.MESSAGING_MESSAGE_ID" String
+              "$SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES" Long
+              "$SemanticAttributes.MESSAGING_OPERATION" "process"
             }
           }
-
-    processSpans.forEach {
-      it0 ->
-        def parentSpanId = it0.getParentSpanId()
-        def parent0 = receiveSpans.find {
-          v ->
-            (v.spanId == parentSpanId)
         }
-
-        SpanAssert.assertSpan(it0) {
-          kind(INTERNAL)
-          childOf(parent0)
       }
     }
   }