Add support for schema subject naming strategies #581

Athosone · 2025-06-19T14:24:38Z

Summary

This pull request implements support for configurable schema subject naming strategies in NS4Kafka, allowing namespaces to enforce different schema naming conventions beyond the default TopicNameStrategy.

Changes Made

Core Implementation

New enum SubjectNameStrategy: Defines the three main Confluent schema naming strategies:
- TOPIC_NAME (default): {topic}-key/{topic}-value format
- TOPIC_RECORD_NAME: {topic}-{recordName} format
- RECORD_NAME: {recordName} format
Enhanced validation framework: Added SchemaSubjectNameValidator with comprehensive validation logic for all naming strategies, including:
- AVRO schema record name extraction
- Pattern matching for each strategy type
- Support for qualified record names

Integration Points

Topic configuration: Added value.subject.name.strategy configuration support
Namespace validation: Extended namespace specs to define valid naming strategies per namespace
Schema service: Updated schema validation to use configured naming strategies instead of hardcoded TopicNameStrategy
Error handling: Enhanced error messages to show expected formats for configured strategies

Files Modified

SubjectNameStrategy.java (new): Core enum with strategy definitions and format helpers
SchemaSubjectNameValidator.java (new): Validation logic for all naming strategies
TopicValidator.java: Added naming strategy configuration validation
SchemaService.java: Integrated new validation logic
FormatErrorUtils.java: Enhanced error messages for naming strategy violations
Topic.java: Added subject name strategy configuration support
Comprehensive test coverage in SchemaSubjectNameValidatorTest.java

Benefits

Flexibility: Supports all Confluent schema naming strategies
Backward compatibility: Default behavior unchanged (TopicNameStrategy)
Namespace-level control: Administrators can configure allowed strategies per namespace
Better validation: Clear error messages showing expected formats
Future-ready: Extensible architecture for additional naming strategies

Testing

Added comprehensive unit tests covering all naming strategies
Validated AVRO record name extraction with various schema formats
Integration tests ensure backward compatibility
Edge cases handled (empty schemas, malformed content, etc.)

This implementation provides the foundation for flexible schema management while maintaining full backward compatibility with existing deployments.

Co-authors

Co-authored-by: monitorpattern [email protected]

loicgreffier

@Athosone @monitorpattern Here is my review.

I have mainly reviewed the schema validation for now (basically what's inside schemaService.validateSchema).

I've added "⚠️" in front of the points that need specific attention. The rest are minor code updates.

src/main/java/com/michelin/ns4kafka/service/SchemaService.java

src/main/java/com/michelin/ns4kafka/validation/TopicValidator.java

loicgreffier · 2025-07-09T19:19:29Z

src/main/java/com/michelin/ns4kafka/service/SchemaService.java

            List<String> validationErrors = new ArrayList<>();
+            List<SubjectNameStrategy> namingStrategies = getValidSubjectNameStrategies(namespace);
+            String subjectName = schema.getMetadata().getName();
+            boolean isValid = SchemaSubjectNameValidator.validateSubjectName(


Just pass the whole schema object here instead of these 3 parameters schema.getMetadata().getName(), schema.getSpec().getSchema(), schema.getSpec().getSchemaType(). It'll make it clearer.

Also, we use to define validation functions in services, so validateSubjectName can just be a function of the SchemaService.

loicgreffier · 2025-07-09T19:26:45Z

src/main/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidator.java

+    public static boolean validateSubjectNameWithStrategy(
+            String subjectName, SubjectNameStrategy strategy, String schemaContent, Schema.SchemaType schemaType) {
+        // https://github.com/confluentinc/schema-registry/blob/master/schema-serializer/src/main/java/io/confluent/kafka/serializers/subject
+        switch (strategy) {


I think this switch-case could be written using the return-switch pattern:

return switch (strategy) { case TOPIC_NAME -> { String topicName = extractTopicName(subjectName, strategy).orElse(""); yield subjectName.equals(topicName + "-key") || subjectName.equals(topicName + "-value"); } case TOPIC_RECORD_NAME -> { String topicName = extractTopicName(subjectName, strategy).orElse(""); Optional<String> recordName = extractRecordName(schemaContent, schemaType); yield recordName.isPresent() && subjectName.equals(topicName + "-" + recordName.get()); } case RECORD_NAME -> { Optional<String> recordName = extractRecordName(schemaContent, schemaType); yield recordName.isPresent() && subjectName.equals(recordName.get()); } };

Also, feel free to pass the whole schema in parameter to avoid having so many parameters.

src/main/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidator.java

loicgreffier · 2025-07-09T19:54:54Z

src/main/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidator.java

+        }
+
+        try {
+            switch (schemaType) {


The IDE is yelling that a single condition switch-case should be replaced with an if, so let's make it happy.

⚠️ I think the whole extractAvroRecordName function can be replaced with new AvroSchema(schemaContent).name() which return namespace + name for us 😅.

So overall, extracting the record name from the schema can just be:

if (schemaContent == null || schemaContent.trim().isEmpty()) { return Optional.empty(); } if (schemaType == Schema.SchemaType.AVRO) { return Optional.of(new AvroSchema(schemaContent).name()); } return Optional.empty();

Actually it throws an exception when the references aren't resolved which break existing tests and current behavior.
Also it would throw when parsing union of reference :(

Hence, we hesitate between setting a constant value for the record name in the case of union references that do not have namespace/name, and letting the namespace/name parsing. In either case, we need to use the json parser. What do you think?

This should not be a big issue. All functions are available in the SchemaService to build a new AvroSchema even with some references:

getSchemaReferences(schema, namespace) .map(schemaRefs -> new AvroSchema( schema.getSpec().getSchema(), getReferences(schema), schemaRefs, null ) .name())

Call the getSchemaReferences() function. The schema and namespace are needed

Build a new AvroSchema(...) with the references

As getSchemaReferences is asynchronous, extractRecordName should now return a Mono<String> as well as all calling methods

loicgreffier · 2025-07-09T19:57:16Z

src/main/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidator.java

+public final class SchemaSubjectNameValidator {
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    private SchemaSubjectNameValidator() {}


We use the @NoArgsConstructor(access = AccessLevel.PRIVATE) convention in this project instead

loicgreffier · 2025-07-09T20:26:06Z

src/main/java/com/michelin/ns4kafka/validation/TopicValidator.java


+    public List<SubjectNameStrategy> getValidSubjectNameStrategies() {
+        ResourceValidator.Validator namingStrategies =
+                getValidationConstraints().get(VALUE_SUBJECT_NAME_STRATEGY);


⚠️ Be careful here:

If the given subject name ends with -key we should verify it against the KEY_SUBJECT_NAME_STRATEGY.

If the given subject name is RecordName or TopicRecordName, we should verify it against both KEY_SUBJECT_NAME_STRATEGY or VALUE_SUBJECT_NAME_STRATEGY.

Currently, if a namespace has the following topic validation rules:

VALUE_SUBJECT_NAME_STRATEGY authorized for TopicName

KEY_SUBJECT_NAME_STRATEGY authorized for RecordName
And a schema following the RecordName strategy is being deployed, the user will be denied even though they're authorized to deploy that schema, right?

ThomasCAI-mlv

Here's my review, feel free to challenge :)

src/main/java/com/michelin/ns4kafka/model/schema/SubjectNameStrategy.java

src/main/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidator.java

ThomasCAI-mlv · 2025-07-10T14:24:44Z

src/main/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidator.java

+     * @param strategy The naming strategy
+     * @return The topic name if it can be determined
+     */
+    public static Optional<String> extractTopicName(String subjectName, SubjectNameStrategy strategy) {


With the feedback on validateSubjectNameWithStrategy(), this method would be used only once, you might consider putting its content in extractTopicName(String subjectName, List<SubjectNameStrategy> strategies)

src/main/java/com/michelin/ns4kafka/validation/TopicValidator.java

ThomasCAI-mlv · 2025-07-10T15:25:33Z

src/test/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidatorTest.java

+class SchemaSubjectNameValidatorTest {
+
+    @Test
+    void testValidateSubjectName_TopicNameStrategy_Valid() {


Can you follow the naming convention shouldActLikeThisWhenThatCondition() or shouldActLikeThisIfThatCondition() or shouldActLikeThisForThatConstraint() for test methods?

I suggest shouldValidateSubjectForTopicNameStrategy() and shouldNotValidateSubjectForTopicNameStrategy()

ThomasCAI-mlv · 2025-07-10T15:34:04Z

src/test/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidatorTest.java

+                subject, List.of(SubjectNameStrategy.RECORD_NAME), schemaContent, Schema.SchemaType.AVRO);
+        assertTrue(result);
+    }
+


Might be better to add an "OK" test with multiple strategies and a "KO" test with multiple strategies

ThomasCAI-mlv · 2025-07-10T15:36:28Z

src/test/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidatorTest.java

+        boolean result = SchemaSubjectNameValidator.validateSubjectName(
+                subject, List.of(SubjectNameStrategy.TOPIC_RECORD_NAME), schemaContent, Schema.SchemaType.AVRO);
+        assertTrue(result);
+    }


For TopicRecordName, might be good to test:

KO when the subject name has no -

OK when the subject name has multiple -
In addition to the simple cases:

OK when the subject name has one -

KO when the subject name does not match the schema namespace / name

ThomasCAI-mlv · 2025-07-10T16:05:48Z

src/main/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidator.java

@@ -0,0 +1,160 @@
+/*


I'm not sure the content of this file should be in a "Validator" class, it makes more sense in the "SchemaService". What do you think?

missing doc: michelin#581 (comment) useless constraints: michelin#581 (comment)

loicgreffier · 2025-08-20T08:45:46Z

src/main/java/com/michelin/ns4kafka/model/schema/SubjectNameStrategy.java

+    TOPIC_RECORD_NAME("io.confluent.kafka.serializers.subject.TopicRecordNameStrategy"),
+    RECORD_NAME("io.confluent.kafka.serializers.subject.RecordNameStrategy");
+
+    private final String STRATEGY_PREFIX = "io.confluent.kafka.serializers.subject.";


static is missing: private static final

loicgreffier · 2025-08-20T08:46:49Z

src/main/java/com/michelin/ns4kafka/model/schema/SubjectNameStrategy.java

+     * @return The format for subject value (i.e. the SchemaResource metadata name) according to subject name strategy
+     */
+    public String toExpectedFormat() {
+        switch (this) {


You can use return-switch here like:

return switch (this) { case TOPIC_NAME -> "{topic}-{key|value}"; case TOPIC_RECORD_NAME -> "{topic}-{recordName}"; case RECORD_NAME -> "{recordName}"; };

loicgreffier · 2025-08-20T08:56:30Z

src/main/java/com/michelin/ns4kafka/validation/ValidSubjectNameStrategies.java

+ */
+@AllArgsConstructor
+public class ValidSubjectNameStrategies {
+    public List<SubjectNameStrategy> validValueStrategies;


Make both attributes private and define a @Getter on the class

loicgreffier · 2025-08-20T09:02:42Z

src/main/java/com/michelin/ns4kafka/service/SchemaService.java

+     */
+    public static boolean validateSubjectName(ValidSubjectNameStrategies validStrategies, Schema schema) {
+        if (schema.getMetadata().getName().endsWith("-key")) {
+            return validStrategies.validKeyStrategies.stream()


Use getter here, based on my comment above: validStrategies.getValidKeyStrategies()

loicgreffier · 2025-08-20T09:03:49Z

src/main/java/com/michelin/ns4kafka/service/SchemaService.java

+        if (subjectName == null || subjectName.trim().isEmpty()) {
+            return false;
+        }
+        switch (strategy) {


return-switch here like:

return switch (strategy) { case TOPIC_NAME -> subjectName.endsWith("-key") || subjectName.endsWith("-value"); case TOPIC_RECORD_NAME -> { Optional<String> recordName = extractRecordName(schemaContent, schemaType); yield recordName.isPresent() && (recordName.get() == UNION_AVRO_RECORD_CLASS_NAME || subjectName.endsWith("-" + recordName.get())); } case RECORD_NAME -> { Optional<String> recordNameOnly = extractRecordName(schemaContent, schemaType); yield recordNameOnly.isPresent() && subjectName.equals(recordNameOnly.get()); } };

loicgreffier · 2025-08-20T14:35:00Z

src/main/java/com/michelin/ns4kafka/validation/SchemaSubjectNameValidator.java

+        }
+
+        try {
+            switch (schemaType) {


This should not be a big issue. All functions are available in the SchemaService to build a new AvroSchema even with some references:

getSchemaReferences(schema, namespace) .map(schemaRefs -> new AvroSchema( schema.getSpec().getSchema(), getReferences(schema), schemaRefs, null ) .name())

Call the getSchemaReferences() function. The schema and namespace are needed

Build a new AvroSchema(...) with the references

As getSchemaReferences is asynchronous, extractRecordName should now return a Mono<String> as well as all calling methods

… null Strategy is set as optional, so it can be null

Co-authored-by: monitorpattern <[email protected]>

missing doc: michelin#581 (comment) useless constraints: michelin#581 (comment)

monitorpattern force-pushed the feature/schema-naming-strategies branch from 36f2d5d to 9db47c1 Compare June 19, 2025 15:03

Athosone force-pushed the feature/schema-naming-strategies branch 2 times, most recently from 0d70c20 to 5272cda Compare June 19, 2025 15:06

monitorpattern force-pushed the feature/schema-naming-strategies branch from 910f0a5 to 990a391 Compare June 19, 2025 19:44

Athosone closed this Jun 19, 2025

Athosone reopened this Jun 19, 2025

monitorpattern force-pushed the feature/schema-naming-strategies branch from cfc225d to 5edb5ba Compare June 30, 2025 11:09

loicgreffier changed the title ~~feat: Add support for schema subject naming strategies~~ Add support for schema subject naming strategies Jul 8, 2025

loicgreffier requested review from ThomasCAI-mlv and loicgreffier July 8, 2025 16:26

loicgreffier added the feature This issue or pull request contains a new feature label Jul 8, 2025

loicgreffier reviewed Jul 9, 2025

View reviewed changes

ThomasCAI-mlv reviewed Jul 10, 2025

View reviewed changes

monitorpattern added a commit to Athosone/ns4kafka that referenced this pull request Jul 18, 2025

Fix comments

2b32bcf

missing doc: michelin#581 (comment) useless constraints: michelin#581 (comment)

monitorpattern added a commit to Athosone/ns4kafka that referenced this pull request Jul 24, 2025

Move SchemasSubjectNameValidator into SchemaService

efec984

missing doc: michelin#581 (comment) useless constraints: michelin#581 (comment)

loicgreffier reviewed Aug 20, 2025

View reviewed changes

Athosone and others added 14 commits October 24, 2025 23:48

wip: validate subject based on the namespace configuration

35e62a1

Polish code and rename classes (Schema>Subject)

2edf399

Add constants + spotlessApply

b6a0584

Remove methods that hamper introspection + do not test if strategy is…

7f7d9b6

… null Strategy is set as optional, so it can be null

Fix type of strategy config (ValidList -> ValidString)

1133944

Add comments and TODO

2d21eff

wip: fixing todos

357b1ea

Fix SchemasServiceTest with new subject strategy handling

f0a7efb

Fix enum error for SubjectNameStrategy

129403a

change name of field 'name' in Enum (Enum has a default method name).

464b110

Simplify error message when subject name strategy is not respected

eceed51

Co-authored-by: monitorpattern <[email protected]>

Polish code + add test for topic record name strategy

304fc9a

Update a method comment and apply spotlessApply

736dea3

Fix tests by adding topicValidator to all namespace specs

bf6224f

monitorpattern and others added 26 commits October 24, 2025 23:48

Polish the code + spotlessApply

eb63bf5

Compare subject name including namespace

4448344

Polish the Readme

2ad6768

Add an integration test for subject name strategy other than topic name

9134fbe

Apply spotlessApply task

cfe90bc

Fix comments

a5c061c

missing doc: michelin#581 (comment) useless constraints: michelin#581 (comment)

Fix topicvalidator comments

b0b2226

Move SchemasSubjectNameValidator into SchemaService

b065b7f

missing doc: michelin#581 (comment) useless constraints: michelin#581 (comment)

Apply spotlessApply

77fd29e

Refactor signatures of methods that involve schemas

456abac

fix: AllArgsConstructor

9cfcb13

fix: replace substring by replace :p

97f4463

fix: add description

e179aa8

fix: refactor validate subject name strategy

c628791

fix: remove extractRecordName

630d8cf

fix: handle key subject name strategies

39fbff6

fix: fix javadoc

aae2e6a

Rename tests wIcône xlsx MBO Team Mood.xlsx.r.t naming conventions

70bd39a

Rename tests wrt naming conventions WIP

78885fd

fix: extract name, also renamed tests

0fc294b

fix: test from comment review

94134aa

fix: test with wrong input data

954df59

fix: rolled back extract topic record name to handle union of reference

b5f75a2

Some updates

6b27a5d

Fixes before unit tests

4cc6abc

update

afbadec

loicgreffier force-pushed the feature/schema-naming-strategies branch from 460d310 to afbadec Compare October 24, 2025 21:48

loicgreffier added 2 commits October 25, 2025 00:59

update

8083a44

Run test on CCloud cluster

c2d4a72

loicgreffier requested a review from ThomasCAI-mlv October 25, 2025 00:01

Add support for schema subject naming strategies #581

Are you sure you want to change the base?

Add support for schema subject naming strategies #581

Uh oh!

Conversation

Athosone commented Jun 19, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Changes Made

Core Implementation

Integration Points

Files Modified

Benefits

Testing

Co-authors

Uh oh!

loicgreffier left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ThomasCAI-mlv left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Athosone commented Jun 19, 2025 •

edited

Loading