fix(schemaregistry): fix subject mode

Author: fhussonnois

providers/jikkou-provider-aiven/src/main/java/io/streamthoughts/jikkou/extension/aiven/api/AivenAsyncSchemaRegistryApi.java

@@ -184,7 +184,7 @@ public Mono<ModeObject> getGlobalMode() {
      * {@inheritDoc}
      **/
     @Override
-    public Mono<ModeObject> getSubjectMode(@NotNull String subject, boolean defaultToGlobal) {
+    public Mono<ModeObject> getSubjectMode(@NotNull String subject) {
         throw new UnsupportedOperationException("Aiven schema registry does not support subject mode");
     }
 

providers/jikkou-provider-schema-registry/src/main/java/io/streamthoughts/jikkou/schema/registry/api/AsyncSchemaRegistryApi.java

@@ -106,10 +106,9 @@ Mono<CompatibilityCheck> testCompatibility(@NotNull String subject,
      * Gets mode level for the specified subject.
      *
      * @param subject         the name of the subject.
-     * @param defaultToGlobal flag to default to global mode.
      * @return                the mode.
      */
-    Mono<ModeObject> getSubjectMode(@NotNull String subject, boolean defaultToGlobal);
+    Mono<ModeObject> getSubjectMode(@NotNull String subject);
 
     /**
      * Updates mode for the specified subject.

providers/jikkou-provider-schema-registry/src/main/java/io/streamthoughts/jikkou/schema/registry/api/DefaultAsyncSchemaRegistryApi.java

@@ -106,8 +106,8 @@ public Mono<CompatibilityObject> deleteSubjectCompatibilityLevel(@NotNull final
     }
 
     @Override
-    public Mono<ModeObject> getSubjectMode(@NotNull String subject, boolean defaultToGlobal) {
-        return Mono.fromCallable(() -> api.getMode(subject, defaultToGlobal));
+    public Mono<ModeObject> getSubjectMode(@NotNull String subject) {
+        return Mono.fromCallable(() -> api.getMode(subject));
     }
 
     @Override

providers/jikkou-provider-schema-registry/src/main/java/io/streamthoughts/jikkou/schema/registry/api/SchemaRegistryApi.java

@@ -249,8 +249,7 @@ CompatibilityLevelObject getConfigCompatibility(@PathParam("subject") String sub
     @GET
     @Path("mode/{subject}")
     @Produces("application/vnd.schemaregistry.v1+json")
-    ModeObject getMode(@PathParam("subject") String subject,
-                       @QueryParam("defaultToGlobal") @DefaultValue("false") boolean defaultToGlobal);
+    ModeObject getMode(@PathParam("subject") String subject);
 
     @PUT
     @Path("mode/{subject}")

providers/jikkou-provider-schema-registry/src/main/java/io/streamthoughts/jikkou/schema/registry/reconciler/SchemaRegistrySubjectCollector.java

@@ -7,6 +7,7 @@
 package io.streamthoughts.jikkou.schema.registry.reconciler;
 
 import com.google.common.base.Strings;
+import io.streamthoughts.jikkou.common.utils.Enums;
 import io.streamthoughts.jikkou.core.annotation.SupportedResource;
 import io.streamthoughts.jikkou.core.config.ConfigProperty;
 import io.streamthoughts.jikkou.core.config.Configuration;
@@ -124,16 +125,16 @@ private ResourceList<V1SchemaRegistrySubject> listAll(@NotNull Configuration con
                             .onErrorResume(SchemaRegistrySubjectCollector::emptyOn404);
                     // Get Schema Registry Subject Mode
                     Mono<String> modeMono =
-                        api.getGlobalMode()
+                        api.getSubjectMode(subjectSchemaVersion.subject())
                             .map(ModeObject::mode)
                             .onErrorResume(SchemaRegistrySubjectCollector::emptyOn404);
 
                     return Mono.zip(compatibilityMono.defaultIfEmpty(EMPTY_STRING), modeMono.defaultIfEmpty(EMPTY_STRING))
-                        .map(tuple -> {
-                            CompatibilityLevels compatibilityLevel = Strings.isNullOrEmpty(tuple.getT1()) ? null : CompatibilityLevels.valueOf(tuple.getT1());
-                            Modes mode = Strings.isNullOrEmpty(tuple.getT1()) ? null : Modes.valueOf(tuple.getT2());
-                            return schemaRegistrySubjectFactory.createSchemaRegistrySubject(subjectSchemaVersion, compatibilityLevel, mode);
-                        });
+                        .map(tuple -> schemaRegistrySubjectFactory.createSchemaRegistrySubject(
+                            subjectSchemaVersion,
+                            Enums.safeValueOf(CompatibilityLevels.class, tuple.getT1()),
+                            Enums.safeValueOf(Modes.class, tuple.getT2())
+                        ));
                 });
         try {
             return new V1SchemaRegistrySubjectList.Builder().withItems(flux.collectList().block()).build();