-
Notifications
You must be signed in to change notification settings - Fork 228
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(config): add table topic conversion type configuration (#2203)
* feat(config): add table topic conversion type configurations * feat(config): rename table topic type to schema type and update related configurations * feat(config): add table topic schema registry URL configuration and validation * test(config): add unit tests for ControllerConfigurationValidator table topic schema configuration * fix(tests): update exception type in ControllerConfigurationValidatorTableTest for schema validation * feat(config): polish code
- Loading branch information
Showing
9 changed files
with
152 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
51 changes: 51 additions & 0 deletions
51
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTableTest.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| /* | ||
| * Copyright 2024, AutoMQ HK Limited. | ||
| * | ||
| * The use of this file is governed by the Business Source License, | ||
| * as detailed in the file "/LICENSE.S3Stream" included in this repository. | ||
| * | ||
| * As of the Change Date specified in that file, in accordance with | ||
| * the Business Source License, use of this software will be governed | ||
| * by the Apache License, Version 2.0 | ||
| */ | ||
|
|
||
| import kafka.automq.AutoMQConfig | ||
| import kafka.server.{ControllerConfigurationValidator, KafkaConfig} | ||
| import kafka.utils.TestUtils | ||
| import org.apache.kafka.common.config.ConfigResource | ||
| import org.apache.kafka.common.config.ConfigResource.Type.TOPIC | ||
| import org.apache.kafka.common.config.TopicConfig.TABLE_TOPIC_SCHEMA_TYPE_CONFIG | ||
| import org.apache.kafka.common.errors.InvalidConfigurationException | ||
| import org.apache.kafka.server.record.TableTopicSchemaType | ||
| import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows} | ||
| import org.junit.jupiter.api.{Tag, Test} | ||
|
|
||
| import java.util | ||
|
|
||
| @Tag("S3Unit") | ||
| class ControllerConfigurationValidatorTableTest { | ||
| val config = new KafkaConfig(TestUtils.createDummyBrokerConfig()) | ||
| val validator = new ControllerConfigurationValidator(config) | ||
|
|
||
| @Test | ||
| def testInvalidTableTopicSchemaConfig(): Unit = { | ||
| val config = new util.TreeMap[String, String]() | ||
| config.put(TABLE_TOPIC_SCHEMA_TYPE_CONFIG, TableTopicSchemaType.SCHEMA.name) | ||
|
|
||
| // Test without schema registry URL configured | ||
| val exception = assertThrows(classOf[InvalidConfigurationException], () => { | ||
| validator.validate(new ConfigResource(TOPIC, "foo"), config) | ||
| }) | ||
| assertEquals("Table topic schema type is set to SCHEMA but schema registry URL is not configured", exception.getMessage) | ||
|
|
||
| // Test with schema registry URL configured | ||
| val brokerConfigWithSchemaRegistry = TestUtils.createDummyBrokerConfig() | ||
| brokerConfigWithSchemaRegistry.put(AutoMQConfig.TABLE_TOPIC_SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081") | ||
|
|
||
| val kafkaConfigWithSchemaRegistry = new KafkaConfig(brokerConfigWithSchemaRegistry) | ||
| val validatorWithSchemaRegistry = new ControllerConfigurationValidator(kafkaConfigWithSchemaRegistry) | ||
|
|
||
| // No exception should be thrown when schema registry URL is configured properly | ||
| validatorWithSchemaRegistry.validate(new ConfigResource(TOPIC, "foo"), config) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
41 changes: 41 additions & 0 deletions
41
server-common/src/main/java/org/apache/kafka/server/record/TableTopicSchemaType.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| /* | ||
| * Copyright 2024, AutoMQ HK Limited. | ||
| * | ||
| * The use of this file is governed by the Business Source License, | ||
| * as detailed in the file "/LICENSE.S3Stream" included in this repository. | ||
| * | ||
| * As of the Change Date specified in that file, in accordance with | ||
| * the Business Source License, use of this software will be governed | ||
| * by the Apache License, Version 2.0 | ||
| */ | ||
|
|
||
| package org.apache.kafka.server.record; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Locale; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import static java.util.Arrays.asList; | ||
|
|
||
| public enum TableTopicSchemaType { | ||
| SCHEMALESS("schemaless"), | ||
| SCHEMA("schema"); | ||
|
|
||
| public final String name; | ||
| private static final List<TableTopicSchemaType> VALUES = asList(values()); | ||
|
|
||
| TableTopicSchemaType(String name) { | ||
| this.name = name; | ||
| } | ||
|
|
||
| public static List<String> names() { | ||
| return VALUES.stream().map(v -> v.name).collect(Collectors.toList()); | ||
| } | ||
|
|
||
| public static TableTopicSchemaType forName(String n) { | ||
| String name = n.toLowerCase(Locale.ROOT); | ||
| return VALUES.stream().filter(v -> v.name.equals(name)).findFirst().orElseThrow(() -> | ||
| new IllegalArgumentException("Unknown table topic type name: " + name) | ||
| ); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters