GH-2538: KafkaTemplate - Fix Eager Init Cluster Id

Resolves #2538

The `KafkaTemplate` already has code to lazily fetch the cluster id
when first needed. Remove the eager fetch in `afterSingletonsInstantiated`.

Also, when creating a new `KafkaAdmin` propagate the `operationTimeout`
from the source admin (in both the template and listener container).

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2022 the original author or authors.
+ * Copyright 2017-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -132,6 +132,15 @@ public void setOperationTimeout(int operationTimeout) {
 		this.operationTimeout = operationTimeout;
 	}
 
+	/**
+	 * Return the operation timeout in seconds.
+	 * @return the timeout.
+	 * @since 3.0.2
+	 */
+	public int getOperationTimeout() {
+		return this.operationTimeout;
+	}
+
 	/**
 	 * Set to true if you want the application context to fail to load if we are unable
 	 * to connect to the broker during initialization, to check/add topics.

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2022 the original author or authors.
+ * Copyright 2015-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -434,9 +434,10 @@ public void afterSingletonsInstantiated() {
 				if (!producerServers.equals(adminServers)) {
 					Map<String, Object> props = new HashMap<>(this.kafkaAdmin.getConfigurationProperties());
 					props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, producerServers);
+					int opTo = this.kafkaAdmin.getOperationTimeout();
 					this.kafkaAdmin = new KafkaAdmin(props);
+					this.kafkaAdmin.setOperationTimeout(opTo);
 				}
-				this.clusterId = this.kafkaAdmin.clusterId();
 			}
 		}
 		else if (this.micrometerEnabled) {

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -945,7 +945,9 @@ private KafkaAdmin obtainAdmin() {
 						Map<String, Object> props = new HashMap<>(admin.getConfigurationProperties());
 						if (!props.get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG).equals(this.bootstrapServers)) {
 							props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
+							int opTo = admin.getOperationTimeout();
 							admin = new KafkaAdmin(props);
+							admin.setOperationTimeout(opTo);
 						}
 					}
 					return admin;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -103,12 +103,14 @@ void endToEnd(@Autowired Listener listener, @Autowired KafkaTemplate<Integer, St
 		SimpleSpan span = spans.poll();
 		assertThat(span.getTags()).containsEntry("spring.kafka.template.name", "template");
 		assertThat(span.getName()).isEqualTo("observation.testT1 send");
+		assertThat(span.getRemoteServiceName()).startsWith("Apache Kafka: ");
 		await().until(() -> spans.peekFirst().getTags().size() == 3);
 		span = spans.poll();
 		assertThat(span.getTags())
 				.containsAllEntriesOf(
 						Map.of("spring.kafka.listener.id", "obs1-0", "foo", "some foo value", "bar", "some bar value"));
 		assertThat(span.getName()).isEqualTo("observation.testT1 receive");
+		assertThat(span.getRemoteServiceName()).startsWith("Apache Kafka: ");
 		await().until(() -> spans.peekFirst().getTags().size() == 1);
 		span = spans.poll();
 		assertThat(span.getTags()).containsEntry("spring.kafka.template.name", "template");
@@ -179,24 +181,27 @@ public KeyValues getLowCardinalityKeyValues(KafkaRecordReceiverContext context)
 		assertThat(admin.getConfigurationProperties())
 				.containsEntry(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString());
 		// producer factory broker different to admin
-		assertThat(
-				KafkaTestUtils.getPropertyValue(template, "kafkaAdmin", KafkaAdmin.class).getConfigurationProperties())
-						.containsEntry(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
-								broker.getBrokersAsString() + "," + broker.getBrokersAsString());
+		KafkaAdmin pAdmin = KafkaTestUtils.getPropertyValue(template, "kafkaAdmin", KafkaAdmin.class);
+		assertThat(pAdmin.getOperationTimeout()).isEqualTo(admin.getOperationTimeout());
+		assertThat(pAdmin.getConfigurationProperties())
+				.containsEntry(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
+						broker.getBrokersAsString() + "," + broker.getBrokersAsString());
 		Object container = KafkaTestUtils
 				.getPropertyValue(endpointRegistry.getListenerContainer("obs1"), "containers", List.class).get(0);
 		// consumer factory broker different to admin
-		assertThat(KafkaTestUtils.getPropertyValue(container, "listenerConsumer.kafkaAdmin", KafkaAdmin.class)
-				.getConfigurationProperties())
-						.containsEntry(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
-								broker.getBrokersAsString() + "," + broker.getBrokersAsString() + ","
-										+ broker.getBrokersAsString());
+		KafkaAdmin cAdmin = KafkaTestUtils.getPropertyValue(container, "listenerConsumer.kafkaAdmin", KafkaAdmin.class);
+		assertThat(cAdmin.getOperationTimeout()).isEqualTo(admin.getOperationTimeout());
+		assertThat(cAdmin.getConfigurationProperties())
+				.containsEntry(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
+						broker.getBrokersAsString() + "," + broker.getBrokersAsString() + ","
+								+ broker.getBrokersAsString());
 		// broker override in annotation
 		container = KafkaTestUtils
 				.getPropertyValue(endpointRegistry.getListenerContainer("obs2"), "containers", List.class).get(0);
-		assertThat(KafkaTestUtils.getPropertyValue(container, "listenerConsumer.kafkaAdmin", KafkaAdmin.class)
-				.getConfigurationProperties())
-						.containsEntry(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString());
+		cAdmin = KafkaTestUtils.getPropertyValue(container, "listenerConsumer.kafkaAdmin", KafkaAdmin.class);
+		assertThat(cAdmin.getOperationTimeout()).isEqualTo(admin.getOperationTimeout());
+		assertThat(cAdmin.getConfigurationProperties())
+				.containsEntry(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString());
 	}
 
 	@Configuration
@@ -205,7 +210,10 @@ public static class Config {
 
 		@Bean
 		KafkaAdmin admin(EmbeddedKafkaBroker broker) {
-			return new KafkaAdmin(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString()));
+			KafkaAdmin admin = new KafkaAdmin(
+					Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString()));
+			admin.setOperationTimeout(42);
+			return admin;
 		}
 
 		@Bean