From 032fbc49751ee498a1fb93dac12e4eb89696c6e3 Mon Sep 17 00:00:00 2001 From: Said Boudjelda Date: Fri, 30 Oct 2026 23:28:32 +0100 Subject: [PATCH 1/2] fix(topics): a created topic is described once the broker knows it POST /api/kafka/topics answered 500 for a topic it had just created on main's CI (d120b8cb, run 41136562284, KafkaApiIT): the topic was created at 22:20:46.135, and describing it right after failed with "Failed to describe topic". CreateTopics returns once the controller has the topic, and a broker can still answer it unknown until its metadata catches up, which a fresh test broker shows most. TopicService.describeCreatedTopic retries describeTopicDetail every 200 ms, up to 10 times, and the create endpoint uses it. A plain GET of a topic keeps its immediate answer. TopicServiceTest has the broker answer UnknownTopicOrPartition twice before it knows the topic; without the retry the test fails. MCP comment refs to TopicService.java re-pointed past the 12 new lines. --- cli/cmd/mcp_caveats.go | 2 +- cli/cmd/mcp_tools_cluster_topology.go | 4 ++-- .../kates/api/KafkaClientResource.java | 2 +- .../bmscomp/kates/service/TopicService.java | 12 ++++++++++ .../kates/service/TopicServiceTest.java | 24 +++++++++++++++++++ 5 files changed, 40 insertions(+), 4 deletions(-) diff --git a/cli/cmd/mcp_caveats.go b/cli/cmd/mcp_caveats.go index 8515fcea..6b90ba8a 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:182-210", "charts/kafka-cluster/values.yaml:137", }, }, diff --git a/cli/cmd/mcp_tools_cluster_topology.go b/cli/cmd/mcp_tools_cluster_topology.go index fb77b9ce..6b9395b8 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:182-210). 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:150-152,213-215; // 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 { 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..e24ea066 100644 --- a/kates/src/main/java/com/bmscomp/kates/service/TopicService.java +++ b/kates/src/main/java/com/bmscomp/kates/service/TopicService.java @@ -126,6 +126,18 @@ 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. + */ + @Retry(maxRetries = 10, delay = 200) + @Timeout(35_000) + public Map describeCreatedTopic(String topicName) { + return describeTopicDetail(topicName); + } + @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..7eae6f09 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,8 @@ 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.apache.kafka.common.internals.KafkaFutureImpl; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -154,6 +156,28 @@ 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()); + } + 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); } From d638298be82a2a6098376f820e2793061356e9f8 Mon Sep 17 00:00:00 2001 From: Said Boudjelda Date: Fri, 30 Oct 2026 23:37:51 +0100 Subject: [PATCH 2/2] fix(topics): retry a created topic's describe only while it is unknown Review on #564: describeCreatedTopic retried every failure, so a broker that does not answer held the request for 11 describe timeouts, about 5.5 minutes, before the 500. It now retries only the broker's UnknownTopicOrPartition answer, which it rethrows as such for @Retry's retryOn; anything else surfaces at once. TopicServiceTest checks that a timed-out describe is tried once. The new import moved TopicService.java down a line: MCP refs re-pointed, the anchored ones included. --- cli/cmd/mcp_caveats.go | 2 +- cli/cmd/mcp_tools_cluster.go | 4 ++-- cli/cmd/mcp_tools_cluster_topology.go | 6 +++--- .../com/bmscomp/kates/service/TopicService.java | 16 ++++++++++++++-- .../bmscomp/kates/service/TopicServiceTest.java | 13 +++++++++++++ 5 files changed, 33 insertions(+), 8 deletions(-) diff --git a/cli/cmd/mcp_caveats.go b/cli/cmd/mcp_caveats.go index 6b90ba8a..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:182-210", + 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 6b9395b8..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:182-210). 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:150-152,213-215; +// 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/service/TopicService.java b/kates/src/main/java/com/bmscomp/kates/service/TopicService.java index e24ea066..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; @@ -131,11 +132,22 @@ public Map describeTopics(Collection topicName * 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) + @Retry(maxRetries = 10, delay = 200, retryOn = UnknownTopicOrPartitionException.class) @Timeout(35_000) public Map describeCreatedTopic(String topicName) { - return describeTopicDetail(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) 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 7eae6f09..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,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.TimeoutException; import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; import org.apache.kafka.common.internals.KafkaFutureImpl; import org.junit.jupiter.api.BeforeEach; @@ -178,6 +179,18 @@ void aCreatedTopicIsDescribedOnceTheBrokerKnowsIt() { 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); }