diff --git a/cli/cmd/mcp_caveats.go b/cli/cmd/mcp_caveats.go index 8515fcea..89ec6042 100644 --- a/cli/cmd/mcp_caveats.go +++ b/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", }, }, diff --git a/cli/cmd/mcp_tools_cluster.go b/cli/cmd/mcp_tools_cluster.go index cf838557..c1479fbb 100644 --- a/cli/cmd/mcp_tools_cluster.go +++ b/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 listTopics()", "client.listTopics()"), + mcpAnchoredRef(mcpJava+"service/TopicService.java:33", "CACHE_TTL_MS = 30_000"), + mcpAnchoredRef(mcpJava+"service/TopicService.java:103-117", "public Set listTopics()", "client.listTopics()"), mcpJava + "api/ClusterResource.java:75-105", }, }, diff --git a/cli/cmd/mcp_tools_cluster_topology.go b/cli/cmd/mcp_tools_cluster_topology.go index fb77b9ce..4f9988d1 100644 --- a/cli/cmd/mcp_tools_cluster_topology.go +++ b/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{ diff --git a/kates/src/main/java/com/bmscomp/kates/api/KafkaClientResource.java b/kates/src/main/java/com/bmscomp/kates/api/KafkaClientResource.java index 9ae80f22..e0d2426d 100644 --- a/kates/src/main/java/com/bmscomp/kates/api/KafkaClientResource.java +++ b/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() diff --git a/kates/src/main/java/com/bmscomp/kates/service/TopicService.java b/kates/src/main/java/com/bmscomp/kates/service/TopicService.java index 132c5adc..950ca494 100644 --- a/kates/src/main/java/com/bmscomp/kates/service/TopicService.java +++ b/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 describeTopics(Collection 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 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 describeTopicDetail(String topicName) { AdminClient client = adminService.getClient(); diff --git a/kates/src/test/java/com/bmscomp/kates/service/TopicServiceTest.java b/kates/src/test/java/com/bmscomp/kates/service/TopicServiceTest.java index 63f4571f..ff909dba 100644 --- a/kates/src/test/java/com/bmscomp/kates/service/TopicServiceTest.java +++ b/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> 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> 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); }