Repository: provectus/kafka-ui
Issue: Timeouts with large amount of topics
Labels: type/bug
scope/backend
status/accepted
status/confirmed

**Describe the bug**
<!--(A clear and concise description of what the bug is.)-->
Can't connect my Stage Environment Kafka.

**Set up**
<!--
(How do you run the app? 
Which version of the app are you running? Provide either docker image version or check commit hash at the top left corner. We won't be able to help you without this information.)
-->

docker-compose installation

> version: "3.7"
> services:
>   kafka-ui:
>     container_name: kafka-ui
>     image: "provectuslabs/kafka-ui:latest"
>     restart: always
>     privileged: true
>     ports:
>       - "8099:8080"
>     environment:
>       KAFKA_CLUSTERS_0_NAME: Stage
>       KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: 10.2.20.143:9092


* version: vv0.4.0 [521ba0c](https://github.com/provectus/kafka-ui/commit/521ba0c)

**Steps to Reproduce**
Steps to reproduce the behavior:

1. create container with `docker-compose up -d`
2. check container logs
<details>
 _   _ ___    __             _                _          _  __      __ _
| | | |_ _|  / _|___ _ _    /_\  _ __ __ _ __| |_  ___  | |/ /__ _ / _| |_____
| |_| || |  |  _/ _ | '_|  / _ \| '_ / _` / _| ' \/ -_) | ' </ _` |  _| / / _`|
 \___/|___| |_| \___|_|   /_/ \_| .__\__,_\__|_||_\___| |_|\_\__,_|_| |_\_\__,|
                                 |_|                                             

2022-05-18 20:50:35,957 INFO  [background-preinit] o.h.v.i.u.Version: HV000001: Hibernate Validator 6.2.0.Final
2022-05-18 20:50:35,982 INFO  [main] c.p.k.u.KafkaUiApplication: Starting KafkaUiApplication using Java 13.0.9 on 373c8e33779a with PID 1 (/kafka-ui-api.jar started by kafkaui in /)
2022-05-18 20:50:35,983 DEBUG [main] c.p.k.u.KafkaUiApplication: Running with Spring Boot v2.6.3, Spring v5.3.15
2022-05-18 20:50:35,983 INFO  [main] c.p.k.u.KafkaUiApplication: No active profile set, falling back to default profiles: default
2022-05-18 20:50:38,790 INFO  [main] o.s.d.r.c.RepositoryConfigurationDelegate: Bootstrapping Spring Data LDAP repositories in DEFAULT mode.
2022-05-18 20:50:38,891 INFO  [main] o.s.d.r.c.RepositoryConfigurationDelegate: Finished Spring Data repository scanning in 91 ms. Found 0 LDAP repository interfaces.
2022-05-18 20:50:40,168 INFO  [main] c.p.k.u.s.DeserializationService: Using SimpleRecordSerDe for cluster 'Stage'
2022-05-18 20:50:42,407 INFO  [main] o.s.b.a.e.w.EndpointLinksResolver: Exposing 2 endpoint(s) beneath base path '/actuator'
2022-05-18 20:50:42,630 INFO  [main] o.s.b.a.s.r.ReactiveUserDetailsServiceAutoConfiguration: 

Using generated security password: 35b6dd3f-87da-4515-bf27-74f9979869ad

2022-05-18 20:50:42,824 WARN  [main] c.p.k.u.c.a.DisabledAuthSecurityConfig: Authentication is disabled. Access will be unrestricted.
2022-05-18 20:50:43,120 INFO  [main] o.s.l.c.s.AbstractContextSource: Property 'userDn' not set - anonymous context will be used for read-write operations
2022-05-18 20:50:44,149 INFO  [main] o.s.b.w.e.n.NettyWebServer: Netty started on port 8080
2022-05-18 20:50:44,195 INFO  [main] c.p.k.u.KafkaUiApplication: Started KafkaUiApplication in 9.109 seconds (JVM running for 10.11)
2022-05-18 20:50:44,250 DEBUG [parallel-1] c.p.k.u.s.ClustersMetricsScheduler: Start getting metrics for kafkaCluster: Stage
2022-05-18 20:50:44,278 INFO  [parallel-1] o.a.k.c.a.AdminClientConfig: AdminClientConfig values: 
	bootstrap.servers = [10.2.20.143:9092]
	client.dns.lookup = use_all_dns_ips
	client.id = 
	connections.max.idle.ms = 300000
	default.api.timeout.ms = 60000
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 2147483647
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	socket.connection.setup.timeout.max.ms = 30000
	socket.connection.setup.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.3
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS

2022-05-18 20:50:44,403 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka version: 2.8.0
2022-05-18 20:50:44,403 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka commitId: ebb1d6e21cc92130
2022-05-18 20:50:44,403 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka startTimeMs: 1652907044401
</details>

3. here is the dasboard
![image](https://user-images.githubusercontent.com/23381187/169153727-0d4bedcf-1e3a-426b-8840-6084b97ea404.png)
4. topics
![image](https://user-images.githubusercontent.com/23381187/169153801-017d41a6-b88e-4cb4-99d0-3e7686ab31b0.png)

Connection via Conduktor
![image](https://user-images.githubusercontent.com/23381187/169154182-e6729820-faf4-453f-b726-84e1e6bd7f50.png)

Connection via CLI
`bin/kafka-console-producer.sh --broker-list 10.2.20.143:9092 --topic my_topic`
![image](https://user-images.githubusercontent.com/23381187/169154610-1a06a56d-a1f4-478a-ba21-62670ac41126.png)

**Expected behavior**
<!--
(A clear and concise description of what you expected to happen)
-->
Successfully connected kafka-ui app.
**Screenshots**
<!--
(If applicable, add screenshots to help explain your problem)
-->


**Additional context**
<!--
(Add any other context about the problem here)
-->


Issue comments:
Hello there keremcankabadayi! 👋

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. 👀
Hi, it's not an issue with the app, but a docker-network related one.
if you're planning to run it on your local machine, you can try docker's `host` network mode.
https://docs.docker.com/compose/compose-file/compose-file-v3/#network_mode
Feel free to leave a reply if the problem persists, either here or on discord.

I think thats not the problem cause i also tried with `jar` without docker. results and logs were same.

```
export KAFKA_CLUSTERS_0_NAME= Stage
export  KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS= 10.2.20.143:9092
java -jar kafka-ui-api-v0.4.0.jar
```

Its kind of connection problem.
Okay then, you should've mentioned it beforehand :)

Please share the full log when you run it as a jar file. The one you provided is just a little bit of it.
**Set up**
```
export KAFKA_CLUSTERS_0_NAME=Stage
export  KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=10.2.20.143:9092
java -jar kafka-ui-api-v0.4.0.jar
```

**Steps to Reproduce**
1. Run with above commands.
2. Check logs


```
 _   _ ___    __             _                _          _  __      __ _
| | | |_ _|  / _|___ _ _    /_\  _ __ __ _ __| |_  ___  | |/ /__ _ / _| |_____
| |_| || |  |  _/ _ | '_|  / _ \| '_ / _` / _| ' \/ -_) | ' </ _` |  _| / / _`|
 \___/|___| |_| \___|_|   /_/ \_| .__\__,_\__|_||_\___| |_|\_\__,_|_| |_\_\__,|
                                 |_|                                             
2022-05-20 13:11:01,833 INFO  [background-preinit] o.h.v.i.u.Version: HV000001: Hibernate Validator 6.2.0.Final
2022-05-20 13:11:01,857 INFO  [main] c.p.k.u.KafkaUiApplication: Starting KafkaUiApplication using Java 13.0.11 on C02FV4V3MD6R with PID 21758 (/Users/keremcan.kabadayi/Downloads/kafka-ui-api-v0.4.0.jar started by keremcan.kabadayi in /Users/keremcan.kabadayi/Downloads)
2022-05-20 13:11:01,858 DEBUG [main] c.p.k.u.KafkaUiApplication: Running with Spring Boot v2.6.3, Spring v5.3.15
2022-05-20 13:11:01,858 INFO  [main] c.p.k.u.KafkaUiApplication: No active profile set, falling back to default profiles: default
2022-05-20 13:11:03,165 INFO  [main] o.s.d.r.c.RepositoryConfigurationDelegate: Bootstrapping Spring Data LDAP repositories in DEFAULT mode.
2022-05-20 13:11:03,211 INFO  [main] o.s.d.r.c.RepositoryConfigurationDelegate: Finished Spring Data repository scanning in 37 ms. Found 0 LDAP repository interfaces.
2022-05-20 13:11:03,843 INFO  [main] c.p.k.u.s.DeserializationService: Using SimpleRecordSerDe for cluster 'Stage'
2022-05-20 13:11:04,547 INFO  [main] o.s.b.a.e.w.EndpointLinksResolver: Exposing 2 endpoint(s) beneath base path '/actuator'
2022-05-20 13:11:04,716 INFO  [main] o.s.b.a.s.r.ReactiveUserDetailsServiceAutoConfiguration: 

Using generated security password: bda90694-ace1-4553-ad51-6ffc9dc66db5

2022-05-20 13:11:04,817 WARN  [main] c.p.k.u.c.a.DisabledAuthSecurityConfig: Authentication is disabled. Access will be unrestricted.
2022-05-20 13:11:05,009 INFO  [main] o.s.l.c.s.AbstractContextSource: Property 'userDn' not set - anonymous context will be used for read-write operations
2022-05-20 13:11:05,310 INFO  [main] o.s.b.w.e.n.NettyWebServer: Netty started on port 8080
2022-05-20 13:11:05,326 INFO  [main] c.p.k.u.KafkaUiApplication: Started KafkaUiApplication in 4.326 seconds (JVM running for 5.173)
2022-05-20 13:11:05,350 DEBUG [parallel-1] c.p.k.u.s.ClustersMetricsScheduler: Start getting metrics for kafkaCluster: Stage
2022-05-20 13:11:05,362 INFO  [parallel-1] o.a.k.c.a.AdminClientConfig: AdminClientConfig values: 
        bootstrap.servers = [10.2.20.143:9092]
        client.dns.lookup = use_all_dns_ips
        client.id = 
        connections.max.idle.ms = 300000
        default.api.timeout.ms = 60000
        metadata.max.age.ms = 300000
        metric.reporters = []
        metrics.num.samples = 2
        metrics.recording.level = INFO
        metrics.sample.window.ms = 30000
        receive.buffer.bytes = 65536
        reconnect.backoff.max.ms = 1000
        reconnect.backoff.ms = 50
        request.timeout.ms = 30000
        retries = 2147483647
        retry.backoff.ms = 100
        sasl.client.callback.handler.class = null
        sasl.jaas.config = null
        sasl.kerberos.kinit.cmd = /usr/bin/kinit
        sasl.kerberos.min.time.before.relogin = 60000
        sasl.kerberos.service.name = null
        sasl.kerberos.ticket.renew.jitter = 0.05
        sasl.kerberos.ticket.renew.window.factor = 0.8
        sasl.login.callback.handler.class = null
        sasl.login.class = null
        sasl.login.refresh.buffer.seconds = 300
        sasl.login.refresh.min.period.seconds = 60
        sasl.login.refresh.window.factor = 0.8
        sasl.login.refresh.window.jitter = 0.05
        sasl.mechanism = GSSAPI
        security.protocol = PLAINTEXT
        security.providers = null
        send.buffer.bytes = 131072
        socket.connection.setup.timeout.max.ms = 30000
        socket.connection.setup.timeout.ms = 10000
        ssl.cipher.suites = null
        ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
        ssl.endpoint.identification.algorithm = https
        ssl.engine.factory.class = null
        ssl.key.password = null
        ssl.keymanager.algorithm = SunX509
        ssl.keystore.certificate.chain = null
        ssl.keystore.key = null
        ssl.keystore.location = null
        ssl.keystore.password = null
        ssl.keystore.type = JKS
        ssl.protocol = TLSv1.3
        ssl.provider = null
        ssl.secure.random.implementation = null
        ssl.trustmanager.algorithm = PKIX
        ssl.truststore.certificates = null
        ssl.truststore.location = null
        ssl.truststore.password = null
        ssl.truststore.type = JKS

2022-05-20 13:11:05,433 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka version: 2.8.0
2022-05-20 13:11:05,433 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka commitId: ebb1d6e21cc92130
2022-05-20 13:11:05,433 INFO  [parallel-1] o.a.k.c.u.AppInfoParser: Kafka startTimeMs: 1653041465431
2022-05-20 13:12:08,873 ERROR [parallel-1] c.p.k.u.s.MetricsService: Failed to collect cluster Stage info
org.apache.kafka.common.errors.TimeoutException: Call(callName=describeConfigs, deadlineMs=1653041528867, tries=2, nextAllowedTryMs=1653041528968) timed out at 1653041528868 after 2 attempt(s)
Caused by: org.apache.kafka.common.errors.DisconnectException: Cancelled describeConfigs request with correlation id 31 due to node 6 being disconnected
2022-05-20 13:12:08,873 DEBUG [parallel-1] c.p.k.u.s.ClustersMetricsScheduler: Metrics updated for cluster: Stage
2022-05-20 13:12:08,883 DEBUG [parallel-14] c.p.k.u.s.ClustersMetricsScheduler: Start getting metrics for kafkaCluster: Stage
2022-05-20 13:13:11,933 ERROR [parallel-14] c.p.k.u.s.MetricsService: Failed to collect cluster Stage info
org.apache.kafka.common.errors.TimeoutException: Call(callName=describeConfigs, deadlineMs=1653041591780, tries=1, nextAllowedTryMs=1653041592031) timed out at 1653041591931 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.DisconnectException: Cancelled describeConfigs request with correlation id 51 due to node 4 being disconnected
2022-05-20 13:13:11,934 DEBUG [parallel-14] c.p.k.u.s.ClustersMetricsScheduler: Metrics updated for cluster: Stage
2022-05-20 13:13:11,938 DEBUG [parallel-9] c.p.k.u.s.ClustersMetricsScheduler: Start getting metrics for kafkaCluster: Stage
```

**Additional context**

- I need to connect VPN for this Stage Kafka. I can connect with 3rd party apps i mentioned above `Conduktor`, `Kafka CLI` and [Kouncil](https://github.com/Consdata/kouncil)
- I can ping ip `10.2.20.143`

Thanks for your attention. 
@keremcankabadayi thanks. Do you run this kouncil tool as a docker container as well?
Yes, running docker container now.

```
CONTAINER ID   IMAGE                           COMMAND                  CREATED        STATUS        PORTS                                                                             NAMES
7080f97d3117   consdata/kouncil:latest         "java -Djava.securit…"   39 hours ago   Up 39 hours   0.0.0.0:8099->8080/tcp                                                            kouncil
```

![image](https://user-images.githubusercontent.com/23381187/169529877-41fdd63d-e857-4341-a43e-6d766d6105dd.png)

@keremcankabadayi can you ping/traceroute kafka's IP from within kafka-ui container? via `docker exec -it kafka-ui sh`
@Haarolean here logs
```
➜  apps docker exec -it kafka-ui sh
/ $ ping 10.2.20.143
PING 10.2.20.143 (10.2.20.143): 56 data bytes
64 bytes from 10.2.20.143: seq=0 ttl=42 time=57.378 ms
64 bytes from 10.2.20.143: seq=1 ttl=42 time=58.189 ms
64 bytes from 10.2.20.143: seq=2 ttl=42 time=66.688 ms
64 bytes from 10.2.20.143: seq=3 ttl=42 time=54.729 ms
64 bytes from 10.2.20.143: seq=4 ttl=42 time=54.852 ms
64 bytes from 10.2.20.143: seq=5 ttl=42 time=104.845 ms
64 bytes from 10.2.20.143: seq=6 ttl=42 time=154.781 ms
64 bytes from 10.2.20.143: seq=7 ttl=42 time=100.420 ms
64 bytes from 10.2.20.143: seq=8 ttl=42 time=105.424 ms
64 bytes from 10.2.20.143: seq=9 ttl=42 time=107.621 ms
64 bytes from 10.2.20.143: seq=10 ttl=42 time=63.469 ms
64 bytes from 10.2.20.143: seq=11 ttl=42 time=95.412 ms
64 bytes from 10.2.20.143: seq=12 ttl=42 time=95.873 ms
64 bytes from 10.2.20.143: seq=13 ttl=42 time=56.913 ms
64 bytes from 10.2.20.143: seq=14 ttl=42 time=49.255 ms
64 bytes from 10.2.20.143: seq=15 ttl=42 time=73.346 ms
64 bytes from 10.2.20.143: seq=16 ttl=42 time=53.972 ms
^C
--- 10.2.20.143 ping statistics ---
17 packets transmitted, 17 packets received, 0% packet loss
round-trip min/avg/max = 49.255/79.598/154.781 ms
```
@keremcankabadayi wanna discuss this [on discord](https://discord.gg/4DWzD7pGE5) to make it quicker? or somewhere else if you don't have one. 
The problem seems to be about having a huge amount of topics present in kafka.
workaround for this issue, `http://localhost:8098/ui/clusters/<cluster_name>/topics/<topic_name>` you can visit directly topic without all topics loads.
@keremcankabadayi hey, can we talk on discord once again? I have a beta build to test it out

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ReactiveAdminClient.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ReactiveAdminClient.java
index dedd609743c..731824cdf08 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ReactiveAdminClient.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ReactiveAdminClient.java
@@ -4,6 +4,8 @@
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toMap;
 
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterators;
 import com.provectus.kafka.ui.exception.IllegalEntityStateException;
 import com.provectus.kafka.ui.exception.NotFoundException;
 import com.provectus.kafka.ui.util.MapUtil;
@@ -11,6 +13,7 @@
 import java.io.Closeable;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -18,6 +21,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
@@ -131,6 +136,16 @@ public Mono<Map<String, List<ConfigEntry>>> getTopicsConfig() {
   }
 
   public Mono<Map<String, List<ConfigEntry>>> getTopicsConfig(Collection<String> topicNames) {
+    // we need to partition calls, because it can lead to AdminClient timeouts in case of large topics count
+    return partitionCalls(
+        topicNames,
+        200,
+        this::getTopicsConfigImpl,
+        (m1, m2) -> ImmutableMap.<String, List<ConfigEntry>>builder().putAll(m1).putAll(m2).build()
+    );
+  }
+
+  private Mono<Map<String, List<ConfigEntry>>> getTopicsConfigImpl(Collection<String> topicNames) {
     List<ConfigResource> resources = topicNames.stream()
         .map(topicName -> new ConfigResource(ConfigResource.Type.TOPIC, topicName))
         .collect(toList());
@@ -162,6 +177,16 @@ public Mono<Map<String, TopicDescription>> describeTopics() {
   }
 
   public Mono<Map<String, TopicDescription>> describeTopics(Collection<String> topics) {
+    // we need to partition calls, because it can lead to AdminClient timeouts in case of large topics count
+    return partitionCalls(
+        topics,
+        200,
+        this::describeTopicsImpl,
+        (m1, m2) -> ImmutableMap.<String, TopicDescription>builder().putAll(m1).putAll(m2).build()
+    );
+  }
+
+  private Mono<Map<String, TopicDescription>> describeTopicsImpl(Collection<String> topics) {
     return toMonoWithExceptionFilter(
         client.describeTopics(topics).values(),
         UnknownTopicOrPartitionException.class
@@ -402,6 +427,27 @@ private Mono<Void> alterConfig(String topicName, Map<String, String> configs) {
     return toMono(client.alterConfigs(Map.of(topicResource, config)).all());
   }
 
+  /**
+   * Splits input collection into batches, applies each batch sequentially to function
+   * and merges output Monos into one Mono.
+   */
+  private static <R, I> Mono<R> partitionCalls(Collection<I> items,
+                                               int partitionSize,
+                                               Function<Collection<I>, Mono<R>> call,
+                                               BiFunction<R, R, R> merger) {
+    if (items.isEmpty()) {
+      return call.apply(items);
+    }
+    Iterator<List<I>> parts = Iterators.partition(items.iterator(), partitionSize);
+    Mono<R> mono = call.apply(parts.next());
+    while (parts.hasNext()) {
+      var nextPart = parts.next();
+      // calls will be executed sequentially
+      mono = mono.flatMap(res1 -> call.apply(nextPart).map(res2 -> merger.apply(res1, res2)));
+    }
+    return mono;
+  }
+
   @Override
   public void close() {
     client.close();


Test patch: