ISSUE-671 Fixed topic creation & deletion (#821)

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

@@ -1,10 +1,15 @@
 package com.provectus.kafka.ui.config;
 
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Profile;
+import org.springframework.core.io.ClassPathResource;
 import org.springframework.web.reactive.config.CorsRegistry;
 import org.springframework.web.reactive.config.EnableWebFlux;
 import org.springframework.web.reactive.config.WebFluxConfigurer;
+import org.springframework.web.reactive.function.server.RouterFunction;
+import org.springframework.web.reactive.function.server.RouterFunctions;
+import org.springframework.web.reactive.function.server.ServerResponse;
 
 @Configuration
 @EnableWebFlux
@@ -19,4 +24,22 @@ public class CorsGlobalConfiguration implements WebFluxConfigurer {
         .allowedHeaders("*")
         .allowCredentials(true);
   }
+
+  @Bean
+  public RouterFunction<ServerResponse> cssFilesRouter() {
+    return RouterFunctions
+        .resources("/static/css/**", new ClassPathResource("static/static/css/"));
+  }
+
+  @Bean
+  public RouterFunction<ServerResponse> jsFilesRouter() {
+    return RouterFunctions
+        .resources("/static/js/**", new ClassPathResource("static/static/js/"));
+  }
+
+  @Bean
+  public RouterFunction<ServerResponse> mediaFilesRouter() {
+    return RouterFunctions
+        .resources("/static/media/**", new ClassPathResource("static/static/media/"));
+  }
 }

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

@@ -247,7 +247,7 @@ public class ClusterService {
         .orElseThrow(TopicNotFoundException::new);
     if (cluster.getFeatures().contains(Feature.TOPIC_DELETION)) {
       return kafkaService.deleteTopic(cluster, topic.getName())
-          .doOnNext(t -> updateCluster(topicName, clusterName, cluster));
+          .doOnSuccess(t -> updateCluster(topicName, clusterName, cluster));
     } else {
       return Mono.error(new ValidationException("Topic deletion restricted"));
     }
@@ -315,9 +315,11 @@ public class ClusterService {
     return clustersStorage.getClusterByName(clusterName)
         .map(cluster -> adminClientService.getOrCreateAdminClient(cluster)
             .map(ExtendedAdminClient::getAdminClient)
-            .map(adminClient -> adminClient.deleteConsumerGroups(List.of(groupId)))
-            .map(DeleteConsumerGroupsResult::all)
-            .flatMap(ClusterUtil::toMono)
+            .flatMap(adminClient ->
+                ClusterUtil.toMono(
+                    adminClient.deleteConsumerGroups(List.of(groupId)).all()
+                )
+            )
             .onErrorResume(this::reThrowCustomException)
         )
         .orElse(Mono.empty());

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

@@ -2,7 +2,6 @@ package com.provectus.kafka.ui.service;
 
 import com.provectus.kafka.ui.config.ClustersProperties;
 import com.provectus.kafka.ui.mapper.ClusterMapper;
-import com.provectus.kafka.ui.model.Feature;
 import com.provectus.kafka.ui.model.KafkaCluster;
 import java.util.Collection;
 import java.util.HashMap;

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

@@ -290,12 +290,7 @@ public class KafkaService {
             topicData ->
                 getTopicsData(adminClient, Collections.singleton(topicData.getName()))
                     .next()
-        ).switchIfEmpty(Mono.error(new RuntimeException("Can't find created topic")))
-        .flatMap(t ->
-            loadTopicsConfig(adminClient, Collections.singletonList(t.getName()))
-                .map(c -> mergeWithConfigs(Collections.singletonList(t), c))
-                .map(m -> m.values().iterator().next())
-        );
+        ).switchIfEmpty(Mono.error(new RuntimeException("Can't find created topic")));
   }
 
   public Mono<InternalTopic> createTopic(KafkaCluster cluster, Mono<TopicCreation> topicCreation) {
@@ -306,8 +301,9 @@ public class KafkaService {
   public Mono<Void> deleteTopic(KafkaCluster cluster, String topicName) {
     return adminClientService.getOrCreateAdminClient(cluster)
         .map(ExtendedAdminClient::getAdminClient)
-        .map(adminClient -> adminClient.deleteTopics(List.of(topicName)))
-        .then();
+        .flatMap(adminClient ->
+                ClusterUtil.toMono(adminClient.deleteTopics(List.of(topicName)).all())
+        );
   }
 
   @SneakyThrows
@@ -667,7 +663,9 @@ public class KafkaService {
         .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
     return adminClientService.getOrCreateAdminClient(cluster)
         .map(ExtendedAdminClient::getAdminClient)
-        .map(ac -> ac.deleteRecords(records)).then();
+        .flatMap(ac ->
+            ClusterUtil.toMono(ac.deleteRecords(records).all())
+        );
   }
 
   public Mono<RecordMetadata> sendMessage(KafkaCluster cluster, String topic,

kafka-ui-api/src/main/resources/application-local.yml

@@ -1,9 +1,9 @@
 kafka:
   clusters:
     - name: local
-      bootstrapServers: localhost:9093
+      bootstrapServers: localhost:9092
       zookeeper: localhost:2181
-      schemaRegistry: http://localhost:8081
+      schemaRegistry: http://localhost:8085
       ksqldbServer: http://localhost:8088
       kafkaConnect:
         - name: first

kafka-ui-api/src/main/resources/application-sdp.yml

@@ -1,9 +1,11 @@
 kafka:
   clusters:
     - name: local
-      bootstrapServers: b-1.kad-msk.uxahxx.c6.kafka.eu-west-1.amazonaws.com:9092
+      bootstrapServers: b-1.kad-msk.57w67o.c6.kafka.eu-central-1.amazonaws.com:9094
+      properties:
+        security.protocol: SSL
 #      zookeeper: localhost:2181
-      schemaRegistry: http://kad-ecs-application-lb-857515197.eu-west-1.elb.amazonaws.com:9000/api/schema-registry
+#      schemaRegistry: http://kad-ecs-application-lb-857515197.eu-west-1.elb.amazonaws.com:9000/api/schema-registry
   #    -
   #      name: secondLocal
   #      zookeeper: zookeeper1:2181