Review fixes for spring-kafka-2.7:testing (#17736)

Co-authored-by: otelbot <197425009+otelbot@users.noreply.github.com>

Author: otelbot[bot]

instrumentation/spring/spring-kafka-2.7/testing/src/main/java/io/opentelemetry/testing/AbstractSpringKafkaTest.java

@@ -106,7 +106,7 @@ void tearDownApp() {
                   "send",
                   MethodType.methodType(
                       listenableFutureClass, String.class, Object.class, Object.class));
-    } catch (ClassNotFoundException | NoSuchMethodException e) {
+    } catch (ClassNotFoundException | NoSuchMethodException ignored) {
       // spring-kafka 3.0 changed the return type
       try {
         sendMethod =
@@ -131,8 +131,8 @@ void tearDownApp() {
   protected void send(String topic, String key, String data) {
     try {
       send.invoke(kafkaTemplate, topic, key, data);
-    } catch (Throwable e) {
-      throw new AssertionError(e);
+    } catch (Throwable t) {
+      throw new AssertionError(t);
     }
   }
 
@@ -143,7 +143,7 @@ protected void sendBatchMessages(Map<String, String> keyToData) throws Interrupt
     // a batch.
     int maxAttempts = 5;
     for (int i = 1; i <= maxAttempts; i++) {
-      BatchRecordListener.reset();
+      BatchRecordListener.reset(keyToData.size());
 
       testing()
           .runWithSpan(
@@ -157,7 +157,7 @@ protected void sendBatchMessages(Map<String, String> keyToData) throws Interrupt
               });
 
       BatchRecordListener.waitForMessages();
-      if (BatchRecordListener.getLastBatchSize() == 2) {
+      if (BatchRecordListener.getLastBatchSize() == keyToData.size()) {
         break;
       } else if (i < maxAttempts) {
         testing().waitForTraces(2);
@@ -166,6 +166,7 @@ protected void sendBatchMessages(Map<String, String> keyToData) throws Interrupt
         logger.info("Messages weren't received as batch, retrying");
       }
     }
+    assertThat(BatchRecordListener.getLastBatchSize()).isEqualTo(keyToData.size());
   }
 
   protected static Consumer<List<? extends LinkData>> links(SpanContext... spanContexts) {

instrumentation/spring/spring-kafka-2.7/testing/src/main/java/io/opentelemetry/testing/BatchRecordListener.java

@@ -6,6 +6,7 @@
 package io.opentelemetry.testing;
 
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.assertj.core.api.Assertions.assertThat;
 
 import io.opentelemetry.instrumentation.testing.GlobalTraceUtil;
 import java.util.List;
@@ -39,13 +40,13 @@ record -> {
         });
   }
 
-  public static void reset() {
-    messageReceived = new CountDownLatch(2);
+  public static void reset(int expectedMessages) {
+    messageReceived = new CountDownLatch(expectedMessages);
     lastBatchSize.set(0);
   }
 
   public static void waitForMessages() throws InterruptedException {
-    messageReceived.await(30, SECONDS);
+    assertThat(messageReceived.await(30, SECONDS)).isTrue();
   }
 
   public static int getLastBatchSize() {

instrumentation/spring/spring-kafka-2.7/testing/src/main/java/io/opentelemetry/testing/ConsumerConfig.java

@@ -72,7 +72,7 @@ public ConcurrentKafkaListenerContainerFactory<String, String> singleFactory(
       // available since spring 2.8
       Class.forName("org.springframework.kafka.listener.CommonErrorHandler");
       ConsumerConfigUtil.addErrorHandler(factory);
-    } catch (ClassNotFoundException e) {
+    } catch (ClassNotFoundException ignored) {
       // ignore
     }
     customizerProvider.ifAvailable(factory::setContainerCustomizer);