Repository: provectus/kafka-ui
Issue: Zookeeper thread leaking
Labels: type/bug
good first issue
scope/backend
status/accepted
status/confirmed

Hi!

Thanks in advance for your help! kafka-ui is a super helpful tool!

**Describe the bug**
After running for a day the kafka-ui docker container (image: provectuslabs/kafka-ui:0.3.1) is using over 5000 Threads leading to high CPU and Memory Usage and evntually a crash of the host machine if nothing is done.

kafka-ui throws a `zookeeper exception` at regular intervals, which may be related (see the log below).

**Set up**
+ debian buster
+ docker
+ provectuslabs/kafka-ui:0.3.1 Image
+ orchestration by docker-compose with kafka in the same project:
```yaml
version: "3.7"

services:
  zookeeper:
    image: bitnami/zookeeper:3.7.0
    restart: unless-stopped
    networks:
      - kafka
    volumes:
     - "zookeeper_data:/bitnami"

  kafka:
    image: bitnami/kafka:2.8.1
    restart: unless-stopped
    networks:
      - kafka
    volumes:
      - "kafka_data:/bitnami"
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
      - JMX_PORT=9001
    depends_on:
      - zookeeper

kafkaui:
    image: provectuslabs/kafka-ui:0.3.1
    networks:
      - kafka
    environment:
      - KAFKA_CLUSTERS_0_NAME=cluster-name
      - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092
      - KAFKA_CLUSTERS_0_ZOOKEEPER=zookeeper:2183
      - KAFKA_CLUSTERS_0_JMXPORT=9001
      - LOGGING_LEVEL_ROOT=error
      - LOGGING_LEVEL_COM_PROVECTUS=error
    depends_on:
      - zookeeper
      - kafka
...
```

This problem also occurs with older versions (0.2.*). The load of the kafka instance is low.

**Steps to Reproduce**
See above.

**Expected behavior**
The resource usage should be stable over time.

**Screenshots**
`docker stats` output after the container has been running for a day:

<img width="772" alt="Screenshot 2022-01-25 085956" src="https://user-images.githubusercontent.com/6784851/150935803-9ef6956d-7b43-42ad-8442-fae981c742ba.png">

Second column is the CPU Usage, last column is the PIDS column.

**Additional context**
Log output: 
```
Jan 25 08:07:55: #033[30m2022-01-25 08:07:55,596#033[0;39m #033[1;31mERROR#033[0;39m [#033[34mkafka-admin-client-thread | adminclient-1#033[0;39m] #033[33mc.p.k.u.s.ZookeeperService#033[0;39m: A zookeeper exception has occurred
Jan 25 08:07:55: org.apache.zookeeper.KeeperException$ConnectionLossException: KeeperErrorCode = ConnectionLoss for /brokers/ids
Jan 25 08:07:55: #011at org.apache.zookeeper.KeeperException.create(KeeperException.java:102)
Jan 25 08:07:55: #011at org.apache.zookeeper.KeeperException.create(KeeperException.java:54)
Jan 25 08:07:55: #011at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:2366)
Jan 25 08:07:55: #011at com.provectus.kafka.ui.service.ZookeeperService.isZkClientConnected(ZookeeperService.java:57)
Jan 25 08:07:55: #011at com.provectus.kafka.ui.service.ZookeeperService.isZookeeperOnline(ZookeeperService.java:49)
Jan 25 08:07:55: #011at com.provectus.kafka.ui.service.ZookeeperService.lambda$getZkStatus$0(ZookeeperService.java:37)
Jan 25 08:07:55: #011at reactor.core.publisher.MonoSupplier.subscribe(MonoSupplier.java:57)
Jan 25 08:07:55: #011at reactor.core.publisher.Mono.subscribe(Mono.java:4399)
Jan 25 08:07:55: #011at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128)
Jan 25 08:07:55: #011at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157)
Jan 25 08:07:55: #011at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:165)
Jan 25 08:07:55: #011at com.provectus.kafka.ui.service.ReactiveAdminClient.lambda$describeCluster$18(ReactiveAdminClient.java:223)
Jan 25 08:07:55: #011at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:177)
Jan 25 08:07:55: #011at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:162)
Jan 25 08:07:55: #011at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221)
Jan 25 08:07:55: #011at org.apache.kafka.common.KafkaFuture$AllOfAdapter.maybeComplete(KafkaFuture.java:82)
Jan 25 08:07:55: #011at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:76)
Jan 25 08:07:55: #011at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:57)
Jan 25 08:07:55: #011at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221)
Jan 25 08:07:55: #011at org.apache.kafka.clients.admin.KafkaAdminClient$5.handleResponse(KafkaAdminClient.java:1875)
Jan 25 08:07:55: #011at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.handleResponses(KafkaAdminClient.java:1189)
Jan 25 08:07:55: #011at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1341)
Jan 25 08:07:55: #011at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1264)
Jan 25 08:07:55: #011at java.base/java.lang.Thread.run(Thread.java:830)
Jan 25 08:08:25: #033[30m2022-01-25 08:08:25,516#033[0;39m #033[1;31mERROR#033[0;39m [#033[34mkafka-admin-client-thread | adminclient-1#033[0;39m] #033[33mc.p.k.u.s.ZookeeperService#033[0;39m: A zookeeper exception has occurred
Jan 25 08:08:25: org.apache.zookeeper.KeeperException$ConnectionLossException: KeeperErrorCode = ConnectionLoss for /brokers/ids
Jan 25 08:08:25: #011at org.apache.zookeeper.KeeperException.create(KeeperException.java:102)
Jan 25 08:08:25: #011at org.apache.zookeeper.KeeperException.create(KeeperException.java:54)
Jan 25 08:08:25: #011at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:2366)
Jan 25 08:08:25: #011at com.provectus.kafka.ui.service.ZookeeperService.isZkClientConnected(ZookeeperService.java:57)
Jan 25 08:08:25: #011at com.provectus.kafka.ui.service.ZookeeperService.isZookeeperOnline(ZookeeperService.java:49)
Jan 25 08:08:25: #011at com.provectus.kafka.ui.service.ZookeeperService.lambda$getZkStatus$0(ZookeeperService.java:37)
Jan 25 08:08:25: #011at reactor.core.publisher.MonoSupplier.subscribe(MonoSupplier.java:57)
Jan 25 08:08:25: #011at reactor.core.publisher.Mono.subscribe(Mono.java:4399)
Jan 25 08:08:25: #011at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128)
Jan 25 08:08:25: #011at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157)
Jan 25 08:08:25: #011at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:165)
Jan 25 08:08:25: #011at com.provectus.kafka.ui.service.ReactiveAdminClient.lambda$describeCluster$18(ReactiveAdminClient.java:223)
Jan 25 08:08:25: #011at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:177)
Jan 25 08:08:25: #011at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:162)
Jan 25 08:08:25: #011at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221)
Jan 25 08:08:25: #011at org.apache.kafka.common.KafkaFuture$AllOfAdapter.maybeComplete(KafkaFuture.java:82)
Jan 25 08:08:25: #011at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:76)
Jan 25 08:08:25: #011at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:57)
Jan 25 08:08:25: #011at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221)
Jan 25 08:08:25: #011at org.apache.kafka.clients.admin.KafkaAdminClient$5.handleResponse(KafkaAdminClient.java:1875)
Jan 25 08:08:25: #011at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.handleResponses(KafkaAdminClient.java:1189)
Jan 25 08:08:25: #011at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1341)
Jan 25 08:08:25: #011at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1264)
Jan 25 08:08:25: #011at java.base/java.lang.Thread.run(Thread.java:830)
Jan 25 08:08:55: #033[30m2022-01-25 08:08:55,645#033[0;39m #033[1;31mERROR#033[0;39m [#033[34mkafka-admin-client-thread | adminclient-1#033[0;39m] #033[33mc.p.k.u.s.ZookeeperService#033[0;39m: A zookeeper exception has occurred
Jan 25 08:08:55: org.apache.zookeeper.KeeperException$ConnectionLossException: KeeperErrorCode = ConnectionLoss for /brokers/ids
Jan 25 08:08:55: #011at org.apache.zookeeper.KeeperException.create(KeeperException.java:102)
Jan 25 08:08:55: #011at org.apache.zookeeper.KeeperException.create(KeeperException.java:54)
Jan 25 08:08:55: #011at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:2366)
Jan 25 08:08:55: #011at com.provectus.kafka.ui.service.ZookeeperService.isZkClientConnected(ZookeeperService.java:57)
Jan 25 08:08:55: #011at com.provectus.kafka.ui.service.ZookeeperService.isZookeeperOnline(ZookeeperService.java:49)
Jan 25 08:08:55: #011at com.provectus.kafka.ui.service.ZookeeperService.lambda$getZkStatus$0(ZookeeperService.java:37)
Jan 25 08:08:55: #011at reactor.core.publisher.MonoSupplier.subscribe(MonoSupplier.java:57)
Jan 25 08:08:55: #011at reactor.core.publisher.Mono.subscribe(Mono.java:4399)
Jan 25 08:08:55: #011at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128)
Jan 25 08:08:55: #011at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157)
Jan 25 08:08:55: #011at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:165)
Jan 25 08:08:55: #011at com.provectus.kafka.ui.service.ReactiveAdminClient.lambda$describeCluster$18(ReactiveAdminClient.java:223)
Jan 25 08:08:55: #011at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:177)
Jan 25 08:08:55: #011at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:162)
Jan 25 08:08:55: #011at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221)
Jan 25 08:08:55: #011at org.apache.kafka.common.KafkaFuture$AllOfAdapter.maybeComplete(KafkaFuture.java:82)
Jan 25 08:08:55: #011at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:76)
Jan 25 08:08:55: #011at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:57)
Jan 25 08:08:55: #011at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221)
Jan 25 08:08:55: #011at org.apache.kafka.clients.admin.KafkaAdminClient$5.handleResponse(KafkaAdminClient.java:1875)
Jan 25 08:08:55: #011at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.handleResponses(KafkaAdminClient.java:1189)
Jan 25 08:08:55: #011at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1341)
Jan 25 08:08:55: #011at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1264)
Jan 25 08:08:55: #011at java.base/java.lang.Thread.run(Thread.java:830)
```


