Add support for schema subject naming strategies

.docker/resources/admin/namespace.yml

@@ -8,6 +8,10 @@ metadata:
 spec:
   kafkaUser: "user1"
   transactionsEnabled: true
+  subjectNameStrategies:
+    - "TopicNameStrategy"
+    - "RecordNameStrategy"
+    - "TopicRecordNameStrategy"
   connectClusters:
   - "local"
   topicValidator:

src/main/java/com/michelin/ns4kafka/model/AccessControlEntry.java

@@ -80,12 +80,12 @@ public enum Permission {
     public static class AccessControlEntrySpec {
         @NotNull protected AccessControlEntry.ResourceType resourceType;
 
-        @NotNull @NotBlank protected String resource;
+        @NotBlank protected String resource;
 
         @NotNull protected ResourcePatternType resourcePatternType;
 
         @NotNull protected Permission permission;
 
-        @NotBlank @NotNull protected String grantedTo;
+        @NotBlank protected String grantedTo;
     }
 }

src/main/java/com/michelin/ns4kafka/model/Namespace.java

@@ -20,6 +20,7 @@
 
 import static com.michelin.ns4kafka.util.enumation.Kind.NAMESPACE;
 
+import com.michelin.ns4kafka.model.schema.SubjectNameStrategy;
 import com.michelin.ns4kafka.validation.ConnectValidator;
 import com.michelin.ns4kafka.validation.TopicValidator;
 import io.micronaut.core.annotation.Introspected;
@@ -66,6 +67,9 @@ public static class NamespaceSpec {
         @Builder.Default
         private boolean transactionsEnabled = true;
 
+        @Builder.Default
+        private List<SubjectNameStrategy> subjectNameStrategies = SubjectNameStrategy.defaultStrategy();
+
         @Builder.Default
         private List<String> connectClusters = List.of();
 

src/main/java/com/michelin/ns4kafka/model/RoleBinding.java

@@ -98,6 +98,6 @@ public static class Role {
     public static class Subject {
         @NotNull private SubjectType subjectType;
 
-        @NotNull @NotBlank private String subjectName;
+        @NotBlank private String subjectName;
     }
 }

src/main/java/com/michelin/ns4kafka/model/consumer/group/ConsumerGroupResetOffsets.java

@@ -73,7 +73,7 @@ public enum ResetOffsetsMethod {
     @NoArgsConstructor
     @AllArgsConstructor
     public static class ConsumerGroupResetOffsetsSpec {
-        @NotNull @NotBlank private String topic;
+        @NotBlank private String topic;
 
         @NotNull private ResetOffsetsMethod method;
 

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

@@ -24,6 +24,8 @@
 import com.michelin.ns4kafka.model.MetadataResource;
 import io.micronaut.core.annotation.Introspected;
 import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
 import java.util.List;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
@@ -88,7 +90,7 @@ public static class SchemaSpec {
         @Builder.Default
         private Compatibility compatibility = Compatibility.GLOBAL;
 
-        private List<Reference> references;
+        @Valid private List<Reference> references;
 
         /** Schema reference. */
         @Getter
@@ -98,9 +100,11 @@ public static class SchemaSpec {
         @NoArgsConstructor
         @AllArgsConstructor
         public static class Reference {
-            private String name;
-            private String subject;
-            private Integer version;
+            @NotBlank private String name;
+
+            @NotBlank private String subject;
+
+            @NotNull private Integer version;
         }
     }
 }

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

@@ -0,0 +1,80 @@
+/*
+ * 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 com.michelin.ns4kafka.model.schema;
+
+import com.fasterxml.jackson.annotation.JsonValue;
+import java.util.Arrays;
+import java.util.List;
+import lombok.AllArgsConstructor;
+
+/** Schema subject naming strategies supported by Schema Registry. */
+@AllArgsConstructor
+public enum SubjectNameStrategy {
+    TOPIC_NAME("TopicNameStrategy"),
+    TOPIC_RECORD_NAME("TopicRecordNameStrategy"),
+    RECORD_NAME("RecordNameStrategy");
+
+    private final String value;
+
+    /**
+     * Convert the SubjectNameStrategy to its string representation.
+     *
+     * @return The string representation of the SubjectNameStrategy
+     */
+    @JsonValue
+    @Override
+    public String toString() {
+        return value;
+    }
+
+    /**
+     * Get the default SubjectNameStrategy.
+     *
+     * @return The default SubjectNameStrategy
+     */
+    public static List<SubjectNameStrategy> defaultStrategy() {
+        return List.of(TOPIC_NAME);
+    }
+
+    /**
+     * Get SubjectNameStrategy from its string representation.
+     *
+     * @param stringValue The string representation of the SubjectNameStrategy
+     * @return The SubjectNameStrategy
+     */
+    public static SubjectNameStrategy from(final String stringValue) {
+        return Arrays.stream(values())
+                .filter(s -> s.value.equals(stringValue))
+                .findFirst()
+                .orElseThrow(() -> new IllegalArgumentException("Unknown strategy: " + stringValue));
+    }
+
+    /**
+     * Get the expected format for the subject name based on the strategy.
+     *
+     * @return The expected format for the subject name
+     */
+    public String toExpectedFormat() {
+        return switch (this) {
+            case TOPIC_NAME -> "{topic}-{key|value}";
+            case TOPIC_RECORD_NAME -> "{topic}-{recordName}";
+            case RECORD_NAME -> "{recordName}";
+        };
+    }
+}