Repository: provectus/kafka-ui
Issue: Schema registry back compatibility
Labels: type/bug
good first issue
scope/backend
status/accepted
status/confirmed

**Describe the bug**
There's incompatibility problems with schema registry


**Set up**
docker-compose kafka-ui.yaml

**Steps to Reproduce**
Cluster 1' schema registry (5.2.4): creating a protobuf schema doesn't work.
Schema example:
```
syntax = "proto3";
package com.provectus;

message TestProtoRecord {
  string f1 = 1;
  int32 f2 = 2;
}

```
Cluster 2' schema registry (5.5.0) works fine.

**Additional context**
```
{code: 4006, message: "Unrecognized field: schemaType", timestamp: 1639472891435,…}
code: 4006
fieldsErrors: null
message: "Unrecognized field: schemaType"
requestId: "51c9accd-15"
stackTrace: "com.provectus.kafka.ui.exception.UnprocessableEntityException: Unrecognized field: schemaType\n\tat com.provectus.kafka.ui.service.SchemaRegistryService.lambda$checkSchemaOnDuplicate$15(SchemaRegistryService.java:234)\n\tSuppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: \nError has been observed at the following site(s):\n\t*__checkpoint ⇢ 422 from POST http://schemaregistry0:8085/subjects/proto1 [DefaultWebClient]\n\t*__checkpoint ⇢ 
```

Issue comments:

Pull request review comments:
Could we state the minimum required version of SR which supports schemaType here please?
Updated exception message with the minimum required version of SR. @Haarolean  Could you check it, pls?
Nice, thanks

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/SchemaTypeIsNotSupportedException.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/SchemaTypeIsNotSupportedException.java
new file mode 100644
index 00000000000..eabaaf97e52
--- /dev/null
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/SchemaTypeIsNotSupportedException.java
@@ -0,0 +1,12 @@
+package com.provectus.kafka.ui.exception;
+
+public class SchemaTypeIsNotSupportedException extends UnprocessableEntityException {
+
+  private static final String REQUIRED_SCHEMA_REGISTRY_VERSION = "5.5.0";
+
+  public SchemaTypeIsNotSupportedException() {
+    super(String.format("Current version of Schema Registry does "
+            + "not support provided schema type,"
+            + " version %s or later is required here.", REQUIRED_SCHEMA_REGISTRY_VERSION));
+  }
+}
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
index 52db7d7ee30..16677e4c6c2 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
@@ -5,6 +5,7 @@
 
 import com.provectus.kafka.ui.exception.DuplicateEntityException;
 import com.provectus.kafka.ui.exception.SchemaNotFoundException;
+import com.provectus.kafka.ui.exception.SchemaTypeIsNotSupportedException;
 import com.provectus.kafka.ui.exception.UnprocessableEntityException;
 import com.provectus.kafka.ui.exception.ValidationException;
 import com.provectus.kafka.ui.mapper.ClusterMapper;
@@ -51,6 +52,8 @@ public class SchemaRegistryService {
   private static final String URL_SUBJECT_BY_VERSION = "/subjects/{schemaName}/versions/{version}";
   private static final String LATEST = "latest";
 
+  private static final String UNRECOGNIZED_FIELD_SCHEMA_TYPE = "Unrecognized field: schemaType";
+
   private final ClusterMapper mapper;
   private final WebClient webClient;
 
@@ -195,7 +198,9 @@ private Mono<SubjectIdResponse> submitNewSchema(String subject,
         .retrieve()
         .onStatus(UNPROCESSABLE_ENTITY::equals,
             r -> r.bodyToMono(ErrorResponse.class)
-                .flatMap(x -> Mono.error(new UnprocessableEntityException(x.getMessage()))))
+                .flatMap(x -> Mono.error(isUnrecognizedFieldSchemaTypeMessage(x.getMessage())
+                        ? new SchemaTypeIsNotSupportedException()
+                        : new UnprocessableEntityException(x.getMessage()))))
         .bodyToMono(SubjectIdResponse.class);
   }
 
@@ -213,7 +218,9 @@ private Mono<SchemaSubjectDTO> checkSchemaOnDuplicate(String subject,
         .onStatus(NOT_FOUND::equals, res -> Mono.empty())
         .onStatus(UNPROCESSABLE_ENTITY::equals,
             r -> r.bodyToMono(ErrorResponse.class)
-                .flatMap(x -> Mono.error(new UnprocessableEntityException(x.getMessage()))))
+                .flatMap(x -> Mono.error(isUnrecognizedFieldSchemaTypeMessage(x.getMessage())
+                        ? new SchemaTypeIsNotSupportedException()
+                        : new UnprocessableEntityException(x.getMessage()))))
         .bodyToMono(SchemaSubjectDTO.class)
         .filter(s -> Objects.isNull(s.getId()))
         .switchIfEmpty(Mono.error(new DuplicateEntityException("Such schema already exists")));
@@ -321,4 +328,8 @@ private WebClient.RequestBodySpec configuredWebClient(InternalSchemaRegistry sch
         .uri(schemaRegistry.getFirstUrl() + uri, params)
         .headers(headers -> setBasicAuthIfEnabled(schemaRegistry, headers));
   }
+
+  private boolean isUnrecognizedFieldSchemaTypeMessage(String errorMessage) {
+    return errorMessage.contains(UNRECOGNIZED_FIELD_SCHEMA_TYPE);
+  }
 }


Test patch: