[BE] Fix a case when controller is not found (#3030)

* ISSUE-3024: kui not working when controller not found
1. ReactiveAdminClient::getClusterVersion now takes into account that controller can not be visible
2. Uniq clientId generation added for consumers and adminClients
3. ReactiveAdminClient::toMonoWithExceptionFilter implementation simplified

* uniq adminclient clientId generation

Co-authored-by: iliax <ikuramshin@provectus.com>

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

@@ -37,7 +37,7 @@ public class AdminClientServiceImpl implements AdminClientService, Closeable {
       properties
           .put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.getBootstrapServers());
       properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, clientTimeout);
-      properties.putIfAbsent(AdminClientConfig.CLIENT_ID_CONFIG, "kafka-ui-app");
+      properties.putIfAbsent(AdminClientConfig.CLIENT_ID_CONFIG, "kafka-ui-admin-client-" + System.currentTimeMillis());
       return AdminClient.create(properties);
     })
         .flatMap(ReactiveAdminClient::create)

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

@@ -11,7 +11,6 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-import java.util.UUID;
 import java.util.function.ToIntFunction;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
@@ -197,7 +196,7 @@ public class ConsumerGroupService {
                                                     Map<String, Object> properties) {
     Properties props = new Properties();
     props.putAll(cluster.getProperties());
-    props.put(ConsumerConfig.CLIENT_ID_CONFIG, "kafka-ui-" + UUID.randomUUID());
+    props.put(ConsumerConfig.CLIENT_ID_CONFIG, "kafka-ui-consumer-" + System.currentTimeMillis());
     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.getBootstrapServers());
     props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, BytesDeserializer.class);
     props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BytesDeserializer.class);

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

@@ -25,9 +25,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletionException;
-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.function.Predicate;
@@ -260,37 +258,32 @@ public class ReactiveAdminClient implements Closeable {
    * such topics in resulting map.
    * <p/>
    * This method converts input map into Mono[Map] ignoring keys for which KafkaFutures
-   * finished with <code>clazz</code> exception.
+   * finished with <code>clazz</code> exception and empty Monos.
    */
   static <K, V> Mono<Map<K, V>> toMonoWithExceptionFilter(Map<K, KafkaFuture<V>> values,
-                                                           Class<? extends KafkaException> clazz) {
+                                                          Class<? extends KafkaException> clazz) {
     if (values.isEmpty()) {
       return Mono.just(Map.of());
     }
 
-    List<Mono<Tuple2<K, V>>> monos = values.entrySet().stream()
-        .map(e -> toMono(e.getValue()).map(r -> Tuples.of(e.getKey(), r)))
-        .collect(toList());
-
-    return Mono.create(sink -> {
-      var finishedCnt = new AtomicInteger();
-      var results = new ConcurrentHashMap<K, V>();
-      monos.forEach(mono -> mono.subscribe(
-          r -> {
-            results.put(r.getT1(), r.getT2());
-            if (finishedCnt.incrementAndGet() == monos.size()) {
-              sink.success(results);
-            }
-          },
-          th -> {
-            if (!th.getClass().isAssignableFrom(clazz)) {
-              sink.error(th);
-            } else if (finishedCnt.incrementAndGet() == monos.size()) {
-              sink.success(results);
-            }
-          }
-      ));
-    });
+    List<Mono<Tuple2<K, Optional<V>>>> monos = values.entrySet().stream()
+        .map(e ->
+            toMono(e.getValue())
+                .map(r -> Tuples.of(e.getKey(), Optional.of(r)))
+                .defaultIfEmpty(Tuples.of(e.getKey(), Optional.empty())) //tracking empty Monos
+                .onErrorResume(
+                    // tracking Monos with suppressible error
+                    th -> th.getClass().isAssignableFrom(clazz),
+                    th -> Mono.just(Tuples.of(e.getKey(), Optional.empty()))))
+        .toList();
+
+    return Mono.zip(
+        monos,
+        resultsArr -> Stream.of(resultsArr)
+            .map(obj -> (Tuple2<K, Optional<V>>) obj)
+            .filter(t -> t.getT2().isPresent()) //skipping empty & suppressible-errors
+            .collect(Collectors.toMap(Tuple2::getT1, t -> t.getT2().get()))
+    );
   }
 
   public Mono<Map<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>>> describeLogDirs() {
@@ -305,6 +298,10 @@ public class ReactiveAdminClient implements Closeable {
   }
 
   public Mono<ClusterDescription> describeCluster() {
+    return describeClusterImpl(client);
+  }
+
+  private static Mono<ClusterDescription> describeClusterImpl(AdminClient client) {
     var r = client.describeCluster();
     var all = KafkaFuture.allOf(r.nodes(), r.clusterId(), r.controller(), r.authorizedOperations());
     return Mono.create(sink -> all.whenComplete((res, ex) -> {
@@ -328,15 +325,20 @@ public class ReactiveAdminClient implements Closeable {
   }
 
   private static Mono<String> getClusterVersion(AdminClient client) {
-    return toMono(client.describeCluster().controller())
-        .flatMap(controller -> loadBrokersConfig(client, List.of(controller.id())))
-        .map(configs -> configs.values().stream()
+    return describeClusterImpl(client)
+        // choosing node from which we will get configs (starting with controller)
+        .flatMap(descr -> descr.controller != null
+            ? Mono.just(descr.controller)
+            : Mono.justOrEmpty(descr.nodes.stream().findFirst())
+        )
+        .flatMap(node -> loadBrokersConfig(client, List.of(node.id())))
+        .flatMap(configs -> configs.values().stream()
             .flatMap(Collection::stream)
             .filter(entry -> entry.name().contains("inter.broker.protocol.version"))
             .findFirst()
-            .map(ConfigEntry::value)
-            .orElse("1.0-UNKNOWN")
-        );
+            .map(configEntry -> Mono.just(configEntry.value()))
+            .orElse(Mono.empty()))
+        .switchIfEmpty(Mono.just("1.0-UNKNOWN"));
   }
 
   public Mono<Void> deleteConsumerGroups(Collection<String> groupIds) {
@@ -419,6 +421,7 @@ public class ReactiveAdminClient implements Closeable {
 
   /**
    * List offset for the topic's partitions and OffsetSpec.
+   *
    * @param failOnUnknownLeader true - throw exception in case of no-leader partitions,
    *                            false - skip partitions with no leader
    */
@@ -432,6 +435,7 @@ public class ReactiveAdminClient implements Closeable {
 
   /**
    * List offset for the specified partitions and OffsetSpec.
+   *
    * @param failOnUnknownLeader true - throw exception in case of no-leader partitions,
    *                            false - skip partitions with no leader
    */

kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ReactiveAdminClientTest.java

@@ -104,7 +104,14 @@ class ReactiveAdminClientTest extends AbstractIntegrationTest {
     var okFuture = new KafkaFutureImpl<String>();
     okFuture.complete("done");
 
-    Map<String, KafkaFuture<String>> arg = Map.of("failure", failedFuture, "ok", okFuture);
+    var emptyFuture = new KafkaFutureImpl<String>();
+    emptyFuture.complete(null);
+
+    Map<String, KafkaFuture<String>> arg = Map.of(
+        "failure", failedFuture,
+        "ok", okFuture,
+        "empty", emptyFuture
+    );
     StepVerifier.create(toMonoWithExceptionFilter(arg, UnknownTopicOrPartitionException.class))
         .assertNext(result -> assertThat(result).hasSize(1).containsEntry("ok", "done"))
         .verifyComplete();