GH-2859: Option to Manually Set the Cluster ID

Resolves #2859

* Suppress fetching cluster id in `initialize()` if set manually.

Author: garyrussell

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/micrometer.adoc

@@ -109,3 +109,6 @@ Starting with version 3.0.6, you can add dynamic tags to the timers and traces,
 To do so, add a custom `KafkaListenerObservationConvention` and/or `KafkaTemplateObservationConvention` to the listener container properties or `KafkaTemplate` respectively.
 The `record` property in both observation contexts contains the `ConsumerRecord` or `ProducerRecord` respectively.
 
+The sender and receiver contexts' `remoteServiceName` properties are set to the Kafka `clusterId` property; this is retrieved by a `KafkaAdmin`.
+If, for some reason - perhaps lack of admin permissions, you cannot retrieve the cluster id, starting with version 3.1, you can set a manual `clusterId` on the `KafkaAdmin` and inject it into `KafkaTemplate` s and listener containers.
+When it is `null` (default), the admin will invoke the `describeCluster` admin operation to retrieve it from the broker.

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

@@ -198,6 +198,16 @@ protected Predicate<NewTopic> getCreateOrModifyTopic() {
 		return this.createOrModifyTopic;
 	}
 
+	/**
+	 * Set the cluster id. Use this to prevent attempting to fetch the cluster id
+	 * from the broker, perhaps if the user does not have admin permissions.
+	 * @param clusterId the clusterId to set
+	 * @since 3.1
+	 */
+	public void setClusterId(String clusterId) {
+		this.clusterId = clusterId;
+	}
+
 	@Override
 	public Map<String, Object> getConfigurationProperties() {
 		Map<String, Object> configs2 = new HashMap<>(this.configs);
@@ -240,8 +250,10 @@ public final boolean initialize() {
 			if (adminClient != null) {
 				try {
 					synchronized (this) {
-						this.clusterId = adminClient.describeCluster().clusterId().get(this.operationTimeout,
-								TimeUnit.SECONDS);
+						if (this.clusterId != null) {
+							this.clusterId = adminClient.describeCluster().clusterId().get(this.operationTimeout,
+									TimeUnit.SECONDS);
+						}
 					}
 					addOrModifyTopicsIfNeeded(adminClient, newTopics);
 					return true;