fix

Author: zeitlinger

instrumentation/kafka/kafka-connect-2.6/testing/src/test/java/io/opentelemetry/instrumentation/kafkaconnect/v2_6/MongoKafkaConnectSinkTaskTest.java

@@ -7,12 +7,15 @@
 
 import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_BATCH_MESSAGE_COUNT;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MessagingOperationTypeIncubatingValues.PROCESS;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MessagingSystemIncubatingValues.KAFKA;
+import static io.opentelemetry.semconv.incubating.ThreadIncubatingAttributes.THREAD_ID;
+import static io.opentelemetry.semconv.incubating.ThreadIncubatingAttributes.THREAD_NAME;
 import static io.restassured.RestAssured.given;
 import static java.lang.String.format;
 import static org.awaitility.Awaitility.await;
@@ -152,7 +155,9 @@ void testSingleMessage() throws Exception {
                             equalTo(MESSAGING_BATCH_MESSAGE_COUNT, 1),
                             equalTo(MESSAGING_DESTINATION_NAME, testTopicName),
                             equalTo(MESSAGING_OPERATION, PROCESS),
-                            equalTo(MESSAGING_SYSTEM, KAFKA)),
+                            equalTo(MESSAGING_SYSTEM, KAFKA),
+                            satisfies(THREAD_ID, val -> val.isNotZero()),
+                            satisfies(THREAD_NAME, val -> val.isNotBlank())),
                 span ->
                     span.hasName("update " + DB_NAME + "." + COLLECTION_NAME)
                         .hasKind(SpanKind.CLIENT)
@@ -261,7 +266,9 @@ void testMultiTopic() throws Exception {
                         .hasAttributesSatisfyingExactly(
                             equalTo(MESSAGING_BATCH_MESSAGE_COUNT, 3),
                             equalTo(MESSAGING_OPERATION, PROCESS),
-                            equalTo(MESSAGING_SYSTEM, KAFKA)),
+                            equalTo(MESSAGING_SYSTEM, KAFKA),
+                            satisfies(THREAD_ID, val -> val.isNotZero()),
+                            satisfies(THREAD_NAME, val -> val.isNotBlank())),
                 span ->
                     span.hasName("update " + DB_NAME + "." + COLLECTION_NAME)
                         .hasKind(SpanKind.CLIENT)

instrumentation/kafka/kafka-connect-2.6/testing/src/test/java/io/opentelemetry/instrumentation/kafkaconnect/v2_6/PostgresKafkaConnectSinkTaskTest.java

@@ -7,12 +7,15 @@
 
 import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_BATCH_MESSAGE_COUNT;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MessagingOperationTypeIncubatingValues.PROCESS;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MessagingSystemIncubatingValues.KAFKA;
+import static io.opentelemetry.semconv.incubating.ThreadIncubatingAttributes.THREAD_ID;
+import static io.opentelemetry.semconv.incubating.ThreadIncubatingAttributes.THREAD_NAME;
 import static io.restassured.RestAssured.given;
 import static java.lang.String.format;
 import static org.awaitility.Awaitility.await;
@@ -172,7 +175,9 @@ void testSingleMessage() throws Exception {
                           equalTo(MESSAGING_BATCH_MESSAGE_COUNT, 1),
                           equalTo(MESSAGING_DESTINATION_NAME, testTopicName),
                           equalTo(MESSAGING_OPERATION, PROCESS),
-                          equalTo(MESSAGING_SYSTEM, KAFKA)),
+                          equalTo(MESSAGING_SYSTEM, KAFKA),
+                          satisfies(THREAD_ID, val -> val.isNotZero()),
+                          satisfies(THREAD_NAME, val -> val.isNotBlank())),
               selectAssertion,
               selectAssertion,
               selectAssertion,
@@ -297,7 +302,9 @@ void testMultiTopic() throws Exception {
                       .hasAttributesSatisfyingExactly(
                           equalTo(MESSAGING_BATCH_MESSAGE_COUNT, 3),
                           equalTo(MESSAGING_OPERATION, PROCESS),
-                          equalTo(MESSAGING_SYSTEM, KAFKA)),
+                          equalTo(MESSAGING_SYSTEM, KAFKA),
+                          satisfies(THREAD_ID, val -> val.isNotZero()),
+                          satisfies(THREAD_NAME, val -> val.isNotBlank())),
               selectAssertion,
               selectAssertion,
               selectAssertion,