GH-2943: Fix KT.clusterId for concurrency (#2944)

Fixes: #2943

The `if (this.kafkaAdmin != null && this.clusterId == null) {` condition
might be always true for concurrent threads (especially virtual).
Therefore, all of those threads are calling `this.kafkaAdmin.clusterId()`
making unnecessary network chats to Kafka broker

* Surround `this.kafkaAdmin.clusterId()` call with `Lock`

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

Author: artembilan

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

@@ -28,6 +28,8 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
 
 import org.apache.commons.logging.LogFactory;
@@ -117,6 +119,8 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
 
 	private final Map<String, String> micrometerTags = new HashMap<>();
 
+	private final Lock clusterIdLock = new ReentrantLock();
+
 	private String beanName = "kafkaTemplate";
 
 	private ApplicationContext applicationContext;
@@ -500,7 +504,15 @@ else if (this.micrometerEnabled) {
 	@Nullable
 	private String clusterId() {
 		if (this.kafkaAdmin != null && this.clusterId == null) {
-			this.clusterId = this.kafkaAdmin.clusterId();
+			this.clusterIdLock.lock();
+			try {
+				if (this.clusterId == null) {
+					this.clusterId = this.kafkaAdmin.clusterId();
+				}
+			}
+			finally {
+				this.clusterIdLock.unlock();
+			}
 		}
 		return this.clusterId;
 	}