GH-2148: Fix MeterRegistry Detection

Resolves #2148

- use Boot's `AutoConfiguredCompositeMeterRegistry`, if only one
  (tested with a Boot application)
- use `@Primary` registry bean

**cherry-pick to 2.8.x**

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/MicrometerHolder.java

@@ -16,10 +16,16 @@
 
 package org.springframework.kafka.support.micrometer;
 
+import java.util.Collections;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
 
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.context.ApplicationContext;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.lang.Nullable;
 
 import io.micrometer.core.instrument.MeterRegistry;
@@ -69,12 +75,56 @@ public MicrometerHolder(@Nullable ApplicationContext context, String name,
 		this.timerDesc = timerDesc;
 		this.name = name;
 		this.tags = tags;
+		registries = filterRegistries(registries, context);
 		if (registries.size() == 1) {
 			this.registry = registries.values().iterator().next();
 			buildTimer(NONE_EXCEPTION_METERS_KEY);
 		}
 		else {
-			throw new IllegalStateException("No micrometer registry present (or more than one)");
+			throw new IllegalStateException("No micrometer registry present (or more than one and "
+					+ "none marked @Primary)");
+		}
+	}
+
+	private Map<String, MeterRegistry> filterRegistries(Map<String, MeterRegistry> registries,
+			ApplicationContext context) {
+
+		if (registries.size() == 1) {
+			return registries;
+		}
+		Map<String, MeterRegistry> filtered = registries.entrySet()
+				.stream()
+				.filter(entry -> entry.getValue()
+						.getClass()
+						.getName()
+						.equals("org.springframework.boot.actuate.autoconfigure.metrics."
+								+ "AutoConfiguredCompositeMeterRegistry"))
+						.collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue()));
+		if (filtered.size() == 1) {
+			return filtered;
+		}
+		MeterRegistry primary = null;
+		if (context instanceof ConfigurableApplicationContext) {
+			BeanDefinitionRegistry bdr = (BeanDefinitionRegistry) ((ConfigurableApplicationContext) context)
+					.getBeanFactory();
+			for (Entry<String, MeterRegistry> entry : registries.entrySet()) {
+				BeanDefinition beanDefinition = bdr.getBeanDefinition(entry.getKey());
+				if (beanDefinition.isPrimary()) {
+					if (primary != null) {
+						primary = null;
+						break;
+					}
+					else {
+						primary = entry.getValue();
+					}
+				}
+			}
+		}
+		if (primary != null) {
+			return Collections.singletonMap("primary", primary);
+		}
+		else {
+			return registries;
 		}
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/MicrometerHolderTests.java

@@ -17,6 +17,7 @@
 package org.springframework.kafka.support.micrometer;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.BDDMockito.given;
@@ -31,6 +32,9 @@
 import org.junit.jupiter.api.Test;
 
 import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Primary;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import io.micrometer.core.instrument.MeterRegistry;
@@ -44,11 +48,12 @@ public class MicrometerHolderTests {
 
 	@SuppressWarnings("unchecked")
 	@Test
-	public void testMicrometerHolderRecordSuccessWorksGracefullyAfterDestroy() {
+	void testMicrometerHolderRecordSuccessWorksGracefullyAfterDestroy() {
 		MeterRegistry meterRegistry = new SimpleMeterRegistry();
 		ApplicationContext ctx = mock(ApplicationContext.class);
 		Timer.Sample sample = mock(Timer.Sample.class);
-		given(ctx.getBeansOfType(any(), anyBoolean(), anyBoolean())).willReturn(Collections.singletonMap("registry", meterRegistry));
+		given(ctx.getBeansOfType(any(), anyBoolean(), anyBoolean()))
+				.willReturn(Collections.singletonMap("registry", meterRegistry));
 
 		MicrometerHolder micrometerHolder = new MicrometerHolder(ctx, "holderName",
 				"timerName", "timerDesc", Collections.emptyMap());
@@ -68,4 +73,70 @@ public void testMicrometerHolderRecordSuccessWorksGracefullyAfterDestroy() {
 		verifyNoMoreInteractions(ctx, sample);
 	}
 
+	@Test
+	void multiReg() {
+		assertThatIllegalStateException().isThrownBy(() -> new MicrometerHolder(
+				new AnnotationConfigApplicationContext(Config1.class), "", "", "", Collections.emptyMap()));
+	}
+
+	@Test
+	void twoPrimaries() {
+		assertThatIllegalStateException().isThrownBy(() -> new MicrometerHolder(
+				new AnnotationConfigApplicationContext(Config2.class), "", "", "", Collections.emptyMap()));
+	}
+
+	@Test
+	void primary() {
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(Config3.class);
+		MicrometerHolder micrometerHolder = new MicrometerHolder(ctx, "holderName",
+				"timerName", "timerDesc", Collections.emptyMap());
+		Map<String, Timer> meters = (Map<String, Timer>) ReflectionTestUtils.getField(micrometerHolder, "meters");
+		assertThat(meters).hasSize(1);
+	}
+
+	static class Config1 {
+
+		@Bean
+		MeterRegistry reg1() {
+			return new SimpleMeterRegistry();
+		}
+
+		@Bean
+		MeterRegistry reg2() {
+			return new SimpleMeterRegistry();
+		}
+
+	}
+
+	static class Config2 {
+
+		@Bean
+		@Primary
+		MeterRegistry reg1() {
+			return new SimpleMeterRegistry();
+		}
+
+		@Bean
+		@Primary
+		MeterRegistry reg2() {
+			return new SimpleMeterRegistry();
+		}
+
+	}
+
+	static class Config3 {
+
+		@Bean
+		@Primary
+		MeterRegistry reg1() {
+			return new SimpleMeterRegistry();
+		}
+
+		@Bean
+		MeterRegistry reg2() {
+			return new SimpleMeterRegistry();
+		}
+
+	}
+
 }