Repository: provectus/kafka-ui
Issue: Error creating new topic
Labels: type/bug
scope/backend
scope/frontend

**Describe the bug** 
Running with `docker-compose`, local cluster is up after a few seconds. 
So i was about to create a new _topic_ via UI `Add a topic`.
- Next i clicked `Add Custom Parameter`
- Clicked `Custom Parameter` to have a look, but i did not select anything
- Clicked the big red `-` button on the right to remove the _parameter row_
- Clicked the button at the buttom to finish creating a new topic
- Error came in with _code_ `500` and _message_ `message.downconversion.enabled` has to be either true or false

Just out of curiosity, i tried another creation with the same pattern, this time another error was show
```shell
Schema docker-compose
500 Unknown topic config name:
```


**Set up**
Running via `docker-compose`, configuration listed below.

```yml
version: "2"
services:
    zookeeper:
        image: confluentinc/cp-zookeeper:latest
        container_name: kafka-zookeeper
        environment:
            ZOOKEEPER_CLIENT_PORT: 2181
            ZOOKEEPER_TICK_TIME: 2000
        ports:
            - 22181:2181

    kafka:
        image: confluentinc/cp-kafka:latest
        container_name: kafka-core
        depends_on:
            - zookeeper
        ports:
            - 29092:29092
        environment:
            KAFKA_BROKER_ID: 1
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
            KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
            KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
            KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

    kafka-ui:
        image: provectuslabs/kafka-ui
        container_name: kafka-ui
        ports:
            - "8080:8080"
        restart: always
        environment:
            - KAFKA_CLUSTERS_0_NAME=local
            - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092
            - KAFKA_CLUSTERS_0_ZOOKEEPER=localhost:2181

```



**Steps to Reproduce**
Steps to reproduce the behavior:

Please see athe top section **Describe the bug**.


**Expected behavior**
I was expecting to be creating a new topic for a local cluster.

**Note:** After leaving out the `Add Custom Parameter` section, creating topics is working fine.

Issue comments:
@anjeyy thank you for creating this issue. This is definitely a bug. We'll fix it in 0.2 version
I am having the same error on topic creation:

```
18:22:49.274 [kafka-admin-client-thread | adminclient-1] ERROR org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler - [f527d091]  500 Server Error for HTTP POST "/api/clusters/stage/topics"
org.apache.kafka.common.errors.UnknownTopicOrPartitionException:
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Error has been observed at the following site(s):
	|_ checkpoint ⇢ Handler com.provectus.kafka.ui.controller.TopicsController#createTopic(String, Mono, ServerWebExchange) [DispatcherHandler]
	|_ checkpoint ⇢ com.provectus.kafka.ui.config.CustomWebFilter [DefaultWebFilterChain]
	|_ checkpoint ⇢ com.provectus.kafka.ui.config.ReadOnlyModeFilter [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 POST "/api/clusters/stage/topics" [ExceptionHandlingWebHandler]
```

The topic seems to get created anyways.
The problem was in requesting configs, right after topic creation. At this point of time cluster is not ready to response it. Fixed in master.

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/CorsGlobalConfiguration.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/CorsGlobalConfiguration.java
index bbb743daac1..6e46b607bcb 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/CorsGlobalConfiguration.java
+++ b/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 void addCorsMappings(CorsRegistry registry) {
         .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/"));
+  }
 }
\ No newline at end of file
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClusterService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClusterService.java
index 026c4bb8967..256aa907cb0 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClusterService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClusterService.java
@@ -247,7 +247,7 @@ public Mono<Void> deleteTopic(String clusterName, String topicName) {
         .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 Mono<Void> deleteConsumerGroupById(String clusterName,
     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());
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClustersStorage.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClustersStorage.java
index 18267df0ee4..6c58470e59b 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClustersStorage.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ClustersStorage.java
@@ -2,7 +2,6 @@
 
 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;
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 b03b19296c4..b966ee06794 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
@@ -290,12 +290,7 @@ public Mono<InternalTopic> createTopic(AdminClient adminClient,
             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 Mono<InternalTopic> createTopic(KafkaCluster cluster, Mono<TopicCreation>
   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 Mono<Void> deleteTopicMessages(KafkaCluster cluster, Map<TopicPartition,
         .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,
diff --git a/kafka-ui-api/src/main/resources/application-local.yml b/kafka-ui-api/src/main/resources/application-local.yml
index 31190a77880..9b1458e1eda 100644
--- a/kafka-ui-api/src/main/resources/application-local.yml
+++ b/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
diff --git a/kafka-ui-api/src/main/resources/application-sdp.yml b/kafka-ui-api/src/main/resources/application-sdp.yml
index 71af1078f9d..2de30ccb7ca 100644
--- a/kafka-ui-api/src/main/resources/application-sdp.yml
+++ b/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


Test patch: