Update Kafka Connect Postgres assertions

Author: trask

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

@@ -6,6 +6,7 @@
 package io.opentelemetry.instrumentation.kafkaconnect.v2_6;
 
 import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
+import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableDatabaseSemconv;
 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;
@@ -18,6 +19,7 @@
 import static io.opentelemetry.semconv.incubating.ThreadIncubatingAttributes.THREAD_NAME;
 import static io.restassured.RestAssured.given;
 import static java.lang.String.format;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
 import io.opentelemetry.api.trace.Span;
@@ -160,8 +162,14 @@ void testSingleMessage() throws Exception {
         trace -> {
           // kafka connect consumer trace, linked to producer span via a span link
           Consumer<SpanDataAssert> selectAssertion =
-              span ->
-                  span.hasName("SELECT test").hasKind(SpanKind.CLIENT).hasParent(trace.getSpan(0));
+              span -> {
+                if (emitStableDatabaseSemconv()) {
+                  span.satisfies(spanData -> assertThat(spanData.getName()).startsWith("SELECT"));
+                } else {
+                  span.hasName("SELECT " + DATABASE_NAME);
+                }
+                span.hasKind(SpanKind.CLIENT).hasParent(trace.getSpan(0));
+              };
 
           trace.hasSpansSatisfyingExactly(
               span ->
@@ -182,7 +190,10 @@ void testSingleMessage() throws Exception {
               selectAssertion,
               selectAssertion,
               span ->
-                  span.hasName("INSERT test." + DB_TABLE_PERSON)
+                  span.hasName(
+                          emitStableDatabaseSemconv()
+                              ? "INSERT \"" + DB_TABLE_PERSON + "\""
+                              : "INSERT " + DATABASE_NAME + "." + DB_TABLE_PERSON)
                       .hasKind(SpanKind.CLIENT)
                       .hasParent(trace.getSpan(0)));
         },
@@ -285,8 +296,14 @@ void testMultiTopic() throws Exception {
         trace -> {
           // kafka connect consumer trace, linked to producer span via a span link
           Consumer<SpanDataAssert> selectAssertion =
-              span ->
-                  span.hasName("SELECT test").hasKind(SpanKind.CLIENT).hasParent(trace.getSpan(0));
+              span -> {
+                if (emitStableDatabaseSemconv()) {
+                  span.satisfies(spanData -> assertThat(spanData.getName()).startsWith("SELECT"));
+                } else {
+                  span.hasName("SELECT " + DATABASE_NAME);
+                }
+                span.hasKind(SpanKind.CLIENT).hasParent(trace.getSpan(0));
+              };
 
           trace.hasSpansSatisfyingExactly(
               span ->
@@ -309,7 +326,10 @@ void testMultiTopic() throws Exception {
               selectAssertion,
               selectAssertion,
               span ->
-                  span.hasName("INSERT test." + DB_TABLE_PERSON)
+                  span.hasName(
+                          emitStableDatabaseSemconv()
+                              ? "BATCH INSERT \"" + DB_TABLE_PERSON + "\""
+                              : "INSERT " + DATABASE_NAME + "." + DB_TABLE_PERSON)
                       .hasKind(SpanKind.CLIENT)
                       .hasParent(trace.getSpan(0)));
         },