This is an automated email from the ASF dual-hosted git repository. github-actions[bot] pushed a commit to branch cherry-pick-948964ab-to-branch-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit d9a4c83c251dbb34e66f51dde96257df8a757aca Author: mchades <[email protected]> AuthorDate: Wed Nov 4 17:41:10 2026 +0800 [#14574] fix(kafka): validate topic names and map invalid topic errors (#14575) ### What changes were proposed in this pull request? - Add independent Kafka topic name validation: 1–128 ASCII letters, digits, underscores, or hyphens. Keep dots excluded for Gravitino authorization compatibility and the 128-character metadata storage limit. Other scopes retain their default rules. - Map Kafka's `InvalidTopicException` to an illegal argument error while preserving the broker's reason. - Document the naming rules, add the `createTopic` HTTP 400 response to OpenAPI, and cover validation and broker-side name collisions with regression tests. ### Why are the changes needed? Illegal characters such as `/` and `=`, and broker-side name conflicts, currently produce HTTP 500 instead of HTTP 400. The inherited default rule also unnecessarily rejects leading hyphens and names longer than 64 characters. Fixes #14574 ### Does this PR introduce _any_ user-facing change? Yes. Invalid Kafka topic names return HTTP 400, including names rejected by the broker. Topic names may start with `-` and contain up to 128 characters; dots remain unsupported for creation through Gravitino. ### How was this patch tested? - `./gradlew :catalogs:catalog-kafka:spotlessApply` - `./gradlew :catalogs:catalog-kafka:check :catalogs:catalog-kafka:javadoc :docs:build -PskipITs` — 23 Kafka tests passed, including an embedded Kafka/ZooKeeper name-collision test through the REST resource. Javadoc and all OpenAPI checks passed. - `git diff --check` - Docker integration tests were not run. # Conflicts: # catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java # catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java # docs/kafka-catalog.md --- catalogs/catalog-kafka/build.gradle.kts | 2 + .../catalog/kafka/KafkaCatalogCapability.java | 27 +++++ .../catalog/kafka/KafkaCatalogOperations.java | 10 ++ .../catalog/kafka/TestKafkaCatalogCapability.java | 128 +++++++++++++++++++++ .../catalog/kafka/TestKafkaCatalogOperations.java | 120 +++++++++++++++++++ docs/kafka-catalog.md | 24 ++++ docs/open-api/topics.yaml | 2 + 7 files changed, 313 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 85a5feaa29..6e3fcebc0b 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 @@ -87,6 +87,7 @@ import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.errors.InvalidConfigurationException; 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; @@ -292,6 +293,15 @@ public class KafkaCatalogOperations implements CatalogOperations, SupportsSchema if (e.getCause() instanceof TopicExistsException) { throw new TopicAlreadyExistsException(e, "Topic %s already exists", ident); +<<<<<<< HEAD +======= + } 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); + +>>>>>>> 948964abb ([#14574] fix(kafka): validate topic names and map invalid topic errors (#14575)) } 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 ff492a17ff..fc6864ad47 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,11 @@ import org.apache.gravitino.messaging.Topic; import org.apache.gravitino.messaging.TopicChange; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.CatalogEntity; +<<<<<<< HEAD +======= +import org.apache.gravitino.meta.SchemaVersion; +import org.apache.gravitino.server.web.rest.TopicOperations; +>>>>>>> 948964abb ([#14574] fix(kafka): validate topic names and map invalid topic errors (#14575)) import org.apache.gravitino.storage.IdGenerator; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.helper.CatalogIds; @@ -442,6 +455,113 @@ public class TestKafkaCatalogOperations extends KafkaClusterEmbedded { } @Test +<<<<<<< HEAD +======= + 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"}) { + NameIdentifier ident = + NameIdentifier.of( + METALAKE_NAME, + CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + "test_create_topic_invalid_partitions_" + partitionCount); + Map<String, String> properties = + ImmutableMap.of(PARTITION_COUNT, partitionCount, REPLICATION_FACTOR, "1"); + + IllegalArgumentException exception = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> kafkaCatalogOperations.createTopic(ident, null, null, properties)); + Assertions.assertTrue( + exception.getMessage().contains("Invalid partition count for topic " + ident), + exception.getMessage()); + Throwable cause = exception.getCause().getCause(); + Assertions.assertInstanceOf(InvalidPartitionsException.class, cause); + Assertions.assertTrue(exception.getMessage().contains(cause.getMessage())); + } + } + + @Test + void testCreateTopicWithDefaultPartitionCount() { + NameIdentifier defaultIdent = + NameIdentifier.of( + METALAKE_NAME, + CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + "test_create_topic_default_partitions"); + Topic defaultTopic = + kafkaCatalogOperations.createTopic( + defaultIdent, null, null, ImmutableMap.of(REPLICATION_FACTOR, "1")); + + NameIdentifier negativeOneIdent = + NameIdentifier.of( + METALAKE_NAME, + CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + "test_create_topic_negative_one_partitions"); + Topic negativeOneTopic = + kafkaCatalogOperations.createTopic( + negativeOneIdent, + null, + null, + ImmutableMap.of(PARTITION_COUNT, "-1", REPLICATION_FACTOR, "1")); + + String defaultPartitionCount = defaultTopic.properties().get(PARTITION_COUNT); + Assertions.assertTrue(Integer.parseInt(defaultPartitionCount) > 0); + Assertions.assertEquals( + defaultPartitionCount, negativeOneTopic.properties().get(PARTITION_COUNT)); + Assertions.assertEquals( + defaultPartitionCount, + kafkaCatalogOperations.loadTopic(negativeOneIdent).properties().get(PARTITION_COUNT)); + } + + @Test +>>>>>>> 948964abb ([#14574] fix(kafka): validate topic names and map invalid topic errors (#14575)) public void testLoadTopic() { Topic topic = kafkaCatalogOperations.loadTopic( diff --git a/docs/kafka-catalog.md b/docs/kafka-catalog.md index 3f561eb28a..da2a9f6db9 100644 --- a/docs/kafka-catalog.md +++ b/docs/kafka-catalog.md @@ -48,6 +48,30 @@ Refer to [Schema operation](./manage-messaging-metadata-using-gravitino.md#schem - The Kafka catalog supports creating, updating, deleting, and listing topics. +<<<<<<< HEAD +======= +### 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. + +::::caution Topic names containing dots +When authorization is enabled, topic names containing dots are unsupported, and one such topic can +cause the entire topic list request to fail. See +[Names containing dots](./security/access-control.md#names-containing-dots) for details and the +workaround. +:::: + +>>>>>>> 948964abb ([#14574] fix(kafka): validate topic names and map invalid topic errors (#14575)) ### 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:
