Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
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`**
This commit is contained in:
Artem Bilan
2023-12-16 20:25:42 -05:00
committed by GitHub
parent eb68d6d845
commit d982c8e822
1 changed files with 13 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -29,6 +29,8 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
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;
@@ -118,6 +120,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;
@@ -501,7 +505,15 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
@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;
}