Fix Race in Micrometer Test

When timers have dynamic tags, creation of the timers is deferred until
the first completion of a timer with a set of tag values.

Use awaitility to wait for the timer to be created before checking the
count.

**Cherry-pick to `2.9.x`**

Author: garyrussell

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -18,6 +18,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -166,6 +167,8 @@
 
 import io.micrometer.core.instrument.ImmutableTag;
 import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+import io.micrometer.core.instrument.search.MeterNotFoundException;
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 
 /**
@@ -563,23 +566,41 @@ public void testNulls() throws Exception {
 		this.kafkaJsonTemplate.send("annotated12", null, null);
 		assertThat(this.listener.latch8.await(60, TimeUnit.SECONDS)).isTrue();
 
-		assertThat(this.meterRegistry.get("spring.kafka.template")
-				.tag("name", "kafkaJsonTemplate")
-				.tag("extraTag", "bar")
-				.tag("topic", "annotated12")
-				.tag("result", "success")
-				.timer()
-				.count())
-						.isGreaterThan(0L);
-
-		assertThat(this.meterRegistry.get("spring.kafka.listener")
-				.tag("name", "quux-0")
-				.tag("extraTag", "foo")
-				.tag("topic", "annotated12")
-				.tag("result", "success")
-				.timer()
-				.count())
-						.isGreaterThan(0L);
+		await().untilAsserted(() -> {
+			Timer timer = null;
+			try {
+				timer = this.meterRegistry.get("spring.kafka.template")
+						.tag("name", "kafkaJsonTemplate")
+						.tag("extraTag", "bar")
+						.tag("topic", "annotated12")
+						.tag("result", "success")
+						.timer();
+			}
+			catch (MeterNotFoundException ex) {
+			}
+			assertThat(timer)
+					.describedAs("Timer not found in " + ((SimpleMeterRegistry) this.meterRegistry).getMetersAsString())
+					.isNotNull();
+			assertThat(timer.count()).isGreaterThan(0L);
+		});
+
+		await().untilAsserted(() -> {
+			Timer timer = null;
+			try {
+				timer = this.meterRegistry.get("spring.kafka.listener")
+						.tag("name", "quux-0")
+						.tag("extraTag", "foo")
+						.tag("topic", "annotated12")
+						.tag("result", "success")
+						.timer();
+			}
+			catch (MeterNotFoundException ex) {
+			}
+			assertThat(timer)
+					.describedAs("Timer not found in " + ((SimpleMeterRegistry) this.meterRegistry).getMetersAsString())
+					.isNotNull();
+			assertThat(timer.count()).isGreaterThan(0L);
+		});
 	}
 
 	@Test