Improve catalog information synchronisation with GraphQL

README.md

@@ -286,11 +286,14 @@ You can configure the page size of the response of the API using the following p
 ns4kafka:
   confluent-cloud:
     stream-catalog:
+      sync-catalog: true
       page-size: 500
 ```
 
 The max page size is at 500 as described in the [Confluent Cloud documentation](https://docs.confluent.io/cloud/current/stream-governance/stream-catalog-rest-apis.html#list-all-topics).
 
+Note that this synchronization is disabled by default.
+
 ### Managed Kafka Clusters
 
 Managed clusters are the clusters where Ns4Kafka namespaces are deployed, and Kafka resources are managed.

build.gradle

@@ -24,33 +24,32 @@ dependencies {
     annotationProcessor("io.micronaut.security:micronaut-security-annotations")
     annotationProcessor("io.micronaut.validation:micronaut-validation-processor")
 
+    implementation('io.confluent:kafka-schema-registry-client:7.8.0')
+    implementation("io.micronaut.kafka:micronaut-kafka")
     implementation("io.micronaut:micronaut-http-client")
     implementation("io.micronaut:micronaut-jackson-databind")
-    implementation("io.micronaut.kafka:micronaut-kafka")
     implementation("io.micronaut:micronaut-management")
-    implementation("io.micronaut.validation:micronaut-validation")
+    implementation("io.micronaut.openapi:micronaut-openapi")
     implementation("io.micronaut.reactor:micronaut-reactor")
     implementation("io.micronaut.security:micronaut-security")
     implementation("io.micronaut.security:micronaut-security-jwt")
     implementation("io.micronaut.security:micronaut-security-ldap")
-    implementation("io.micronaut.openapi:micronaut-openapi")
+    implementation("io.micronaut.validation:micronaut-validation")
     implementation("io.swagger.core.v3:swagger-annotations")
     implementation("jakarta.annotation:jakarta.annotation-api")
     implementation("jakarta.validation:jakarta.validation-api")
-    implementation('io.confluent:kafka-schema-registry-client:7.8.0')
-
 
-    compileOnly("org.projectlombok:lombok")
     compileOnly("com.google.code.findbugs:jsr305") // https://github.com/micronaut-projects/micronaut-core/pull/5691
+    compileOnly("org.projectlombok:lombok")
 
     runtimeOnly("ch.qos.logback:logback-classic")
 
+    testImplementation("org.junit.jupiter:junit-jupiter-params:5.11.4")
     testImplementation("org.mockito:mockito-core")
+    testImplementation("org.mockito:mockito-junit-jupiter:5.15.2")
     testImplementation("org.testcontainers:junit-jupiter:1.20.4")
-    testImplementation("org.testcontainers:testcontainers:1.20.4")
     testImplementation("org.testcontainers:kafka:1.20.4")
-    testImplementation("org.mockito:mockito-junit-jupiter:5.15.2")
-    testImplementation("org.junit.jupiter:junit-jupiter-params:5.11.4")
+    testImplementation("org.testcontainers:testcontainers:1.20.4")
     testImplementation("io.projectreactor:reactor-test")
 
     testAnnotationProcessor("org.projectlombok:lombok")

src/main/java/com/michelin/ns4kafka/property/ConfluentCloudProperties.java

@@ -30,7 +30,7 @@
 @Setter
 @ConfigurationProperties("ns4kafka.confluent-cloud")
 public class ConfluentCloudProperties {
-    private StreamCatalogProperties streamCatalog;
+    private StreamCatalogProperties streamCatalog = new StreamCatalogProperties();
 
     /**
      * Stream Catalog properties.
@@ -40,5 +40,6 @@ public class ConfluentCloudProperties {
     @ConfigurationProperties("stream-catalog")
     public static class StreamCatalogProperties {
         int pageSize = 500;
+        boolean syncCatalog;
     }
 }

src/main/java/com/michelin/ns4kafka/service/client/schema/SchemaRegistryClient.java

@@ -20,6 +20,7 @@
 package com.michelin.ns4kafka.service.client.schema;
 
 import com.michelin.ns4kafka.property.ManagedClusterProperties;
+import com.michelin.ns4kafka.service.client.schema.entities.GraphQueryResponse;
 import com.michelin.ns4kafka.service.client.schema.entities.SchemaCompatibilityCheckResponse;
 import com.michelin.ns4kafka.service.client.schema.entities.SchemaCompatibilityRequest;
 import com.michelin.ns4kafka.service.client.schema.entities.SchemaCompatibilityResponse;
@@ -47,6 +48,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;
 import reactor.core.publisher.Flux;
@@ -283,21 +285,20 @@ public Mono<SchemaCompatibilityResponse> deleteCurrentCompatibilityBySubject(Str
     }
 
     /**
-     * Add a tag to a topic.
+     * List tags.
      *
      * @param kafkaCluster The Kafka cluster
-     * @param tagSpecs     Tags to add
-     * @return Information about added tags
+     * @return List of existing tags
      */
-    public Mono<List<TagTopicInfo>> associateTags(String kafkaCluster, List<TagTopicInfo> tagSpecs) {
+    public Mono<List<TagInfo>> listTags(String kafkaCluster) {
         ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
 
-        HttpRequest<?> request = HttpRequest.POST(
+        HttpRequest<?> request = HttpRequest.GET(
                 URI.create(StringUtils.prependUri(config.getUrl(),
-                    "/catalog/v1/entity/tags")), tagSpecs)
+                    "/catalog/v1/types/tagdefs")))
             .basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
 
-        return Mono.from(httpClient.retrieve(request, Argument.listOf(TagTopicInfo.class)));
+        return Mono.from(httpClient.retrieve(request, Argument.listOf(TagInfo.class)));
     }
 
     /**
@@ -319,7 +320,25 @@ public Mono<List<TagInfo>> createTags(String kafkaCluster, List<TagInfo> tags) {
     }
 
     /**
-     * Delete a tag to a topic.
+     * Add a tag to a topic.
+     *
+     * @param kafkaCluster The Kafka cluster
+     * @param tagSpecs     Tags to add
+     * @return Information about added tags
+     */
+    public Mono<List<TagTopicInfo>> associateTags(String kafkaCluster, List<TagTopicInfo> tagSpecs) {
+        ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
+
+        HttpRequest<?> request = HttpRequest.POST(
+                URI.create(StringUtils.prependUri(config.getUrl(),
+                    "/catalog/v1/entity/tags")), tagSpecs)
+            .basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
+
+        return Mono.from(httpClient.retrieve(request, Argument.listOf(TagTopicInfo.class)));
+    }
+
+    /**
+     * Delete a tag from a topic.
      *
      * @param kafkaCluster The Kafka cluster
      * @param entityName   The topic's name
@@ -338,12 +357,12 @@ public Mono<HttpResponse<Void>> dissociateTag(String kafkaCluster, String entity
     }
 
     /**
-     * List topics with catalog info, including tag & description.
+     * List topics with catalog info including tags & description, using Stream Catalog API.
      *
      * @param kafkaCluster The Kafka cluster
      * @return A list of description
      */
-    public Mono<TopicListResponse> getTopicWithCatalogInfo(String kafkaCluster, int limit, int offset) {
+    public Mono<TopicListResponse> getTopicsWithStreamCatalog(String kafkaCluster, int limit, int offset) {
         ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
 
         HttpRequest<?> request = HttpRequest.GET(
@@ -354,6 +373,48 @@ public Mono<TopicListResponse> getTopicWithCatalogInfo(String kafkaCluster, int
         return Mono.from(httpClient.retrieve(request, TopicListResponse.class));
     }
 
+    /**
+     * Query Stream Catalog information, using GraphQL.
+     *
+     * @param kafkaCluster The Kafka cluster
+     * @param query        The GraphQL query
+     * @return The GraphQL response containing kafka_topic
+     */
+    public Mono<GraphQueryResponse> getTopicsWithGraphQl(String kafkaCluster, String query) {
+        ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
+
+        HttpRequest<?> request = HttpRequest.POST(
+                URI.create(StringUtils.prependUri(config.getUrl(),
+                    "/catalog/graphql")),
+                Map.of("query", query))
+            .basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
+
+        return Mono.from(httpClient.retrieve(request, GraphQueryResponse.class));
+    }
+
+    /**
+     * List topics with tags, using GraphQL.
+     *
+     * @param kafkaCluster The Kafka cluster
+     * @return The GraphQL query response containing list of topic names and tags
+     */
+    public Mono<GraphQueryResponse> getTopicsWithTagsWithGraphQl(String kafkaCluster, List<String> tagsNames) {
+        String query = "query { kafka_topic(tags: [" + String.join(",", tagsNames) + "]) { nameLower tags } }";
+
+        return getTopicsWithGraphQl(kafkaCluster, query);
+    }
+
+    /**
+     * List topics with description, using GraphQL.
+     *
+     * @param kafkaCluster The Kafka cluster
+     * @return The GraphQL query response containing list of topic names and descriptions
+     */
+    public Mono<GraphQueryResponse> getTopicsWithDescriptionWithGraphQl(String kafkaCluster) {
+        String query = "query { kafka_topic(where: {description: {_gte: null}}) { nameLower description } }";
+        return getTopicsWithGraphQl(kafkaCluster, query);
+    }
+
     /**
      * Update a topic description.
      *

src/main/java/com/michelin/ns4kafka/service/client/schema/entities/GraphQueryData.java

@@ -0,0 +1,33 @@
+/*
+ * 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.service.client.schema.entities;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+import lombok.Builder;
+
+/**
+ * GraphQL query data.
+ *
+ * @param kafkaTopic the list of queried kafka topics
+ */
+@Builder
+public record GraphQueryData(@JsonProperty("kafka_topic") List<GraphQueryTopic> kafkaTopic) {
+}

src/main/java/com/michelin/ns4kafka/service/client/schema/entities/GraphQueryResponse.java

@@ -0,0 +1,31 @@
+/*
+ * 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.service.client.schema.entities;
+
+import lombok.Builder;
+
+/**
+ * GraphQL query response.
+ *
+ * @param data the response data
+ */
+@Builder
+public record GraphQueryResponse(GraphQueryData data) {
+}

src/main/java/com/michelin/ns4kafka/service/client/schema/entities/GraphQueryTopic.java

@@ -0,0 +1,35 @@
+/*
+ * 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.service.client.schema.entities;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+import lombok.Builder;
+
+/**
+ * GraphQL query kafka topic.
+ *
+ * @param name        the topic name
+ * @param description the topic description
+ * @param tags        the topic tags
+ */
+@Builder
+public record GraphQueryTopic(@JsonProperty("nameLower") String name, String description, List<String> tags) {
+}