Issue comments:
Hello there cleboo! 👋

Thank you and congratulations 🎉 for opening your very first issue in this project! 💖

In case you want to claim this issue, please comment down below! We will try to get back to you as soon as we can. 👀
Hey, thanks for reaching out.

This is for sure because of the zookeeper. We'll take a look.
You have a wrong port set up:
`   - KAFKA_CLUSTERS_0_ZOOKEEPER=zookeeper:2183`
The default one is 2181.
Tried your setup (a close one, since the one you provided wasn't valid enough), works fine for me.

```
version: "3.7"

services:
  zookeeper:
    image: bitnami/zookeeper:3.7.0
    restart: unless-stopped
    volumes:
     - "./zookeeper_data:/bitnami"
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes

  kafka:
    image: bitnami/kafka:2.8.1
    restart: unless-stopped
    volumes:
      - "./kafka_data:/bitnami"
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
      - JMX_PORT=9001
    depends_on:
      - zookeeper

  kafkaui:
    image: provectuslabs/kafka-ui:latest
    environment:
      - KAFKA_CLUSTERS_0_NAME=cluster-name
      - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092
      - KAFKA_CLUSTERS_0_ZOOKEEPER=zookeeper:2181
      - KAFKA_CLUSTERS_0_JMXPORT=9001
      - LOGGING_LEVEL_ROOT=error
      - LOGGING_LEVEL_COM_PROVECTUS=error
    depends_on:
      - zookeeper
      - kafka
    ports:
      - 8080:8080
```
![image](https://user-images.githubusercontent.com/1494347/150969595-f9ef6938-edbe-4d11-8de7-ddb0aa3fda80.png)

Ahh, thanks  @Haarolean for the fix!

Still, might be worth fixing the PID Issue as it can crash the host if you accidentally misconfigure the zookeeper connection.
Sure, we'll take a look.

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ZookeeperService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ZookeeperService.java
index 03638afc7f6..6e8f1a5f516 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ZookeeperService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ZookeeperService.java
@@ -56,6 +56,7 @@ private boolean isZkClientConnected(ZooKeeper zkClient) {
       zkClient.getChildren("/brokers/ids", null);
     } catch (KeeperException e) {
       log.error("A zookeeper exception has occurred", e);
+      closeZkClientSession(zkClient, e);
       return false;
     } catch (InterruptedException e) {
       log.error("Interrupted: ", e);
@@ -64,6 +65,15 @@ private boolean isZkClientConnected(ZooKeeper zkClient) {
     return true;
   }
 
+  private void closeZkClientSession(ZooKeeper zkClient, KeeperException e) {
+    try {
+      zkClient.close();
+    } catch (InterruptedException ex) {
+      log.error("Unable to close zkClient session: ", e);
+      Thread.currentThread().interrupt();
+    }
+  }
+
   @Nullable
   private ZooKeeper getOrCreateZkClient(KafkaCluster cluster) {
     final var clusterName = cluster.getName();


Test patch:
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ZookeeperServiceTest.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ZookeeperServiceTest.java
new file mode 100644
index 00000000000..92da37f27a8
--- /dev/null
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ZookeeperServiceTest.java
@@ -0,0 +1,52 @@
+package com.provectus.kafka.ui.service;
+
+import com.provectus.kafka.ui.AbstractIntegrationTest;
+import com.provectus.kafka.ui.model.KafkaCluster;
+import com.provectus.kafka.ui.model.ServerStatusDTO;
+import java.util.Properties;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import reactor.test.StepVerifier;
+
+class ZookeeperServiceTest extends AbstractIntegrationTest {
+  private ZookeeperService zookeeperService;
+
+  @BeforeEach
+  void init() {
+    AdminClientServiceImpl adminClientService = new AdminClientServiceImpl();
+    adminClientService.setClientTimeout(5_000);
+    zookeeperService = new ZookeeperService();
+  }
+
+  @Test
+  void getZkStatusEmptyConfig() {
+    KafkaCluster kafkaCluster =
+        KafkaCluster.builder()
+            .name(LOCAL)
+            .bootstrapServers(kafka.getBootstrapServers())
+            .properties(new Properties())
+            .build();
+
+    ZookeeperService.ZkStatus zkStatus = new ZookeeperService.ZkStatus(ServerStatusDTO.OFFLINE, null);
+    StepVerifier.create(zookeeperService.getZkStatus(kafkaCluster))
+        .expectNext(zkStatus)
+        .verifyComplete();
+  }
+
+  @Test
+  void getZkStatusWrongConfig() {
+    KafkaCluster kafkaCluster =
+        KafkaCluster.builder()
+            .name(LOCAL)
+            .bootstrapServers(kafka.getBootstrapServers())
+            .zookeeper("localhost:1000")
+            .properties(new Properties())
+            .build();
+
+    ZookeeperService.ZkStatus zkStatus = new ZookeeperService.ZkStatus(ServerStatusDTO.OFFLINE, null);
+    StepVerifier.create(zookeeperService.getZkStatus(kafkaCluster))
+        .expectNext(zkStatus)
+        .verifyComplete();
+  }
+
+}