Repository: provectus/kafka-ui
Issue: Too many connections to Zookeeper
Labels: type/bug
scope/backend

I have a problem with many connections to Zookeeper from kafka-ui.

Zookeeper logs:

![image](https://user-images.githubusercontent.com/2510489/107490229-aa3a1180-6b9a-11eb-9994-55bc960908c3.png)

docker-compose.yml

Kafka UI:
![image](https://user-images.githubusercontent.com/2510489/107490589-22083c00-6b9b-11eb-85b0-3b20935aeba0.png)

Zookeeper:
![image](https://user-images.githubusercontent.com/2510489/107490633-2fbdc180-6b9b-11eb-995c-995e667fa25d.png)

What am I doing wrong?


Issue comments:
@imoisey thank you for your issue. This bug will be fixed soon by https://github.com/provectus/kafka-ui/pull/185
Thanks

Pull request review comments:
Let's pass a zookeeper connection timeout from the Java class/app properties


Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/SchemaRegistryService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/SchemaRegistryService.java
index 609edb27634..b67ce348d36 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/SchemaRegistryService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/SchemaRegistryService.java
@@ -9,6 +9,7 @@
 import com.provectus.kafka.ui.model.CompatibilityLevel;
 import com.provectus.kafka.ui.model.NewSchemaSubject;
 import com.provectus.kafka.ui.model.SchemaSubject;
+import java.util.Formatter;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.http.HttpStatus;
@@ -51,9 +52,12 @@ public Flux<Integer> getSchemaSubjectVersions(String clusterName, String schemaN
                 .map(cluster -> webClient.get()
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECT_VERSIONS, schemaName)
                         .retrieve()
-                        .onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such schema %s".formatted(schemaName))))
-                        .bodyToFlux(Integer.class))
-                .orElse(Flux.error(new NotFoundException("No such cluster")));
+                        .onStatus(HttpStatus.NOT_FOUND::equals,
+                            resp -> Mono.error(
+                                new NotFoundException(formatted("No such schema %s"))
+                            )
+                        ).bodyToFlux(Integer.class)
+                ).orElse(Flux.error(new NotFoundException("No such cluster")));
     }
 
     public Mono<SchemaSubject> getSchemaSubjectByVersion(String clusterName, String schemaName, Integer version) {
@@ -70,8 +74,12 @@ private Mono<SchemaSubject> getSchemaSubject(String clusterName, String schemaNa
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECT_BY_VERSION, schemaName, version)
                         .retrieve()
                         .onStatus(HttpStatus.NOT_FOUND::equals,
-                                resp -> Mono.error(new NotFoundException("No such schema %s with version %s".formatted(schemaName, version))))
-                        .bodyToMono(SchemaSubject.class)
+                                resp -> Mono.error(
+                                    new NotFoundException(
+                                        formatted("No such schema %s with version %s", schemaName, version)
+                                    )
+                                )
+                        ).bodyToMono(SchemaSubject.class)
                         .zipWith(getSchemaCompatibilityInfoOrGlobal(clusterName, schemaName))
                         .map(tuple -> {
                             SchemaSubject schema = tuple.getT1();
@@ -97,9 +105,13 @@ private Mono<ResponseEntity<Void>> deleteSchemaSubject(String clusterName, Strin
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECT_BY_VERSION, schemaName, version)
                         .retrieve()
                         .onStatus(HttpStatus.NOT_FOUND::equals,
-                                resp -> Mono.error(new NotFoundException("No such schema %s with version %s".formatted(schemaName, version))))
-                        .toBodilessEntity())
-                .orElse(Mono.error(new NotFoundException("No such cluster")));
+                                resp -> Mono.error(
+                                    new NotFoundException(
+                                        formatted("No such schema %s with version %s", schemaName, version)
+                                    )
+                                )
+                        ).toBodilessEntity()
+                ).orElse(Mono.error(new NotFoundException("No such cluster")));
     }
 
     public Mono<ResponseEntity<Void>> deleteSchemaSubject(String clusterName, String schemaName) {
@@ -107,7 +119,13 @@ public Mono<ResponseEntity<Void>> deleteSchemaSubject(String clusterName, String
                 .map(cluster -> webClient.delete()
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECT, schemaName)
                         .retrieve()
-                        .onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such schema %s".formatted(schemaName))))
+                        .onStatus(HttpStatus.NOT_FOUND::equals,
+                            resp -> Mono.error(
+                                new NotFoundException(
+                                    formatted("No such schema %s", schemaName)
+                                )
+                            )
+                        )
                         .toBodilessEntity())
                 .orElse(Mono.error(new NotFoundException("No such cluster")));
     }
@@ -120,7 +138,9 @@ public Mono<ResponseEntity<SchemaSubject>> createNewSubject(String clusterName,
                         .body(BodyInserters.fromPublisher(newSchemaSubject, NewSchemaSubject.class))
                         .retrieve()
                         .onStatus(HttpStatus.NOT_FOUND::equals,
-                                resp -> Mono.error(new NotFoundException("No such schema %s".formatted(schemaName))))
+                                resp -> Mono.error(
+                                    new NotFoundException(formatted("No such schema %s", schemaName)))
+                        )
                         .toEntity(SchemaSubject.class)
                         .log())
                 .orElse(Mono.error(new NotFoundException("No such cluster")));
@@ -142,7 +162,7 @@ public Mono<Void> updateSchemaCompatibility(String clusterName, String schemaNam
                             .body(BodyInserters.fromPublisher(compatibilityLevel, CompatibilityLevel.class))
                             .retrieve()
                             .onStatus(HttpStatus.NOT_FOUND::equals,
-                                    resp -> Mono.error(new NotFoundException("No such schema %s".formatted(schemaName))))
+                                    resp -> Mono.error(new NotFoundException(formatted("No such schema %s", schemaName))))
                             .bodyToMono(Void.class);
                 }).orElse(Mono.error(new NotFoundException("No such cluster")));
     }
@@ -181,10 +201,14 @@ public Mono<CompatibilityCheckResponse> checksSchemaCompatibility(String cluster
                         .body(BodyInserters.fromPublisher(newSchemaSubject, NewSchemaSubject.class))
                         .retrieve()
                         .onStatus(HttpStatus.NOT_FOUND::equals,
-                                resp -> Mono.error(new NotFoundException("No such schema %s".formatted(schemaName))))
+                                resp -> Mono.error(new NotFoundException(formatted("No such schema %s", schemaName))))
                         .bodyToMono(InternalCompatibilityCheck.class)
                         .map(mapper::toCompatibilityCheckResponse)
                         .log()
                 ).orElse(Mono.error(new NotFoundException("No such cluster")));
     }
+
+    public String formatted(String str, Object... args) {
+        return new Formatter().format(str, args).toString();
+    }
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/zookeeper/ZookeeperService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/zookeeper/ZookeeperService.java
index 26b6f300b2a..855ea62c46f 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/zookeeper/ZookeeperService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/zookeeper/ZookeeperService.java
@@ -1,6 +1,7 @@
 package com.provectus.kafka.ui.zookeeper;
 
 import com.provectus.kafka.ui.cluster.model.KafkaCluster;
+import java.util.concurrent.ConcurrentHashMap;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.I0Itec.zkclient.ZkClient;
@@ -14,7 +15,7 @@
 @Log4j2
 public class ZookeeperService {
 
-    private final Map<String, ZkClient> cachedZkClient = new HashMap<>();
+    private final Map<String, ZkClient> cachedZkClient = new ConcurrentHashMap<>();
 
     public boolean isZookeeperOnline(KafkaCluster kafkaCluster) {
         var isConnected = false;
@@ -33,7 +34,10 @@ private boolean isZkClientConnected(ZkClient zkClient) {
 
     private ZkClient getOrCreateZkClient (KafkaCluster cluster) {
         try {
-            return cachedZkClient.getOrDefault(cluster.getName(), new ZkClient(cluster.getZookeeper(), 1000));
+            return cachedZkClient.computeIfAbsent(
+                cluster.getName(),
+                (n) -> new ZkClient(cluster.getZookeeper(), 1000)
+            );
         } catch (Exception e) {
             log.error("Error while creating zookeeper client for cluster {}", cluster.getName());
             return null;


Test patch: