Improve tags managment

michelin · Dec 4, 2023 · b71928b · b71928b
1 parent b599d02
commit b71928b
Show file tree

Hide file tree

Showing 2 changed files with 44 additions and 8 deletions.
diff --git a/src/main/java/com/michelin/ns4kafka/services/TopicService.java b/src/main/java/com/michelin/ns4kafka/services/TopicService.java
@@ -48,6 +48,8 @@ public class TopicService {
     @Inject
     SchemaRegistryClient schemaRegistryClient;
 
+    public static final String DYNAMIC_TAGS_CREATION = "dynamic.tags.creation";
+
     /**
      * Find all topics.
      *
@@ -351,18 +353,37 @@ public List<String> validateTags(Namespace namespace, Topic topic) {
         Set<String> tagNames = schemaRegistryClient.getTags(namespace.getMetadata().getCluster())
             .map(tags -> tags.stream().map(TagInfo::name).collect(Collectors.toSet())).block();
 
+        List<String> unavailableTagNames = topic.getSpec().getTags()
+                .stream()
+                .filter(tagName -> tagNames != null && !tagNames.contains(tagName))
+                .toList();
+
+        if(topicCluster.isPresent() &&
+                Boolean.parseBoolean(
+                topicCluster.get().getConfig().getProperty(
+                        DYNAMIC_TAGS_CREATION,
+                        "true"))) {
+
+            List<TagInfo> tagsToCreate = unavailableTagNames
+                    .stream()
+                    .map(TagInfo::new)
+                    .collect(Collectors.toList());
+
+            schemaRegistryClient.createTags(
+                    tagsToCreate,
+                    namespace.getMetadata().getCluster())
+                    .block();
+
+            return validationErrors;
+        }
+
         if (tagNames == null || tagNames.isEmpty()) {
             validationErrors.add(String.format(
                 "Invalid value %s for tags: No tags allowed.",
                 String.join(", ", topic.getSpec().getTags())));
             return validationErrors;
         }
 
-        List<String> unavailableTagNames = topic.getSpec().getTags()
-            .stream()
-            .filter(tagName -> !tagNames.contains(tagName))
-            .toList();
-
         if (!unavailableTagNames.isEmpty()) {
             validationErrors.add(String.format(
                 "Invalid value %s for tags: Available tags are %s.",

diff --git a/src/main/java/com/michelin/ns4kafka/services/clients/schema/SchemaRegistryClient.java b/src/main/java/com/michelin/ns4kafka/services/clients/schema/SchemaRegistryClient.java
@@ -186,9 +186,9 @@ public Mono<SchemaCompatibilityResponse> deleteCurrentCompatibilityBySubject(Str
     public Mono<List<TagInfo>> getTags(String kafkaCluster) {
         ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
         HttpRequest<?> request = HttpRequest
-            .GET(URI.create(StringUtils.prependUri(
-                config.getUrl(), "/catalog/v1/types/tagdefs")))
-            .basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
+                .GET(URI.create(StringUtils.prependUri(
+                        config.getUrl(), "/catalog/v1/types/tagdefs")))
+                .basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
         return Mono.from(httpClient.retrieve(request, Argument.listOf(TagInfo.class)));
     }
 
@@ -225,6 +225,21 @@ public Mono<List<TagTopicInfo>> addTags(String kafkaCluster, List<TagTopicInfo>
         return Mono.from(httpClient.retrieve(request, Argument.listOf(TagTopicInfo.class)));
     }
 
+    /**
+     * Create tags.
+     *
+     * @param tags The list of tags to create
+     * @param kafkaCluster The Kafka cluster
+     * @return Information about created tags
+     */
+    public Mono<List<TagInfo>> createTags(List<TagInfo> tags, String kafkaCluster) {
+        ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
+        HttpRequest<?> request = HttpRequest.POST(URI.create(StringUtils.prependUri(
+                        config.getUrl(), "/catalog/v1/types/tagdefs")), tags)
+                .basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
+        return Mono.from(httpClient.retrieve(request, Argument.listOf(TagInfo.class)));
+    }
+
     /**
      * Delete a tag to a topic.
      *