Repository: provectus/kafka-ui
Issue: TopicNotFound upon creating a topic
Labels: type/bug
scope/backend
status/accepted
status/confirmed

**Describe the bug**

A topic is not created from the first time in KRaft mode. Most of the time. Sometimes a topic is created immediately.

**Set up**
![изображение](https://user-images.githubusercontent.com/630071/154234076-0f70bc96-4809-4c39-8498-eddd531b4006.png)

**Steps to Reproduce**
Steps to reproduce the behavior:

![изображение](https://user-images.githubusercontent.com/630071/154232665-ab3e95aa-1819-49e9-a487-3361d9492acf.png)

**Expected behavior**

A topic is created immediately without errors 

**Screenshots**
<!--
(If applicable, add screenshots to help explain your problem)
-->


**Additional context**
<!--
(Add any other context about the problem here)
-->


Issue comments:
Hello there pikhovkin! 👋

Thank you and congratulations 🎉 for opening your very first issue in this project! 💖

In case you want to claim this issue, please comment down below! We will try to get back to you as soon as we can. 👀
Hey, thanks for raising the issue.

Could you share the details of a red (failed) request in network tab of browser's developer console?
@Haarolean 
![изображение](https://user-images.githubusercontent.com/630071/154239054-ccf0dea7-3df4-4942-9fff-c9f20bc33d79.png)
![изображение](https://user-images.githubusercontent.com/630071/154239075-8ed8f17a-3692-4061-9e3c-88f416d41523.png)


Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumerGroupService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumerGroupService.java
index cba51a921af..beb9f849798 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumerGroupService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumerGroupService.java
@@ -210,6 +210,7 @@ public KafkaConsumer<Bytes, Bytes> createConsumer(KafkaCluster cluster,
     props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BytesDeserializer.class);
     props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
     props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+    props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "false");
     props.putAll(properties);
 
     return new KafkaConsumer<>(props);
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java
index b4629cf39c6..265103a7cb1 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java
@@ -71,6 +71,10 @@ public class TopicsService {
   private int recreateMaxRetries;
   @Value("${topic.recreate.delay.seconds:1}")
   private int recreateDelayInSeconds;
+  @Value("${topic.load.after.create.maxRetries:10}")
+  private int loadTopicAfterCreateRetries;
+  @Value("${topic.load.after.create.delay.ms:500}")
+  private int loadTopicAfterCreateDelayInMs;
 
   public Mono<TopicsResponseDTO> getTopics(KafkaCluster cluster,
                                            Optional<Integer> pageNum,
@@ -115,7 +119,32 @@ private Mono<List<InternalTopic>> loadTopics(KafkaCluster c, List<String> topics
 
   private Mono<InternalTopic> loadTopic(KafkaCluster c, String topicName) {
     return loadTopics(c, List.of(topicName))
-        .map(lst -> lst.stream().findFirst().orElseThrow(TopicNotFoundException::new));
+        .flatMap(lst -> lst.stream().findFirst()
+            .map(Mono::just)
+            .orElse(Mono.error(TopicNotFoundException::new)));
+  }
+
+  /**
+   *  After creation topic can be invisible via API for some time.
+   *  To workaround this, we retyring topic loading until it becomes visible.
+   */
+  private Mono<InternalTopic> loadTopicAfterCreation(KafkaCluster c, String topicName) {
+    return loadTopic(c, topicName)
+        .retryWhen(
+            Retry
+                .fixedDelay(
+                    loadTopicAfterCreateRetries,
+                    Duration.ofMillis(loadTopicAfterCreateDelayInMs)
+                )
+                .filter(TopicNotFoundException.class::isInstance)
+                .onRetryExhaustedThrow((spec, sig) ->
+                    new TopicMetadataException(
+                        String.format(
+                            "Error while loading created topic '%s' - topic is not visible via API "
+                                + "after waiting for %d ms.",
+                            topicName,
+                            loadTopicAfterCreateDelayInMs * loadTopicAfterCreateRetries)))
+        );
   }
 
   private List<InternalTopic> createList(List<String> orderedNames,
@@ -182,7 +211,7 @@ private Mono<InternalTopic> createTopic(KafkaCluster c, ReactiveAdminClient admi
             ).thenReturn(topicData)
         )
         .onErrorResume(t -> Mono.error(new TopicMetadataException(t.getMessage())))
-        .flatMap(topicData -> loadTopic(c, topicData.getName()));
+        .flatMap(topicData -> loadTopicAfterCreation(c, topicData.getName()));
   }
 
   public Mono<TopicDTO> createTopic(KafkaCluster cluster, Mono<TopicCreationDTO> topicCreation) {
@@ -194,23 +223,30 @@ public Mono<TopicDTO> createTopic(KafkaCluster cluster, Mono<TopicCreationDTO> t
   public Mono<TopicDTO> recreateTopic(KafkaCluster cluster, String topicName) {
     return loadTopic(cluster, topicName)
         .flatMap(t -> deleteTopic(cluster, topicName)
-            .thenReturn(t).delayElement(Duration.ofSeconds(recreateDelayInSeconds))
-            .flatMap(topic -> adminClientService.get(cluster).flatMap(ac -> ac.createTopic(topic.getName(),
-                        topic.getPartitionCount(),
-                        (short) topic.getReplicationFactor(),
-                        topic.getTopicConfigs()
-                            .stream()
-                            .collect(Collectors
-                                .toMap(InternalTopicConfig::getName,
-                                    InternalTopicConfig::getValue)))
-                    .thenReturn(topicName))
-                .retryWhen(Retry.fixedDelay(recreateMaxRetries,
-                        Duration.ofSeconds(recreateDelayInSeconds))
-                    .filter(TopicExistsException.class::isInstance)
-                    .onRetryExhaustedThrow((a, b) ->
-                        new TopicRecreationException(topicName,
-                            recreateMaxRetries * recreateDelayInSeconds)))
-                .flatMap(a -> loadTopic(cluster, topicName)).map(clusterMapper::toTopic)
+            .thenReturn(t)
+            .delayElement(Duration.ofSeconds(recreateDelayInSeconds))
+            .flatMap(topic ->
+                adminClientService.get(cluster)
+                    .flatMap(ac ->
+                        ac.createTopic(
+                                topic.getName(),
+                                topic.getPartitionCount(),
+                                (short) topic.getReplicationFactor(),
+                                topic.getTopicConfigs()
+                                    .stream()
+                                    .collect(Collectors.toMap(InternalTopicConfig::getName,
+                                        InternalTopicConfig::getValue))
+                            )
+                            .thenReturn(topicName)
+                    )
+                    .retryWhen(
+                        Retry.fixedDelay(recreateMaxRetries, Duration.ofSeconds(recreateDelayInSeconds))
+                            .filter(TopicExistsException.class::isInstance)
+                            .onRetryExhaustedThrow((a, b) ->
+                                new TopicRecreationException(topicName,
+                                    recreateMaxRetries * recreateDelayInSeconds))
+                    )
+                    .flatMap(a -> loadTopicAfterCreation(cluster, topicName)).map(clusterMapper::toTopic)
             )
         );
   }
@@ -431,13 +467,21 @@ public TopicMessageSchemaDTO getTopicSchema(KafkaCluster cluster, String topicNa
   public Mono<TopicDTO> cloneTopic(
       KafkaCluster cluster, String topicName, String newTopicName) {
     return loadTopic(cluster, topicName).flatMap(topic ->
-        adminClientService.get(cluster).flatMap(ac -> ac.createTopic(newTopicName,
-            topic.getPartitionCount(),
-            (short) topic.getReplicationFactor(),
-            topic.getTopicConfigs()
-                .stream()
-                .collect(Collectors.toMap(InternalTopicConfig::getName, InternalTopicConfig::getValue)))
-        ).thenReturn(newTopicName).flatMap(a -> loadTopic(cluster, newTopicName)).map(clusterMapper::toTopic));
+        adminClientService.get(cluster)
+            .flatMap(ac ->
+                ac.createTopic(
+                    newTopicName,
+                    topic.getPartitionCount(),
+                    (short) topic.getReplicationFactor(),
+                    topic.getTopicConfigs()
+                        .stream()
+                        .collect(Collectors
+                            .toMap(InternalTopicConfig::getName, InternalTopicConfig::getValue))
+                )
+            ).thenReturn(newTopicName)
+            .flatMap(a -> loadTopicAfterCreation(cluster, newTopicName))
+            .map(clusterMapper::toTopic)
+    );
   }
 
   @VisibleForTesting


Test patch: