fix(topics): a created topic is described once the broker knows it

cli/cmd/mcp_caveats.go

@@ -199,7 +199,7 @@ var mcpCaveatsCore = []mcpCaveat{
 			"it (2), is missing from a topic's configs. Its absence does not mean 1. A current backend reports the " +
 			"value in force wherever it is set, and what set it.",
 		Refs: []string{
-			mcpJava + "service/TopicService.java:170-198",
+			mcpJava + "service/TopicService.java:194-222",
 			"charts/kafka-cluster/values.yaml:137",
 		},
 	},

cli/cmd/mcp_tools_cluster.go

@@ -333,8 +333,8 @@ var mcpCaveatsCluster = []mcpCaveat{
 			"last 30 seconds can be missing from it and one deleted in that time can still be listed. Kafka's internal " +
 			"topics, whose names start __, are never listed.",
 		Refs: []string{
-			mcpAnchoredRef(mcpJava+"service/TopicService.java:32", "CACHE_TTL_MS = 30_000"),
-			mcpAnchoredRef(mcpJava+"service/TopicService.java:102-116", "public Set<String> listTopics()", "client.listTopics()"),
+			mcpAnchoredRef(mcpJava+"service/TopicService.java:33", "CACHE_TTL_MS = 30_000"),
+			mcpAnchoredRef(mcpJava+"service/TopicService.java:103-117", "public Set<String> listTopics()", "client.listTopics()"),
 			mcpJava + "api/ClusterResource.java:75-105",
 		},
 	},

cli/cmd/mcp_tools_cluster_topology.go

@@ -500,7 +500,7 @@ func mcpTopicLayoutFrom(call *mcpCall, d *client.TopicDetail, offset, limit int)
 	t.Leaders = mcpCap(call, leaders, mcpTopologyLeaderSummaries)
 
 	// Topic detail reports the value in force of each key it lists, and in
-	// configSources what set it (TopicService.java:170-198). A backend from
+	// configSources what set it (TopicService.java:194-222). A backend from
 	// before configSources kept only entries set on the topic or left at
 	// Kafka's default: there a present value is also the one in force, but an
 	// absent one is set at broker level and is not in the response.
@@ -578,7 +578,7 @@ const (
 // topic it cannot describe with HTTP 500 whether the topic is missing or Kafka
 // failed: TopicService wraps every failure as "Failed to describe topic", and
 // the resource answers 404 only for a message containing "not found", which
-// that one never does (TopicService.java:138-140,201-203;
+// that one never does (TopicService.java:162-164,225-227;
 // KafkaClientResource.java:115-123). So on a 5xx the topic list decides: a
 // topic missing from it is reported as not found, anything else is passed on.
 func mcpTopicDetailError(ctx context.Context, call *mcpCall, topic string, err error) error {
@@ -590,7 +590,7 @@ func mcpTopicDetailError(ctx context.Context, call *mcpCall, topic string, err e
 		return err
 	}
 	// The list is the backend's, cached for 30 seconds and never cleared
-	// early (TopicService.java:32,101-116), so a topic created since can be
+	// early (TopicService.java:33,102-117), so a topic created since can be
 	// missing from it. The common case is a name that does not exist, so the
 	// error is not marked retryable; the message names the exception.
 	return &mcpToolError{

kates/src/main/java/com/bmscomp/kates/api/KafkaClientResource.java

@@ -215,7 +215,7 @@ public Response createTopic(CreateTopicRequest request) {
             int partitions = request.partitions() > 0 ? request.partitions() : 1;
             int rf = request.replicationFactor() > 0 ? request.replicationFactor() : 1;
             topicService.createTopic(request.name(), partitions, rf, request.configs());
-            var detail = topicService.describeTopicDetail(request.name());
+            var detail = topicService.describeCreatedTopic(request.name());
             return Response.status(201).entity(detail).build();
         } catch (Exception e) {
             return Response.serverError()

kates/src/main/java/com/bmscomp/kates/service/TopicService.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartitionInfo;
 import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.eclipse.microprofile.faulttolerance.Retry;
 import org.eclipse.microprofile.faulttolerance.Timeout;
 import org.jboss.logging.Logger;
@@ -126,6 +127,29 @@ public Map<String, TopicDescription> describeTopics(Collection<String> topicName
         }
     }
 
+    /**
+     * describeTopicDetail for a topic createTopic has just made. CreateTopics
+     * returns once the controller has the topic, and a broker can still answer
+     * it unknown until its metadata catches up: main's CI (d120b8cb) got a 500
+     * from POST /api/kafka/topics for a topic it had created 0.1 s before.
+     * Only that answer is retried: any other failure, a broker that does not
+     * answer among them, surfaces at once rather than after 11 timeouts.
+     */
+    @Retry(maxRetries = 10, delay = 200, retryOn = UnknownTopicOrPartitionException.class)
+    @Timeout(35_000)
+    public Map<String, Object> describeCreatedTopic(String topicName) {
+        try {
+            return describeTopicDetail(topicName);
+        } catch (RuntimeException e) {
+            for (Throwable c = e; c != null; c = c.getCause()) {
+                if (c instanceof UnknownTopicOrPartitionException) {
+                    throw new UnknownTopicOrPartitionException("Topic not yet known to the broker: " + topicName, e);
+                }
+            }
+            throw e;
+        }
+    }
+
     @Timeout(35_000)
     public Map<String, Object> describeTopicDetail(String topicName) {
         AdminClient client = adminService.getClient();

kates/src/test/java/com/bmscomp/kates/service/TopicServiceTest.java

@@ -23,6 +23,9 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartitionInfo;
 import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -154,6 +157,40 @@ void topicDetailReportsEachKeyInForceWithWhereItComesFrom() {
                 detail.get("configSources"));
     }
 
+    @Test
+    void aCreatedTopicIsDescribedOnceTheBrokerKnowsIt() {
+        // As on main's CI: the broker answers the new topic unknown at first.
+        Node broker = new Node(0, "b0", 9092);
+        TopicDescription description = new TopicDescription(
+                "fresh", false, List.of(new TopicPartitionInfo(0, broker, List.of(broker), List.of(broker))));
+        KafkaFutureImpl<Map<String, TopicDescription>> failed = new KafkaFutureImpl<>();
+        failed.completeExceptionally(new UnknownTopicOrPartitionException("This server does not host this topic"));
+        DescribeTopicsResult unknown = mock(DescribeTopicsResult.class);
+        when(unknown.allTopicNames()).thenReturn(failed);
+        DescribeTopicsResult known = mock(DescribeTopicsResult.class);
+        when(known.allTopicNames()).thenReturn(KafkaFuture.completedFuture(Map.of("fresh", description)));
+        when(mockClient.describeTopics(anyCollection())).thenReturn(unknown, unknown, known);
+        DescribeConfigsResult configs = mock(DescribeConfigsResult.class);
+        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, "fresh");
+        when(configs.all()).thenReturn(KafkaFuture.completedFuture(Map.of(resource, new Config(List.of()))));
+        when(mockClient.describeConfigs(anyCollection())).thenReturn(configs);
+
+        assertEquals("fresh", topicService.describeCreatedTopic("fresh").get("name"));
+        verify(mockClient, times(3)).describeTopics(anyCollection());
+    }
+
+    @Test
+    void aCreatedTopicThatCannotBeDescribedForAnotherReasonFailsAtOnce() {
+        KafkaFutureImpl<Map<String, TopicDescription>> failed = new KafkaFutureImpl<>();
+        failed.completeExceptionally(new TimeoutException("Timed out waiting for a node assignment"));
+        DescribeTopicsResult unanswered = mock(DescribeTopicsResult.class);
+        when(unanswered.allTopicNames()).thenReturn(failed);
+        when(mockClient.describeTopics(anyCollection())).thenReturn(unanswered);
+
+        assertThrows(RuntimeException.class, () -> topicService.describeCreatedTopic("fresh"));
+        verify(mockClient, times(1)).describeTopics(anyCollection());
+    }
+
     private static ConfigEntry entry(String name, String value, ConfigEntry.ConfigSource source) {
         return new ConfigEntry(name, value, source, false, false, List.of(), ConfigEntry.ConfigType.STRING, null);
     }