Add new schema compatibilityConfigGCP

Author: magnusdriver

api/src/main/java/io/kafbat/ui/service/SchemaRegistryService.java

@@ -9,11 +9,12 @@
 import io.kafbat.ui.sr.api.KafkaSrClientApi;
 import io.kafbat.ui.sr.model.Compatibility;
 import io.kafbat.ui.sr.model.CompatibilityCheckResponse;
-import io.kafbat.ui.sr.model.CompatibilityConfig;
 import io.kafbat.ui.sr.model.CompatibilityLevelChange;
+import io.kafbat.ui.sr.model.GetGlobalCompatibilityLevel200Response;
 import io.kafbat.ui.sr.model.NewSubject;
 import io.kafbat.ui.sr.model.SchemaSubject;
 import io.kafbat.ui.util.ReactiveFailover;
+import io.kafbat.ui.util.gcp.GcpBearerAuthFilter;
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
@@ -35,9 +36,6 @@ public class SchemaRegistryService {
 
   private static final String LATEST = "latest";
 
-  private static final String GCP_BEARER_AUTH_CUSTOM_PROVIDER_CLASS =
-      "com.google.cloud.hosted.kafka.auth.GcpBearerAuthCredentialProvider";
-
   @AllArgsConstructor
   public static class SubjectWithCompatibilityLevel {
     @Delegate
@@ -152,14 +150,16 @@ public Mono<Compatibility> getSchemaCompatibilityLevel(KafkaCluster cluster,
                                                          String schemaName) {
     return api(cluster)
         .mono(c -> c.getSubjectCompatibilityLevel(schemaName, true))
-        .map(compatibilityConfig -> selectCompatibilityFormat(cluster, compatibilityConfig))
+        .map(compatibilityResponse ->
+            normalizeCompatibilityResponse(cluster, compatibilityResponse))
         .onErrorResume(error -> Mono.empty());
   }
 
   public Mono<Compatibility> getGlobalSchemaCompatibilityLevel(KafkaCluster cluster) {
     return api(cluster)
         .mono(KafkaSrClientApi::getGlobalCompatibilityLevel)
-        .map(compatibilityConfig -> selectCompatibilityFormat(cluster, compatibilityConfig));
+        .map(compatibilityResponse ->
+            normalizeCompatibilityResponse(cluster, compatibilityResponse));
   }
 
   private Mono<Compatibility> getSchemaCompatibilityInfoOrGlobal(KafkaCluster cluster,
@@ -174,13 +174,14 @@ public Mono<CompatibilityCheckResponse> checksSchemaCompatibility(KafkaCluster c
     return api(cluster).mono(c -> c.checkSchemaCompatibility(schemaName, LATEST, true, newSchemaSubject));
   }
 
-  private Compatibility selectCompatibilityFormat(KafkaCluster cluster, CompatibilityConfig compatibilityConfig) {
+  private Compatibility normalizeCompatibilityResponse(KafkaCluster cluster,
+                                                       GetGlobalCompatibilityLevel200Response compatibilityResponse) {
     if (cluster.getOriginalProperties().getSchemaRegistryAuth() != null
         && Objects.equals(cluster.getOriginalProperties().getSchemaRegistryAuth().getBearerAuthCustomProviderClass(),
-        GCP_BEARER_AUTH_CUSTOM_PROVIDER_CLASS)) {
-      return compatibilityConfig.getCompatibility();
+        GcpBearerAuthFilter.getGcpBearerAuthCustomProviderClass())) {
+      return compatibilityResponse.getCompatibility();
     } else {
-      return compatibilityConfig.getCompatibilityLevel();
+      return compatibilityResponse.getCompatibilityLevel();
     }
   }
 }

contract/src/main/resources/swagger/kafka-sr-api.yaml

@@ -176,7 +176,9 @@ paths:
                     content:
                         application/json:
                             schema:
-                                $ref: '#/components/schemas/CompatibilityConfig'
+                                oneOf:
+                                    - $ref: '#/components/schemas/CompatibilityConfig'
+                                    - $ref: '#/components/schemas/CompatibilityConfigGcp'
                 404:
                     description: Not found
         put:
@@ -220,7 +222,9 @@ paths:
                     content:
                         application/json:
                             schema:
-                                $ref: '#/components/schemas/CompatibilityConfig'
+                                oneOf:
+                                    - $ref: '#/components/schemas/CompatibilityConfig'
+                                    - $ref: '#/components/schemas/CompatibilityConfigGcp'
                 404:
                     description: Not found
         put:
@@ -381,13 +385,20 @@ components:
             properties:
                 compatibilityLevel:
                     $ref: '#/components/schemas/Compatibility'
-                # GCP Managed Kafka Schema registries specific fields
+            required:
+                - compatibilityLevel
+
+        CompatibilityConfigGcp:
+            type: object
+            properties:
                 alias:
-                    type: string
+                  type: string
                 compatibility:
-                    $ref: '#/components/schemas/Compatibility'
+                  $ref: '#/components/schemas/Compatibility'
                 normalize:
-                    type: boolean
+                  type: boolean
+            required:
+                - compatibility
 
         CompatibilityLevelChange:
             type: object
@@ -397,7 +408,6 @@ components:
             required:
                 - compatibility
 
-
         Compatibility:
             type: string
             enum:
@@ -409,7 +419,6 @@ components:
                 - FULL_TRANSITIVE
                 - NONE
 
-
         CompatibilityCheckResponse:
             type: object
             properties: