Repository: provectus/kafka-ui
Issue: Connecting SASL_SSL with sasl.mechanisms=PLAIN
Labels: scope/backend

Hello

Im trying to connect this to an instance of Kafka as follows:


`  docker run -p 8080:8080 \
  -e KAFKA_CLUSTERS_0_NAME=cluster01 \
  -e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=<BROKER+PORT> \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=SASL_SSL \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM=PLAIN \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG='org.apache.kafka.common.security.plain.PlainLoginModule required username=<USERNAME> password=<PASSWORD>;'\
  -d provectuslabs/kafka-ui:latest`

It does not want to connect.
I have looked through multiple issues such as [this](https://github.com/provectus/kafka-ui/issues/573) one, but im not able to come right.

Your assistance would be greatly appreciated

Issue comments:
Hi, @david-webill. Thanks for creating this issue. Could you please provide more information:
1. How is your cluster is configured
2. Provide logs from kafka ui
Hi @germanosin 

Thanks for the reply.
Logs are here https://pastebin.com/V7CPQHAG

The instance is a hosted instance with confluent.
They provide all the configuration for you and just provide details on how to connect to the cluster:
`
bootstrap.servers=server_url
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule   required username='{{ CLUSTER_API_KEY }}'   password='{{ CLUSTER_API_SECRET }}';
sasl.mechanism=PLAIN
client.dns.lookup=use_all_dns_ips
acks=all
schema.registry.url=url
basic.auth.credentials.source=USER_INFO
` 
Hi, @david-webill 
according to the confluent doc: https://docs.confluent.io/platform/current/tutorials/examples/clients/docs/java-springboot.html

You have to put one additional config:
```
# Required for correctness in Apache Kafka clients prior to 2.6
client.dns.lookup=use_all_dns_ips
```

Could you try this config?
```
docker run -p 8080:8080 \ 
-e KAFKA_CLUSTERS_0_NAME=cluster01 \ 
-e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=<BROKER+PORT> \ 
-e KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=SASL_SSL \ 
-e KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM=PLAIN \ 
-e KAFKA_CLUSTERS_0_PROPERTIES_CLIENT_DNS_LOOKUP=use_all_dns_ips \ 
-e KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG='org.apache.kafka.common.security.plain.PlainLoginModule required username=<USERNAME> password=<PASSWORD>;' \ 
-d provectuslabs/kafka-ui:latest
```
@germanosin Thank you so much for the reply.
I tried adding the extra setting with the following:

 `docker run -p 8080:8080 \
  -e KAFKA_CLUSTERS_0_NAME=cluster01 \
  -e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=redacted \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=SASL_SSL \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM=PLAIN \
  -e KAFKA_CLUSTERS_0_PROPERTIES_CLIENT_DNS_LOOKUP=use_all_dns_ips \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG='org.apache.kafka.common.security.plain.PlainLoginModule   required username='redacted'   password='redacted';'\
  -d provectuslabs/kafka-ui:latest`


It still does not work.
Logs below.
https://pastebin.com/Niq8gfXA

Thank you for your help, it is much appreciated
@david-webill Looks like something wrong with jaas config:
```
Caused by: java.lang.IllegalArgumentException: JAAS config entry not terminated by semi-colon
```
I check and double checked.
What im passing to the JAAS is this in docker:
-e KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG='org.apache.kafka.common.security.plain.PlainLoginModule   required username=FOMSOVNWOMD   password=RepKT/4MhxVgrY+R8MLYG;'\

Im not sure if i should be wrapping in quotes.
Im sure its a stupid error, but I just cannot get it to work :)
I do have some characters in the password that may be messing up the input?

(NOT the actual username and password)
> I check and double checked.
> What im passing to the JAAS is this in docker:
> -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG='org.apache.kafka.common.security.plain.PlainLoginModule required username=FOMSOVNWOMD password=RepKT/4MhxVgrY+R8MLYG;'\
> 
> Im not sure if i should be wrapping in quotes.
> Im sure its a stupid error, but I just cannot get it to work :)
> I do have some characters in the password that may be messing up the input?
> 
> (NOT the actual username and password)

