[BE] Suppressing failures when describe log dirs not supported (#3262)

* Suppressing failures when describe log dirs not supported

README.md

@@ -199,7 +199,6 @@ For example, if you want to use an environment variable to set the `name` parame
 |`KAFKA_CLUSTERS_0_METRICS_PORT`        	 |Open metrics port of a broker
 |`KAFKA_CLUSTERS_0_METRICS_TYPE`        	 |Type of metrics retriever to use. Valid values are JMX (default) or PROMETHEUS. If Prometheus, then metrics are read from prometheus-jmx-exporter instead of jmx
 |`KAFKA_CLUSTERS_0_READONLY`        	|Enable read-only mode. Default: false
-|`KAFKA_CLUSTERS_0_DISABLELOGDIRSCOLLECTION`        	|Disable collecting segments information. It 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
 |`KAFKA_CLUSTERS_0_KAFKACONNECT_0_USERNAME`| Kafka Connect cluster's basic authentication username

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java

@@ -38,7 +38,6 @@ public class ClustersProperties {
     MetricsConfigData metrics;
     Properties properties;
     boolean readOnly = false;
-    boolean disableLogDirsCollection = false;
     List<SerdeConfig> serde = new ArrayList<>();
     String defaultKeySerde;
     String defaultValueSerde;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java

@@ -26,7 +26,6 @@ public class KafkaCluster {
   private final String bootstrapServers;
   private final Properties properties;
   private final boolean readOnly;
-  private final boolean disableLogDirsCollection;
   private final MetricsConfig metricsConfig;
   private final DataMasking masking;
   private final Supplier<PollingThrottler> throttler;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaClusterFactory.java

@@ -39,7 +39,6 @@ public class KafkaClusterFactory {
     builder.bootstrapServers(clusterProperties.getBootstrapServers());
     builder.properties(Optional.ofNullable(clusterProperties.getProperties()).orElse(new Properties()));
     builder.readOnly(clusterProperties.isReadOnly());
-    builder.disableLogDirsCollection(clusterProperties.isDisableLogDirsCollection());
     builder.masking(DataMasking.create(clusterProperties.getMasking()));
     builder.metricsConfig(metricsConfigDataToMetricsConfig(clusterProperties.getMetrics()));
     builder.throttler(PollingThrottler.throttlerSupplier(clusterProperties));

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ReactiveAdminClient.java

@@ -66,6 +66,7 @@ import org.apache.kafka.common.errors.GroupIdNotFoundException;
 import org.apache.kafka.common.errors.GroupNotEmptyException;
 import org.apache.kafka.common.errors.InvalidRequestException;
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.requests.DescribeLogDirsResponse;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -297,7 +298,12 @@ public class ReactiveAdminClient implements Closeable {
 
   public Mono<Map<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>>> describeLogDirs(
       Collection<Integer> brokerIds) {
-    return toMono(client.describeLogDirs(brokerIds).all());
+    return toMono(client.describeLogDirs(brokerIds).all())
+        .onErrorResume(UnsupportedVersionException.class, th -> Mono.just(Map.of()))
+        .onErrorResume(th -> true, th -> {
+          log.warn("Error while calling describeLogDirs", th);
+          return Mono.just(Map.of());
+        });
   }
 
   public Mono<ClusterDescription> describeCluster() {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/StatisticsService.java

@@ -1,5 +1,7 @@
 package com.provectus.kafka.ui.service;
 
+import static com.provectus.kafka.ui.service.ReactiveAdminClient.ClusterDescription;
+
 import com.provectus.kafka.ui.model.Feature;
 import com.provectus.kafka.ui.model.InternalLogDirStats;
 import com.provectus.kafka.ui.model.KafkaCluster;
@@ -9,10 +11,12 @@ import com.provectus.kafka.ui.model.Statistics;
 import com.provectus.kafka.ui.service.metrics.MetricsCollector;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.common.Node;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Mono;
 
@@ -21,7 +25,7 @@ import reactor.core.publisher.Mono;
 @Slf4j
 public class StatisticsService {
 
-  private final MetricsCollector metricsClusterUtil;
+  private final MetricsCollector metricsCollector;
   private final AdminClientService adminClientService;
   private final FeatureService featureService;
   private final StatisticsCache cache;
@@ -35,8 +39,8 @@ public class StatisticsService {
             ac.describeCluster().flatMap(description ->
                 Mono.zip(
                     List.of(
-                        metricsClusterUtil.getBrokerMetrics(cluster, description.getNodes()),
-                        getLogDirInfo(cluster, ac),
+                        metricsCollector.getBrokerMetrics(cluster, description.getNodes()),
+                        getLogDirInfo(description, ac),
                         featureService.getAvailableFeatures(cluster, description.getController()),
                         loadTopicConfigs(cluster),
                         describeTopics(cluster)),
@@ -58,11 +62,9 @@ public class StatisticsService {
             e -> Mono.just(Statistics.empty().toBuilder().lastKafkaException(e).build()));
   }
 
-  private Mono<InternalLogDirStats> getLogDirInfo(KafkaCluster cluster, ReactiveAdminClient c) {
-    if (!cluster.isDisableLogDirsCollection()) {
-      return c.describeLogDirs().map(InternalLogDirStats::new);
-    }
-    return Mono.just(InternalLogDirStats.empty());
+  private Mono<InternalLogDirStats> getLogDirInfo(ClusterDescription desc, ReactiveAdminClient ac) {
+    var brokerIds = desc.getNodes().stream().map(Node::id).collect(Collectors.toSet());
+    return ac.describeLogDirs(brokerIds).map(InternalLogDirStats::new);
   }
 
   private Mono<Map<String, TopicDescription>> describeTopics(KafkaCluster c) {