Use log assertion helper in Kafka test

Author: trask

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/WrapperTest.java

@@ -26,7 +26,6 @@
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Arrays.asList;
 import static java.util.Collections.singletonList;
-import static java.util.stream.Collectors.toList;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.Mockito.mock;
@@ -36,7 +35,6 @@
 import io.opentelemetry.api.trace.SpanContext;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.instrumentation.testing.junit.message.MessageHeaderUtil;
-import io.opentelemetry.sdk.logs.data.LogRecordData;
 import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
 import io.opentelemetry.sdk.trace.data.LinkData;
 import io.opentelemetry.sdk.trace.data.StatusData;
@@ -47,7 +45,6 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.assertj.core.api.AbstractLongAssert;
 import org.assertj.core.api.AbstractStringAssert;
-import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Test;
 
 @SuppressWarnings("deprecation") // using deprecated semconv
@@ -202,26 +199,15 @@ void testConsumerError() {
                 }));
 
     if (emitExceptionAsLogs()) {
-      Awaitility.await()
-          .untilAsserted(
-              () -> {
-                List<LogRecordData> logs =
-                    testing.logRecords().stream()
-                        .filter(log -> "messaging.receive.exception".equals(log.getEventName()))
-                        .collect(toList());
-                assertThat(logs).hasSize(1);
-                assertThat(logs.get(0))
-                    .hasSeverity(Severity.WARN)
-                    .hasEventName("messaging.receive.exception")
-                    .hasAttributesSatisfyingExactly(
-                        equalTo(EXCEPTION_TYPE, error.getClass().getName()),
-                        equalTo(EXCEPTION_MESSAGE, error.getMessage()),
-                        satisfies(
-                            EXCEPTION_STACKTRACE,
-                            val ->
-                                val.contains(error.getClass().getName())
-                                    .contains("WrapperTest.testConsumerError")));
-              });
+      testing.waitAndAssertLogRecords(
+          logRecord ->
+              logRecord
+                  .hasSeverity(Severity.WARN)
+                  .hasEventName("messaging.receive.exception")
+                  .hasAttributesSatisfyingExactly(
+                      equalTo(EXCEPTION_TYPE, error.getClass().getName()),
+                      equalTo(EXCEPTION_MESSAGE, error.getMessage()),
+                      satisfies(EXCEPTION_STACKTRACE, val -> val.isNotNull())));
     }
   }
 }