It should look like this:

```
-e KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG='org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";'
```
Could you wrap username and password into quotes?
I have tried that, seems I have a new error now 😄 
Docker command:
docker run -p 8080:8080 \
  -e KAFKA_CLUSTERS_0_NAME=cluster01 \
  -e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=.aws.confluent.cloud:9092 \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=SASL_SSL \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM=PLAIN \
  -e KAFKA_CLUSTERS_0_PROPERTIES_CLIENT_DNS_LOOKUP=use_all_dns_ips \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG='org.apache.kafka.common.security.plain.PlainLoginModule   required username="USERNAME"   password="PASSWORD";'\
  -d provectuslabs/kafka-ui:latest

At least the error has seemed to change.

https://pastebin.com/28MSD0Rg

@germanosin 
Not sure if it has anything to do with the SASL_SSL authentication
Ive set up a kafka deployment with sasl_plaintext as the security protocol, and its connecting no problem
Hmm, looks like an NPE somewhere in code. Could you please check latest version with tag master?
```
docker run -p 8080:8080
-e KAFKA_CLUSTERS_0_NAME=cluster01
-e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=.aws.confluent.cloud:9092
-e KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=SASL_SSL
-e KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM=PLAIN
-e KAFKA_CLUSTERS_0_PROPERTIES_CLIENT_DNS_LOOKUP=use_all_dns_ips
-e KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG='org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";'
-d provectuslabs/kafka-ui:master
```
Unfortunately still an error:
```13:23:41.080 [main] INFO  org.springframework.core.KotlinDetector - Kotlin reflection implementation not found at runtime, related features won't be available.


  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /

 =========|_|==============|___/=/_/_/_/

 :: Spring Boot ::        (v2.2.4.RELEASE)


13:23:42.095 [main] INFO  com.provectus.kafka.ui.KafkaUiApplication - Starting KafkaUiApplication on 5a67d335b191 with PID 1 (/kafka-ui-api.jar started by root in /)

13:23:42.096 [main] DEBUG com.provectus.kafka.ui.KafkaUiApplication - Running with Spring Boot v2.2.4.RELEASE, Spring v5.2.3.RELEASE

13:23:42.097 [main] INFO  com.provectus.kafka.ui.KafkaUiApplication - No active profile set, falling back to default profiles: default

13:23:42.539 [background-preinit] WARN  org.springframework.http.converter.json.Jackson2ObjectMapperBuilder - For Jackson Kotlin classes support please add "com.fasterxml.jackson.module:jackson-module-kotlin" to the classpath

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/kafka-ui-api.jar!/BOOT-INF/lib/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]

SLF4J: Found binding in [jar:file:/kafka-ui-api.jar!/BOOT-INF/lib/log4j-slf4j-impl-2.12.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.

SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]

log4j:WARN No appenders could be found for logger (reactor.util.Loggers$LoggerFactory).
log4j:WARN Please initialize the log4j system properly.

log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

13:23:44.281 [main] INFO  org.springframework.boot.autoconfigure.security.reactive.ReactiveUserDetailsServiceAutoConfiguration - 

Using generated security password: 3e4a49ed-0d2c-4bc5-994e-30610ba3c397


13:23:44.355 [main] WARN  org.springframework.boot.autoconfigure.jackson.JacksonAutoConfiguration$JodaDateTimeJacksonConfiguration - Auto-configuration of Jackson's Joda-Time integration is deprecated in favor of using java.time (JSR-310).

13:23:44.559 [main] INFO  com.provectus.kafka.ui.serde.DeserializationService - Using SchemaRegistryAwareRecordSerDe for cluster 'cluster'

13:23:45.430 [main] INFO  org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler - Initializing ExecutorService 'taskScheduler'

13:23:45.565 [parallel-1] DEBUG com.provectus.kafka.ui.service.MetricsUpdateService - Start getting metrics for kafkaCluster: cluster

13:23:45.796 [main] INFO  org.springframework.boot.web.embedded.netty.NettyWebServer - Netty started on port(s): 8080

13:23:45.810 [main] INFO  com.provectus.kafka.ui.KafkaUiApplication - Started KafkaUiApplication in 4.614 seconds (JVM running for 6.591)

13:23:52.825 [boundedElastic-4] DEBUG org.springframework.http.codec.json.Jackson2JsonEncoder - [6524d03a] Encoding [[class Cluster {
    name: cluster
    defaultCluster: null
    status: offline
    brokerCount: nul (truncated)...]

13:24:09.155 [kafka-admin-client-thread | adminclient-2] WARN  com.provectus.kafka.ui.exception.ErrorCode - Multiple class com.provectus.kafka.ui.exception.ErrorCode values refer to code 4001

13:24:09.179 [kafka-admin-client-thread | adminclient-2] ERROR org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler - [de494e71]  500 Server Error for HTTP GET "/api/clusters/cluster/brokers"
java.util.NoSuchElementException: No value present
	at java.util.Optional.orElseThrow(Unknown Source) ~[?:?]
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
	|_ checkpoint ⇢ Handler com.provectus.kafka.ui.controller.BrokersController#getBrokers(String, ServerWebExchange) [DispatcherHandler]
	|_ checkpoint ⇢ com.provectus.kafka.ui.config.ReadOnlyModeFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ com.provectus.kafka.ui.config.CustomWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ org.springframework.security.web.server.authorization.AuthorizationWebFilter [DefaultWebFilterChain]

	|_ checkpoint ⇢ org.springframework.security.web.server.authorization.ExceptionTranslationWebFilter [DefaultWebFilterChain]

	|_ checkpoint ⇢ org.springframework.security.web.server.authentication.logout.LogoutWebFilter [DefaultWebFilterChain]

	|_ checkpoint ⇢ org.springframework.security.web.server.savedrequest.ServerRequestCacheWebFilter [DefaultWebFilterChain]

	|_ checkpoint ⇢ org.springframework.security.web.server.context.SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]

	|_ checkpoint ⇢ org.springframework.security.web.server.context.ReactorContextWebFilter [DefaultWebFilterChain]

	|_ checkpoint ⇢ org.springframework.security.web.server.header.HttpHeaderWriterWebFilter [DefaultWebFilterChain]

	|_ checkpoint ⇢ org.springframework.security.config.web.server.ServerHttpSecurity$ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]

	|_ checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]

	|_ checkpoint ⇢ HTTP GET "/api/clusters/cluster/brokers" [ExceptionHandlingWebHandler]

Stack trace:

		at java.util.Optional.orElseThrow(Unknown Source) ~[?:?]

		at com.provectus.kafka.ui.util.ClusterUtil.getClusterVersion(ClusterUtil.java:368) ~[classes!/:?]

		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]

		at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1637) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]

		at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]

		at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:156) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]

		at com.provectus.kafka.ui.util.ClusterUtil.lambda$toMono$0(ClusterUtil.java:64) ~[classes!/:?]

		at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:177) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:162) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.common.internals.KafkaFutureImpl$Applicant.accept(KafkaFutureImpl.java:65) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.common.internals.KafkaFutureImpl$Applicant.accept(KafkaFutureImpl.java:49) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.common.KafkaFuture$AllOfAdapter.maybeComplete(KafkaFuture.java:82) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:76) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:57) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.clients.admin.KafkaAdminClient$10.handleResponse(KafkaAdminClient.java:1911) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.handleResponses(KafkaAdminClient.java:1076) [kafka-clients-2.4.1.jar!/:?]

		at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1204) [kafka-clients-2.4.1.jar!/:?]

		at java.lang.Thread.run(Unknown Source) [?:?]

13:24:09.227 [kafka-admin-client-thread | adminclient-2] DEBUG org.springframework.http.codec.json.Jackson2JsonEncoder - [de494e71] Encoding [class ErrorResponse {
    code: 5000
    message: No value present
    timestamp: 1629725049160
     (truncated)...]

13:24:15.434 [parallel-1] DEBUG com.provectus.kafka.ui.service.MetricsUpdateService - Start getting metrics for kafkaCluster: cluster
```
> Unfortunately still an error:
> 
> ```
> 
> 
>   .   ____          _            __ _ _
>  /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
> ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
>  \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
>   '  |____| .__|_| |_|_| |_\__, | / / / /
> 
>  =========|_|==============|___/=/_/_/_/
> 
>  :: Spring Boot ::        (v2.2.4.RELEASE)
> 
> 
> 13:23:42.095 [main] INFO  com.provectus.kafka.ui.KafkaUiApplication - Starting KafkaUiApplication on 5a67d335b191 with PID 1 (/kafka-ui-api.jar started by root in /)
> 
> 13:23:42.096 [main] DEBUG com.provectus.kafka.ui.KafkaUiApplication - Running with Spring Boot v2.2.4.RELEASE, Spring v5.2.3.RELEASE
> 
> 13:23:42.097 [main] INFO  com.provectus.kafka.ui.KafkaUiApplication - No active profile set, falling back to default profiles: default
> 
> 13:23:42.539 [background-preinit] WARN  org.springframework.http.converter.json.Jackson2ObjectMapperBuilder - For Jackson Kotlin classes support please add "com.fasterxml.jackson.module:jackson-module-kotlin" to the classpath
> 
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in [jar:file:/kafka-ui-api.jar!/BOOT-INF/lib/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> 
> SLF4J: Found binding in [jar:file:/kafka-ui-api.jar!/BOOT-INF/lib/log4j-slf4j-impl-2.12.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
> 
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> 
> log4j:WARN No appenders could be found for logger (reactor.util.Loggers$LoggerFactory).
> log4j:WARN Please initialize the log4j system properly.
> 
> log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
> 
> 13:23:44.281 [main] INFO  org.springframework.boot.autoconfigure.security.reactive.ReactiveUserDetailsServiceAutoConfiguration - 
> 
> Using generated security password: 3e4a49ed-0d2c-4bc5-994e-30610ba3c397
> 
> 
> 13:23:44.355 [main] WARN  org.springframework.boot.autoconfigure.jackson.JacksonAutoConfiguration$JodaDateTimeJacksonConfiguration - Auto-configuration of Jackson's Joda-Time integration is deprecated in favor of using java.time (JSR-310).
> 
> 13:23:44.559 [main] INFO  com.provectus.kafka.ui.serde.DeserializationService - Using SchemaRegistryAwareRecordSerDe for cluster 'cluster'
> 
> 13:23:45.430 [main] INFO  org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler - Initializing ExecutorService 'taskScheduler'
> 
> 13:23:45.565 [parallel-1] DEBUG com.provectus.kafka.ui.service.MetricsUpdateService - Start getting metrics for kafkaCluster: cluster
> 
> 13:23:45.796 [main] INFO  org.springframework.boot.web.embedded.netty.NettyWebServer - Netty started on port(s): 8080
> 
> 13:23:45.810 [main] INFO  com.provectus.kafka.ui.KafkaUiApplication - Started KafkaUiApplication in 4.614 seconds (JVM running for 6.591)
> 
> 13:23:52.825 [boundedElastic-4] DEBUG org.springframework.http.codec.json.Jackson2JsonEncoder - [6524d03a] Encoding [[class Cluster {
>     name: cluster
>     defaultCluster: null
>     status: offline
>     brokerCount: nul (truncated)...]
> 
> 13:24:09.155 [kafka-admin-client-thread | adminclient-2] WARN  com.provectus.kafka.ui.exception.ErrorCode - Multiple class com.provectus.kafka.ui.exception.ErrorCode values refer to code 4001
> 
> 13:24:09.179 [kafka-admin-client-thread | adminclient-2] ERROR org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler - [de494e71]  500 Server Error for HTTP GET "/api/clusters/cluster/brokers"
> java.util.NoSuchElementException: No value present
> 	at java.util.Optional.orElseThrow(Unknown Source) ~[?:?]
> 	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
> Error has been observed at the following site(s):
> 	|_ checkpoint ⇢ Handler com.provectus.kafka.ui.controller.BrokersController#getBrokers(String, ServerWebExchange) [DispatcherHandler]
> 	|_ checkpoint ⇢ com.provectus.kafka.ui.config.ReadOnlyModeFilter [DefaultWebFilterChain]
> 	|_ checkpoint ⇢ com.provectus.kafka.ui.config.CustomWebFilter [DefaultWebFilterChain]
> 	|_ checkpoint ⇢ org.springframework.security.web.server.authorization.AuthorizationWebFilter [DefaultWebFilterChain]
> 
> 	|_ checkpoint ⇢ org.springframework.security.web.server.authorization.ExceptionTranslationWebFilter [DefaultWebFilterChain]
> 
> 	|_ checkpoint ⇢ org.springframework.security.web.server.authentication.logout.LogoutWebFilter [DefaultWebFilterChain]
> 
> 	|_ checkpoint ⇢ org.springframework.security.web.server.savedrequest.ServerRequestCacheWebFilter [DefaultWebFilterChain]
> 
> 	|_ checkpoint ⇢ org.springframework.security.web.server.context.SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]
> 
> 	|_ checkpoint ⇢ org.springframework.security.web.server.context.ReactorContextWebFilter [DefaultWebFilterChain]
> 
> 	|_ checkpoint ⇢ org.springframework.security.web.server.header.HttpHeaderWriterWebFilter [DefaultWebFilterChain]
> 
> 	|_ checkpoint ⇢ org.springframework.security.config.web.server.ServerHttpSecurity$ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
> 
> 	|_ checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
> 
> 	|_ checkpoint ⇢ HTTP GET "/api/clusters/cluster/brokers" [ExceptionHandlingWebHandler]
> 
> Stack trace:
> 
> 		at java.util.Optional.orElseThrow(Unknown Source) ~[?:?]
> 
> 		at com.provectus.kafka.ui.util.ClusterUtil.getClusterVersion(ClusterUtil.java:368) ~[classes!/:?]
> 
> 		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
> 
> 		at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1637) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
> 
> 		at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
> 
> 		at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:156) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
> 
> 		at com.provectus.kafka.ui.util.ClusterUtil.lambda$toMono$0(ClusterUtil.java:64) ~[classes!/:?]
> 
> 		at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:177) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:162) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.common.internals.KafkaFutureImpl$Applicant.accept(KafkaFutureImpl.java:65) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.common.internals.KafkaFutureImpl$Applicant.accept(KafkaFutureImpl.java:49) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.common.KafkaFuture$AllOfAdapter.maybeComplete(KafkaFuture.java:82) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:76) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:57) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.clients.admin.KafkaAdminClient$10.handleResponse(KafkaAdminClient.java:1911) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.handleResponses(KafkaAdminClient.java:1076) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1204) [kafka-clients-2.4.1.jar!/:?]
> 
> 		at java.lang.Thread.run(Unknown Source) [?:?]
> 
> 13:24:09.227 [kafka-admin-client-thread | adminclient-2] DEBUG org.springframework.http.codec.json.Jackson2JsonEncoder - [de494e71] Encoding [class ErrorResponse {
>     code: 5000
>     message: No value present
>     timestamp: 1629725049160
>      (truncated)...]
> 
> 13:24:15.434 [parallel-1] DEBUG com.provectus.kafka.ui.service.MetricsUpdateService - Start getting metrics for kafkaCluster: cluster
> ```

Hmm, looks strange.
I see in your config cluster name: cluster01
But from logs i see request to: /api/clusters/cluster/brokers
This looks suspicious. Is this error appears when you open UI, or you are querying from cli?
Could you query something like:  /api/clusters/cluster01/brokers
Apologies for the confusion with the cluster name, I changed it at some point, hence cluster is the correct name.
Im using the ui to access.

Just to be clear this is the docker command i am using:
docker run -p 8080:8080 \
  -e KAFKA_CLUSTERS_0_NAME=cluster \
  -e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=.aws.confluent.cloud:9092 \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=SASL_SSL \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM=PLAIN \
  -e KAFKA_CLUSTERS_0_PROPERTIES_CLIENT_DNS_LOOKUP=use_all_dns_ips \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG='org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";'\
  -d provectuslabs/kafka-ui:master

This seems to be the error when hitting /api/clusters/cluster/brokers:

```13:39:35.782 [kafka-admin-client-thread | adminclient-6] ERROR org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler - [338cab37] 500 Server Error for HTTP GET "/api/clusters/cluster/brokers"

java.util.NoSuchElementException: No value present

at java.util.Optional.orElseThrow(Unknown Source) ~[?:?]

Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:

Error has been observed at the following site(s):

|_ checkpoint ⇢ Handler com.provectus.kafka.ui.controller.BrokersController#getBrokers(String, ServerWebExchange) [DispatcherHandler]

|_ checkpoint ⇢ com.provectus.kafka.ui.config.ReadOnlyModeFilter [DefaultWebFilterChain]

|_ checkpoint ⇢ com.provectus.kafka.ui.config.CustomWebFilter [DefaultWebFilterChain]

|_ checkpoint ⇢ org.springframework.security.web.server.authorization.AuthorizationWebFilter [DefaultWebFilterChain]

|_ checkpoint ⇢ org.springframework.security.web.server.authorization.ExceptionTranslationWebFilter [DefaultWebFilterChain]


|_ checkpoint ⇢ org.springframework.security.web.server.authentication.logout.LogoutWebFilter [DefaultWebFilterChain]

|_ checkpoint ⇢ org.springframework.security.web.server.savedrequest.ServerRequestCacheWebFilter [DefaultWebFilterChain]

|_ checkpoint ⇢ org.springframework.security.web.server.context.SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]

|_ checkpoint ⇢ org.springframework.security.web.server.context.ReactorContextWebFilter [DefaultWebFilterChain]

|_ checkpoint ⇢ org.springframework.security.web.server.header.HttpHeaderWriterWebFilter [DefaultWebFilterChain]

|_ checkpoint ⇢ org.springframework.security.config.web.server.ServerHttpSecurity$ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]

|_ checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]

|_ checkpoint ⇢ HTTP GET "/api/clusters/cluster/brokers" [ExceptionHandlingWebHandler]

Stack trace:

at java.util.Optional.orElseThrow(Unknown Source) ~[?:?]

at com.provectus.kafka.ui.util.ClusterUtil.getClusterVersion(ClusterUtil.java:368) ~[classes!/:?]

at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]

at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1637) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]

at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]

at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:156) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]

at com.provectus.kafka.ui.util.ClusterUtil.lambda$toMono$0(ClusterUtil.java:64) ~[classes!/:?]

at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:177) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.common.internals.KafkaFutureImpl$WhenCompleteBiConsumer.accept(KafkaFutureImpl.java:162) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.common.internals.KafkaFutureImpl$Applicant.accept(KafkaFutureImpl.java:65) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.common.internals.KafkaFutureImpl$Applicant.accept(KafkaFutureImpl.java:49) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.common.KafkaFuture$AllOfAdapter.maybeComplete(KafkaFuture.java:82) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:76) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.common.KafkaFuture$AllOfAdapter.accept(KafkaFuture.java:57) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:221) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.clients.admin.KafkaAdminClient$10.handleResponse(KafkaAdminClient.java:1911) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.handleResponses(KafkaAdminClient.java:1076) [kafka-clients-2.4.1.jar!/:?]

at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1204) [kafka-clients-2.4.1.jar!/:?]

at java.lang.Thread.run(Unknown Source) [?:?]

13:39:35.788 [kafka-admin-client-thread | adminclient-6] DEBUG org.springframework.http.codec.json.Jackson2JsonEncoder - [338cab37] Encoding [class ErrorResponse {

code: 5000

message: No value present

timestamp: 1629725975782

(truncated)...]
```
According to this stacktrace, you don't have config inter.broker.protocol.version in your cluster. This looks strange. Let me check it. 
@david-webill I just updated master branch with fix Could you please pull latest image with tag master and check it?
Thank you @germanosin
I will try later today when I get a chance and let you know.
Hello

I have tried with the latest master

I get the following:


```
08:52:16.203 [parallel-1] DEBUG com.provectus.kafka.ui.service.MetricsUpdateService - Start getting metrics for kafkaCluster: cluster


08:52:16.942 [kafka-admin-client-thread | adminclient-1] DEBUG org.springframework.http.codec.json.Jackson2JsonEncoder - [be8f458f] Encoding [[class Broker {

id: 0

host: b0-pkc-q283m.af-south-1.aws.confluent.cloud

}, class Broker {

(truncated)...]

```
> Hello
> 
> I have tried with the latest master
> 
> I get the following:
> 
> ```
> 08:52:16.203 [parallel-1] DEBUG com.provectus.kafka.ui.service.MetricsUpdateService - Start getting metrics for kafkaCluster: cluster
> 
> 
> 08:52:16.942 [kafka-admin-client-thread | adminclient-1] DEBUG org.springframework.http.codec.json.Jackson2JsonEncoder - [be8f458f] Encoding [[class Broker {
> 
> id: 0
> 
> host: b0-pkc-q283m.af-south-1.aws.confluent.cloud
> 
> }, class Broker {
> 
> (truncated)...]
> ```

Hmm, looks lit it's working.
I also thought that, but nothing is showing up in the ui, it still says 1 cluster offline
Looks strange. We'll try to debug it with confluent cloud
I checked it with confluent cloud. Problem was in our algorithm collecting logdirs information to estimate topic size. I added specific config 
```
KAFKA_CLUSTERS_0_DISABLELOGDIRSCOLLECTION
```
If you'll set it to true, it will work fine, but without topics size information. We'll recheck it in future versions.

Pull request review comments:

Patch that resolved the issue:
diff --git a/README.md b/README.md
index db4df22f28b..b24ad772f3e 100644
--- a/README.md
+++ b/README.md
@@ -173,6 +173,7 @@ For example, if you want to use an environment variable to set the `name` parame
 |`KAFKA_CLUSTERS_0_SCHEMANAMETEMPLATE`  |How keys are saved to schemaRegistry
 |`KAFKA_CLUSTERS_0_JMXPORT`        	|Open jmxPosrts of a broker
 |`KAFKA_CLUSTERS_0_READONLY`        	|Enable read only mode. Default: false
+|`KAFKA_CLUSTERS_0_DISABLELOGDIRSCOLLECTION`        	|Disable collecting segments information. Should be true for confluent cloud. Default: false
 |`KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME` |Given name for the Kafka Connect cluster
 |`KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS` |Address of the Kafka Connect service endpoint 
 |`LOGGING_LEVEL_ROOT`        	| Setting log level (all, debug, info, warn, error, fatal, off). Default: debug
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
index 63925376cab..ef0fdd66f70 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
@@ -30,6 +30,7 @@ public static class Cluster {
     int jmxPort;
     Properties properties;
     boolean readOnly = false;
+    boolean disableLogDirsCollection = false;
   }
 
   @Data
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
index c87d4945407..9439eb26004 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
@@ -31,5 +31,6 @@ public class KafkaCluster {
   private final String protobufMessageName;
   private final Properties properties;
   private final Boolean readOnly;
+  private final Boolean disableLogDirsCollection;
   private final List<Feature> features;
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaService.java
index b966ee06794..e94596b4547 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaService.java
@@ -129,11 +129,21 @@ public Mono<KafkaCluster> getUpdatedCluster(KafkaCluster cluster) {
                     getClusterMetrics(ac.getAdminClient())
                         .flatMap(i -> fillJmxMetrics(i, cluster.getName(), ac.getAdminClient()))
                         .flatMap(clusterMetrics ->
-                            getTopicsData(ac.getAdminClient()).flatMap(it ->
-                                updateSegmentMetrics(ac.getAdminClient(), clusterMetrics, it)
+                            getTopicsData(ac.getAdminClient()).flatMap(it -> {
+                                  if (cluster.getDisableLogDirsCollection() == null
+                                      || !cluster.getDisableLogDirsCollection()) {
+                                    return updateSegmentMetrics(
+                                        ac.getAdminClient(), clusterMetrics, it
+                                    );
+                                  } else {
+                                    return emptySegmentMetrics(clusterMetrics, it);
+                                  }
+                                }
                             ).map(segmentSizeDto -> buildFromData(cluster, version, segmentSizeDto))
                         )
             )
+        ).doOnError(e ->
+            log.error("Failed to collect cluster {} info", cluster.getName(), e)
         ).onErrorResume(
             e -> Mono.just(cluster.toBuilder()
                 .status(ServerStatus.OFFLINE)
@@ -484,6 +494,28 @@ private InternalPartition mergeWithStats(String topic, InternalPartition partiti
         .build();
   }
 
+  private Mono<InternalSegmentSizeDto> emptySegmentMetrics(InternalClusterMetrics clusterMetrics,
+                                                            List<InternalTopic> internalTopics) {
+    return Mono.just(
+        InternalSegmentSizeDto.builder()
+        .clusterMetricsWithSegmentSize(
+            clusterMetrics.toBuilder()
+                .segmentSize(0)
+                .segmentCount(0)
+                .internalBrokerDiskUsage(Collections.emptyMap())
+                .build()
+        )
+        .internalTopicWithSegmentSize(
+            internalTopics.stream().collect(
+                Collectors.toMap(
+                    InternalTopic::getName,
+                    i -> i
+                )
+            )
+        ).build()
+    );
+  }
+
   private Mono<InternalSegmentSizeDto> updateSegmentMetrics(AdminClient ac,
                                                             InternalClusterMetrics clusterMetrics,
                                                             List<InternalTopic> internalTopics) {
@@ -491,9 +523,11 @@ private Mono<InternalSegmentSizeDto> updateSegmentMetrics(AdminClient ac,
         internalTopics.stream().map(InternalTopic::getName).collect(Collectors.toList());
     return ClusterUtil.toMono(ac.describeTopics(names).all()).flatMap(topic ->
         ClusterUtil.toMono(ac.describeCluster().nodes()).flatMap(nodes ->
+
             ClusterUtil.toMono(
-                ac.describeLogDirs(nodes.stream().map(Node::id).collect(Collectors.toList())).all())
-                .map(log -> {
+                ac.describeLogDirs(
+                    nodes.stream().map(Node::id).collect(Collectors.toList())).all()
+                ).map(log -> {
                   final List<Tuple3<Integer, TopicPartition, Long>> topicPartitions =
                       log.entrySet().stream().flatMap(b ->
                           b.getValue().entrySet().stream().flatMap(topicMap ->


Test patch: