Fix return type for Schema list API

src/main/java/com/michelin/ns4kafka/controller/SchemaController.java

@@ -8,7 +8,6 @@
 import com.michelin.ns4kafka.model.Namespace;
 import com.michelin.ns4kafka.model.schema.Schema;
 import com.michelin.ns4kafka.model.schema.SchemaCompatibilityState;
-import com.michelin.ns4kafka.model.schema.SchemaList;
 import com.michelin.ns4kafka.service.SchemaService;
 import com.michelin.ns4kafka.util.enumation.ApplyStatus;
 import com.michelin.ns4kafka.util.exception.ResourceValidationException;
@@ -52,8 +51,16 @@ public class SchemaController extends NamespacedResourceController {
      * @return A list of schemas
      */
     @Get
-    public Flux<SchemaList> list(String namespace, @QueryValue(defaultValue = "*") String name) {
-        return schemaService.findByWildcardName(getNamespace(namespace), name);
+    public Flux<Schema> list(String namespace, @QueryValue(defaultValue = "*") String name) {
+        Namespace ns = getNamespace(namespace);
+        return schemaService.findByWildcardName(ns, name)
+            .collectList()
+            .flatMapMany(schemas -> schemas.size() == 1
+                ? Flux.fromIterable(schemas
+                    .stream()
+                    .map(schema -> schemaService.getSubjectLatestVersion(ns, schema.getMetadata().getName()))
+                    .toList()).flatMap(schema -> schema)
+                : Flux.fromIterable(schemas));
     }
 
     /**

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

@@ -6,7 +6,6 @@
 import com.michelin.ns4kafka.model.MetadataResource;
 import io.micronaut.core.annotation.Introspected;
 import jakarta.validation.Valid;
-import jakarta.validation.constraints.NotNull;
 import java.util.List;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
@@ -24,7 +23,6 @@
 @EqualsAndHashCode(callSuper = true)
 public class Schema extends MetadataResource {
     @Valid
-    @NotNull
     private SchemaSpec spec;
 
     /**

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

@@ -8,7 +8,6 @@
 import com.michelin.ns4kafka.model.Metadata;
 import com.michelin.ns4kafka.model.Namespace;
 import com.michelin.ns4kafka.model.schema.Schema;
-import com.michelin.ns4kafka.model.schema.SchemaList;
 import com.michelin.ns4kafka.service.client.schema.SchemaRegistryClient;
 import com.michelin.ns4kafka.service.client.schema.entities.SchemaCompatibilityRequest;
 import com.michelin.ns4kafka.service.client.schema.entities.SchemaCompatibilityResponse;
@@ -48,7 +47,7 @@ public class SchemaService {
      * @param namespace The namespace
      * @return A list of schemas
      */
-    public Flux<SchemaList> findAllForNamespace(Namespace namespace) {
+    public Flux<Schema> findAllForNamespace(Namespace namespace) {
         List<AccessControlEntry> acls = aclService
             .findResourceOwnerGrantedToNamespace(namespace, AccessControlEntry.ResourceType.TOPIC);
         return schemaRegistryClient
@@ -57,7 +56,7 @@ public Flux<SchemaList> findAllForNamespace(Namespace namespace) {
                 String underlyingTopicName = subject.replaceAll("-(key|value)$", "");
                 return aclService.isResourceCoveredByAcls(acls, underlyingTopicName);
             })
-            .map(subject -> SchemaList.builder()
+            .map(subject -> Schema.builder()
                 .metadata(Metadata.builder()
                     .cluster(namespace.getMetadata().getCluster())
                     .namespace(namespace.getMetadata().getName())
@@ -73,11 +72,11 @@ public Flux<SchemaList> findAllForNamespace(Namespace namespace) {
      * @param name      The name filter
      * @return A list of schemas
      */
-    public Flux<SchemaList> findByWildcardName(Namespace namespace, String name) {
+    public Flux<Schema> findByWildcardName(Namespace namespace, String name) {
         List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
         return findAllForNamespace(namespace)
-            .filter(schemaList -> RegexUtils
-                .isResourceCoveredByRegex(schemaList.getMetadata().getName(), nameFilterPatterns));
+            .filter(schema -> RegexUtils
+                .isResourceCoveredByRegex(schema.getMetadata().getName(), nameFilterPatterns));
     }
 
     /**
@@ -107,41 +106,52 @@ public Flux<Schema> getAllSubjectVersions(Namespace namespace, String subject) {
             );
     }
 
+    /**
+     * Build the schema spec from the SchemaResponse.
+     *
+     * @param namespace         The namespace
+     * @param subjectOptional   The subject object from Http response
+     * @return A Subject
+     */
+    public Mono<Schema> buildSchemaSpec(Namespace namespace, SchemaResponse subjectOptional) {
+        return schemaRegistryClient
+            .getCurrentCompatibilityBySubject(namespace.getMetadata().getCluster(), subjectOptional.subject())
+            .map(Optional::of)
+            .defaultIfEmpty(Optional.empty())
+            .map(currentCompatibilityOptional -> {
+                Schema.Compatibility compatibility = currentCompatibilityOptional.isPresent()
+                    ? currentCompatibilityOptional.get().compatibilityLevel() : Schema.Compatibility.GLOBAL;
+
+                return Schema.builder()
+                    .metadata(Metadata.builder()
+                        .cluster(namespace.getMetadata().getCluster())
+                        .namespace(namespace.getMetadata().getName())
+                        .name(subjectOptional.subject())
+                        .build())
+                    .spec(Schema.SchemaSpec.builder()
+                        .id(subjectOptional.id())
+                        .version(subjectOptional.version())
+                        .compatibility(compatibility)
+                        .schema(subjectOptional.schema())
+                        .schemaType(subjectOptional.schemaType() == null ? Schema.SchemaType.AVRO :
+                            Schema.SchemaType.valueOf(subjectOptional.schemaType()))
+                        .build())
+                    .build();
+            });
+    }
+
     /**
      * Get a subject by its name and version.
      *
      * @param namespace The namespace
      * @param subject   The subject
      * @param version   The version
-     * @return A Subject
+     * @return A subject
      */
     public Mono<Schema> getSubjectByVersion(Namespace namespace, String subject, String version) {
         return schemaRegistryClient
             .getSubject(namespace.getMetadata().getCluster(), subject, version)
-            .flatMap(latestSubjectOptional -> schemaRegistryClient
-                .getCurrentCompatibilityBySubject(namespace.getMetadata().getCluster(), subject)
-                .map(Optional::of)
-                .defaultIfEmpty(Optional.empty())
-                .map(currentCompatibilityOptional -> {
-                    Schema.Compatibility compatibility = currentCompatibilityOptional.isPresent()
-                        ? currentCompatibilityOptional.get().compatibilityLevel() : Schema.Compatibility.GLOBAL;
-
-                    return Schema.builder()
-                        .metadata(Metadata.builder()
-                            .cluster(namespace.getMetadata().getCluster())
-                            .namespace(namespace.getMetadata().getName())
-                            .name(latestSubjectOptional.subject())
-                            .build())
-                        .spec(Schema.SchemaSpec.builder()
-                            .id(latestSubjectOptional.id())
-                            .version(latestSubjectOptional.version())
-                            .compatibility(compatibility)
-                            .schema(latestSubjectOptional.schema())
-                            .schemaType(latestSubjectOptional.schemaType() == null ? Schema.SchemaType.AVRO :
-                                Schema.SchemaType.valueOf(latestSubjectOptional.schemaType()))
-                            .build())
-                        .build();
-                }));
+            .flatMap(subjectOptional -> buildSchemaSpec(namespace, subjectOptional));
     }
 
     /**

src/main/java/com/michelin/ns4kafka/util/enumation/Kind.java

@@ -21,7 +21,6 @@ public enum Kind {
     ROLE_BINDING("RoleBinding"),
     SCHEMA("Schema"),
     SCHEMA_COMPATIBILITY_STATE("SchemaCompatibilityState"),
-    SCHEMA_LIST("SchemaList"),
     STATUS("Status"),
     TOPIC("Topic"),
     VAULT_RESPONSE("VaultResponse");