spring-projectsGH-3402: kafkaAdmin clusterId configuration is ignored…

… if observability is enabled and bootstrap supplier is not set fixes spring-projectsGH-3402 (spring-projects#3402) * Re-set clusterId after creating new KafkaAdmin
varmenise · Aug 2, 2024 · f18ba42 · f18ba42
1 parent 20696f2
commit f18ba42
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 1 deletion.
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java
@@ -213,6 +213,14 @@ public void setClusterId(String clusterId) {
 		this.clusterId = clusterId;
 	}
 
+	/**
+	 * Get the clusterId property.
+	 * @since 3.3.0
+	 */
+	public String getClusterId() {
+		return clusterId;
+	}
+
 	@Override
 	public Map<String, Object> getConfigurationProperties() {
 		Map<String, Object> configs2 = new HashMap<>(this.configs);

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java
@@ -485,13 +485,17 @@ public void afterSingletonsInstantiated() {
 				if (this.kafkaAdmin != null) {
 					Object producerServers = this.producerFactory.getConfigurationProperties()
 							.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG);
-					String adminServers = this.kafkaAdmin.getBootstrapServers();
+					String adminServers = getAdminBootstrapAddress();
 					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();
+						String clusterId = this.kafkaAdmin.getClusterId();
 						this.kafkaAdmin = new KafkaAdmin(props);
 						this.kafkaAdmin.setOperationTimeout(opTo);
+						if (clusterId != null && !clusterId.isEmpty()) {
+							this.kafkaAdmin.setClusterId(clusterId);
+						}
 					}
 				}
 			}
@@ -501,6 +505,21 @@ else if (this.micrometerEnabled) {
 		}
 	}
 
+	private String getAdminBootstrapAddress() {
+		// Retrieve bootstrap servers from KafkaAdmin bootstrap supplier if available
+		String adminServers = this.kafkaAdmin.getBootstrapServers();
+
+		// Fallback to configuration properties if bootstrap servers are not set
+		if (adminServers == null) {
+			adminServers = this.kafkaAdmin.getConfigurationProperties().getOrDefault(
+					AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
+					""
+			).toString();
+		}
+
+		return adminServers;
+	}
+
 	@Nullable
 	private String clusterId() {
 		if (this.kafkaAdmin != null && this.clusterId == null) {