This is an automated email from the ASF dual-hosted git repository.
mchades pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 63455e538e [Cherry-pick to branch-1.3] [#14574] fix(kafka): validate
topic names and map invalid topic errors (#14575) (#14620)
63455e538e is described below
commit 63455e538eb678e609818efedff3ab797b9edcf7
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Nov 5 08:56:05 2026 +0800
[Cherry-pick to branch-1.3] [#14574] fix(kafka): validate topic names and
map invalid topic errors (#14575) (#14620)
**Cherry-pick Information:**
- Original commit: 948964abb9ae9801dd47c7138c3fa2e52e9672a3 (#14575)
- Target branch: `branch-1.3`
- Status: Conflicts resolved.
Adds Kafka topic-name validation and maps broker `InvalidTopicException`
errors to HTTP 400. The backport preserves the original commit's changes
and the target branch's existing invalid-partition handling and tests.
**Validation:**
- `./gradlew spotlessApply`
- `./gradlew :catalogs:catalog-kafka:check
:catalogs:catalog-kafka:javadoc :docs:build -PskipITs`
- All 23 Kafka tests passed, including the broker name-collision
regression; OpenAPI validation passed.
- Docker integration tests were not run.
Co-authored-by: mchades <[email protected]>
---
catalogs/catalog-kafka/build.gradle.kts | 2 +
.../catalog/kafka/KafkaCatalogCapability.java | 27 +++++
.../catalog/kafka/KafkaCatalogOperations.java | 4 +
.../catalog/kafka/TestKafkaCatalogCapability.java | 128 +++++++++++++++++++++
.../catalog/kafka/TestKafkaCatalogOperations.java | 54 +++++++++
docs/kafka-catalog.md | 14 +++
docs/open-api/topics.yaml | 2 +
7 files changed, 231 insertions(+)
diff --git a/catalogs/catalog-kafka/build.gradle.kts
b/catalogs/catalog-kafka/build.gradle.kts
index 4bfd96d25a..6a083153ab 100644
--- a/catalogs/catalog-kafka/build.gradle.kts
+++ b/catalogs/catalog-kafka/build.gradle.kts
@@ -45,12 +45,14 @@ dependencies {
testImplementation(libs.commons.io)
testImplementation(libs.commons.lang3)
testImplementation(libs.curator.test)
+ testImplementation(libs.javax.ws.rs.api)
testImplementation(libs.junit.jupiter.api)
testImplementation(libs.kafka)
testImplementation(libs.mockito.core)
testImplementation(libs.mockito.inline)
testImplementation(libs.mysql.driver)
testImplementation(libs.postgresql.driver)
+ testImplementation(libs.servlet)
testImplementation(libs.testcontainers)
testImplementation(libs.testcontainers.mysql)
diff --git
a/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogCapability.java
b/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogCapability.java
index b78e9ccb21..15412eb489 100644
---
a/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogCapability.java
+++
b/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogCapability.java
@@ -23,6 +23,8 @@ import org.apache.gravitino.connector.capability.Capability;
import org.apache.gravitino.connector.capability.CapabilityResult;
public class KafkaCatalogCapability implements Capability {
+ private static final String TOPIC_NAME_PATTERN = "[a-zA-Z0-9_-]{1,128}";
+
@Override
public CapabilityResult managedStorage(Scope scope) {
if (Objects.requireNonNull(scope) == Scope.SCHEMA) {
@@ -31,4 +33,29 @@ public class KafkaCatalogCapability implements Capability {
return CapabilityResult.unsupported(
String.format("Kafka catalog does not support managed storage for
%s.", scope));
}
+
+ /**
+ * Checks topic names against Kafka's character restrictions and Gravitino's
compatibility limits.
+ * Other scopes use Gravitino's default name rules.
+ *
+ * <p>Topic names cannot contain dots because Gravitino authorization does
not support them, and
+ * cannot exceed 128 characters because of the metadata store's name length
limit.
+ *
+ * @param scope The scope of the capability.
+ * @param name The name to check.
+ * @return Whether the name is supported in the given scope.
+ */
+ @Override
+ public CapabilityResult specificationOnName(Scope scope, String name) {
+ if (scope != Scope.TOPIC) {
+ return Capability.super.specificationOnName(scope, name);
+ }
+
+ if (!name.matches(TOPIC_NAME_PATTERN)) {
+ return CapabilityResult.unsupported(
+ "Kafka topic names in Gravitino must be 1 to 128 characters long and
contain only ASCII"
+ + " alphanumerics, '_' and '-'.");
+ }
+ return CapabilityResult.SUPPORTED;
+ }
}
diff --git
a/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java
b/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java
index f234d44fad..d0a8e007f6 100644
---
a/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java
+++
b/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java
@@ -88,6 +88,7 @@ import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.errors.InvalidConfigurationException;
import org.apache.kafka.common.errors.InvalidPartitionsException;
import org.apache.kafka.common.errors.InvalidReplicationFactorException;
+import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.errors.TopicExistsException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.slf4j.Logger;
@@ -296,6 +297,9 @@ public class KafkaCatalogOperations implements
CatalogOperations, SupportsSchema
} else if (e.getCause() instanceof InvalidPartitionsException) {
throw ExceptionMessages.illegalArgument("Invalid partition count for
topic " + ident, e);
+ } else if (e.getCause() instanceof InvalidTopicException) {
+ throw ExceptionMessages.illegalArgument("Invalid topic name " + ident,
e);
+
} else if (e.getCause() instanceof InvalidReplicationFactorException) {
throw ExceptionMessages.illegalArgument("Invalid replication factor
for topic " + ident, e);
diff --git
a/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogCapability.java
b/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogCapability.java
new file mode 100644
index 0000000000..92976d9da8
--- /dev/null
+++
b/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogCapability.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.catalog.kafka;
+
+import java.util.Map;
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.catalog.CatalogManager;
+import org.apache.gravitino.catalog.TopicDispatcher;
+import org.apache.gravitino.catalog.TopicNormalizeDispatcher;
+import org.apache.gravitino.connector.capability.Capability;
+import org.apache.gravitino.connector.capability.Capability.Scope;
+import org.apache.gravitino.connector.capability.CapabilityResult;
+import org.apache.gravitino.dto.requests.TopicCreateRequest;
+import org.apache.gravitino.dto.responses.ErrorConstants;
+import org.apache.gravitino.dto.responses.ErrorResponse;
+import org.apache.gravitino.server.web.rest.TopicOperations;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+/** Tests Kafka topic name validation and its REST error response. */
+public class TestKafkaCatalogCapability {
+
+ private final KafkaCatalogCapability capability = new
KafkaCatalogCapability();
+
+ @Test
+ void testValidTopicNames() {
+ for (String name :
+ new String[] {
+ "a",
+ "Topic_123-test",
+ "_topic",
+ "-topic",
+ "-",
+ "a".repeat(64),
+ "a".repeat(65),
+ "a".repeat(128)
+ }) {
+ Assertions.assertTrue(capability.specificationOnName(Scope.TOPIC,
name).supported(), name);
+ }
+ }
+
+ @Test
+ void testKafkaIllegalTopicNames() {
+ for (String name :
+ new String[] {"", "*", ".", "..", "a/b", "a=b", "a/b=c", "a b", "a$b",
"caf\u00e9"}) {
+ CapabilityResult result = capability.specificationOnName(Scope.TOPIC,
name);
+ Assertions.assertFalse(result.supported(), name);
+ Assertions.assertEquals(
+ "Kafka topic names in Gravitino must be 1 to 128 characters long and
contain only ASCII"
+ + " alphanumerics, '_' and '-'.",
+ result.unsupportedMessage());
+ }
+ }
+
+ @Test
+ void testGravitinoTopicNameRestrictions() {
+ for (String name : new String[] {"a.b", "a".repeat(129)}) {
+ CapabilityResult result = capability.specificationOnName(Scope.TOPIC,
name);
+ Assertions.assertFalse(result.supported(), name);
+ Assertions.assertTrue(result.unsupportedMessage().contains("1 to 128
characters"));
+ }
+ }
+
+ @Test
+ void testOtherScopesUseDefaultNameSpecification() {
+ for (Scope scope : Scope.values()) {
+ if (scope == Scope.TOPIC) {
+ continue;
+ }
+ for (String name :
+ new String[] {"a/b", "a=b", "a.b", "*", "-topic", "a".repeat(65),
"a".repeat(128)}) {
+ CapabilityResult expected =
Capability.DEFAULT.specificationOnName(scope, name);
+ CapabilityResult actual = capability.specificationOnName(scope, name);
+ Assertions.assertEquals(expected.supported(), actual.supported(),
scope + ": " + name);
+ Assertions.assertEquals(
+ expected.unsupportedMessage(), actual.unsupportedMessage(), scope
+ ": " + name);
+ }
+ }
+ }
+
+ @Test
+ void testCreateTopicRejectsIllegalNamesBeforeDispatch() throws
IllegalAccessException {
+ CatalogManager catalogManager = Mockito.mock(CatalogManager.class);
+ Mockito.when(catalogManager.doWithCatalog(Mockito.any(),
Mockito.any())).thenReturn(capability);
+ TopicDispatcher dispatcher = Mockito.mock(TopicDispatcher.class);
+ TopicOperations operations =
+ new TopicOperations(new TopicNormalizeDispatcher(dispatcher,
catalogManager));
+ FieldUtils.writeField(operations, "httpRequest",
Mockito.mock(HttpServletRequest.class), true);
+
+ Map<String, String> properties = Map.of("partition-count", "1",
"replication-factor", "1");
+ for (String name : new String[] {"a/b", "a=b", "a.b", "a".repeat(129)}) {
+ TopicCreateRequest request = new TopicCreateRequest(name, null,
properties);
+ try (Response response = operations.createTopic("metalake", "kafka",
"default", request)) {
+ Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(),
response.getStatus());
+ ErrorResponse error = (ErrorResponse) response.getEntity();
+ Assertions.assertEquals(ErrorConstants.ILLEGAL_ARGUMENTS_CODE,
error.getCode());
+
Assertions.assertEquals(IllegalArgumentException.class.getSimpleName(),
error.getType());
+ Assertions.assertTrue(error.getMessage().contains(name));
+ Assertions.assertTrue(
+ error
+ .getMessage()
+ .contains(
+ "Kafka topic names in Gravitino must be 1 to 128
characters long and contain only"
+ + " ASCII alphanumerics, '_' and '-'."));
+ }
+ }
+ Mockito.verifyNoInteractions(dispatcher);
+ }
+}
diff --git
a/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java
b/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java
index f635ebb0bd..66e3c7dd6b 100644
---
a/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java
+++
b/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java
@@ -53,6 +53,8 @@ import java.io.IOException;
import java.time.Instant;
import java.util.Map;
import java.util.Set;
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.core.Response;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Config;
@@ -64,9 +66,15 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.SchemaChange;
+import org.apache.gravitino.catalog.CatalogManager;
+import org.apache.gravitino.catalog.TopicDispatcher;
+import org.apache.gravitino.catalog.TopicNormalizeDispatcher;
import org.apache.gravitino.catalog.kafka.embedded.KafkaClusterEmbedded;
import org.apache.gravitino.connector.HasPropertyMetadata;
import org.apache.gravitino.connector.PropertiesMetadata;
+import org.apache.gravitino.dto.requests.TopicCreateRequest;
+import org.apache.gravitino.dto.responses.ErrorConstants;
+import org.apache.gravitino.dto.responses.ErrorResponse;
import org.apache.gravitino.exceptions.ConnectionFailedException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NoSuchTopicException;
@@ -75,6 +83,7 @@ import org.apache.gravitino.messaging.Topic;
import org.apache.gravitino.messaging.TopicChange;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.server.web.rest.TopicOperations;
import org.apache.gravitino.storage.IdGenerator;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.helper.CatalogIds;
@@ -442,6 +451,51 @@ public class TestKafkaCatalogOperations extends
KafkaClusterEmbedded {
exception.getMessage());
}
+ @Test
+ void testCreateTopicNameCollisionReturnsBadRequest() throws
IllegalAccessException {
+ NameIdentifier existingTopic =
+ NameIdentifier.of(
+ METALAKE_NAME, CATALOG_NAME, DEFAULT_SCHEMA_NAME,
"test.topic.name.collision");
+ String collidingName = "test_topic_name_collision";
+ Map<String, String> properties = ImmutableMap.of(PARTITION_COUNT, "1",
REPLICATION_FACTOR, "1");
+
+ // Create the dotted name directly in Kafka, bypassing Gravitino's name
validation.
+ kafkaCatalogOperations.createTopic(existingTopic, null, null, properties);
+ try {
+ CatalogManager catalogManager = Mockito.mock(CatalogManager.class);
+ when(catalogManager.doWithCatalog(Mockito.any(), Mockito.any()))
+ .thenReturn(new KafkaCatalogCapability());
+ TopicDispatcher dispatcher = Mockito.mock(TopicDispatcher.class);
+ when(dispatcher.createTopic(Mockito.any(), Mockito.any(), Mockito.any(),
Mockito.any()))
+ .thenAnswer(
+ invocation ->
+ kafkaCatalogOperations.createTopic(
+ invocation.getArgument(0),
+ invocation.getArgument(1),
+ invocation.getArgument(2),
+ invocation.getArgument(3)));
+ TopicOperations operations =
+ new TopicOperations(new TopicNormalizeDispatcher(dispatcher,
catalogManager));
+ FieldUtils.writeField(
+ operations, "httpRequest", Mockito.mock(HttpServletRequest.class),
true);
+
+ TopicCreateRequest request = new TopicCreateRequest(collidingName, null,
properties);
+ try (Response response =
+ operations.createTopic(METALAKE_NAME, CATALOG_NAME,
DEFAULT_SCHEMA_NAME, request)) {
+ Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(),
response.getStatus());
+ ErrorResponse error = (ErrorResponse) response.getEntity();
+ Assertions.assertEquals(ErrorConstants.ILLEGAL_ARGUMENTS_CODE,
error.getCode());
+
Assertions.assertEquals(IllegalArgumentException.class.getSimpleName(),
error.getType());
+ Assertions.assertTrue(error.getMessage().contains("Invalid topic
name"));
+ Assertions.assertTrue(error.getMessage().contains(collidingName));
+ Assertions.assertTrue(error.getMessage().contains("collides with
existing topics"));
+
Assertions.assertTrue(error.getMessage().contains(existingTopic.name()));
+ }
+ } finally {
+ kafkaCatalogOperations.dropTopic(existingTopic);
+ }
+ }
+
@Test
void testCreateTopicWithInvalidPartitionCount() {
for (String partitionCount : new String[] {"0", "-2"}) {
diff --git a/docs/kafka-catalog.md b/docs/kafka-catalog.md
index 3f561eb28a..6abf4bf0a5 100644
--- a/docs/kafka-catalog.md
+++ b/docs/kafka-catalog.md
@@ -48,6 +48,20 @@ Refer to [Schema
operation](./manage-messaging-metadata-using-gravitino.md#schem
- The Kafka catalog supports creating, updating, deleting, and listing topics.
+### Topic Naming Rules
+
+When creating a topic through Gravitino, its name must contain 1 to 128
characters and use only
+ASCII letters, digits, underscores (`_`), and hyphens (`-`). A leading hyphen
is allowed. For example,
+`orders`, `order_events`, and `-orders` are valid names; `a/b`, `a=b`, and
`a.b` are not.
+
+Gravitino excludes dots for authorization compatibility and limits names to
128 characters to fit
+its metadata storage. These restrictions are stricter than Kafka's native
naming rules.
+
+Invalid names return HTTP `400 Bad Request` before reaching Kafka. A name that
passes these checks
+can still be rejected by Kafka, for example when `a_b` collides with an
existing `a.b` in
+ZooKeeper-mode Kafka. These broker-side name rejections also return HTTP `400
Bad Request`, with
+Kafka's rejection reason included in the error message.
+
### Topic Properties
| Property name | Description | Default
value |
Required |
diff --git a/docs/open-api/topics.yaml b/docs/open-api/topics.yaml
index 85737bdf92..3b721d9bef 100644
--- a/docs/open-api/topics.yaml
+++ b/docs/open-api/topics.yaml
@@ -54,6 +54,8 @@ paths:
responses:
"200":
$ref: "#/components/responses/TopicResponse"
+ "400":
+ $ref: "./openapi.yaml#/components/responses/BadRequestErrorResponse"
"409":
description: Conflict - The target topic already exists
content:
