Merge branch 'main' into issues/848

Author: germanosin

README.md

@@ -18,6 +18,7 @@ Versatile, fast and lightweight web UI for managing Apache Kafka® clusters.
     <a href="https://ui.docs.kafbat.io/quick-start/demo-run">Quick Start</a> • 
     <a href="https://discord.gg/4DWzD7pGE5">Community</a>
     <br/>
+    <a href="https://aws.amazon.com/marketplace/pp/prodview-6tdqqzzjwmejq">AWS Marketplace</a>  •
     <a href="https://www.producthunt.com/products/ui-for-apache-kafka/reviews/new">ProductHunt</a>
 </p>
 

api/src/main/java/io/kafbat/ui/service/AdminClientServiceImpl.java

@@ -8,6 +8,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import lombok.extern.slf4j.Slf4j;
@@ -40,7 +41,7 @@ public Mono<ReactiveAdminClient> get(KafkaCluster cluster) {
   }
 
   private Mono<ReactiveAdminClient> createAdminClient(KafkaCluster cluster) {
-    return Mono.fromSupplier(() -> {
+    return Mono.fromFuture(CompletableFuture.supplyAsync(() -> {
       Properties properties = new Properties();
       KafkaClientSslPropertiesUtil.addKafkaSslProperties(cluster.getOriginalProperties().getSsl(), properties);
       properties.putAll(cluster.getProperties());
@@ -51,7 +52,7 @@ private Mono<ReactiveAdminClient> createAdminClient(KafkaCluster cluster) {
           "kafbat-ui-admin-" + Instant.now().getEpochSecond() + "-" + CLIENT_ID_SEQ.incrementAndGet()
       );
       return AdminClient.create(properties);
-    }).flatMap(ac -> ReactiveAdminClient.create(ac).doOnError(th -> ac.close()))
+    })).flatMap(ac -> ReactiveAdminClient.create(ac).doOnError(th -> ac.close()))
         .onErrorMap(th -> new IllegalStateException(
             "Error while creating AdminClient for the cluster " + cluster.getName(), th));
   }