1. Unfortunately this is the http code from the schema registry itself. It's 500 instead of 409 if you send an invalid request, like this:
`{"subject":"test","schema":"test","schemaType":"AVRO"}`
instead of this:
`{"subject":"test","schemaType":"AVRO","schema":"{\"type\": \"string\"}"}`
It produces 409 if the request body is correct.

2. Seems like it's impossible since schema registry does not forward a list of such fields, but we'll forward an error message from schema registry instead.
```
{
  "code": 4006,
  "message": "Either the input schema or one its references is invalid", // <-- this has changed
  "timestamp": 1626784717378,
  "requestId": "73e2c259",
  "fieldsErrors": null
}
```

3. Currently this request returns code 4001:
```
{
  "code": 4001,
  "message": "Topic 'testXXX' already exists.",
}
```
I'm not quite sure if this is still the correct error code. There's a code 4005 (DUPLICATED_ENTITY) as well which might suit this purpose better. What do you folks think? @IldarAlmakaev @workshur 
Backend returns incorrect http statuses
Some Create/Update requests returns unpredictable responses:

1. When I try to create schema dup
```
curl 'http://localhost:3000/api/clusters/secondLocal/schemas' \
  -H 'Connection: keep-alive' \
  -H 'sec-ch-ua: "Google Chrome";v="89", "Chromium";v="89", ";Not A Brand";v="99"' \
  -H 'DNT: 1' \
  -H 'sec-ch-ua-mobile: ?0' \
  -H 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 11_1_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.90 Safari/537.36' \
  -H 'Content-Type: application/json' \
  -H 'Accept: */*' \
  -H 'Origin: http://localhost:3000' \
  -H 'Sec-Fetch-Site: same-origin' \
  -H 'Sec-Fetch-Mode: cors' \
  -H 'Sec-Fetch-Dest: empty' \
  -H 'Referer: http://localhost:3000/ui/clusters/secondLocal/schemas/new' \
  -H 'Accept-Language: en-US,en;q=0.9,ru;q=0.8' \
  --data-raw '{"subject":"my.new","schema":"my.new","schemaType":"AVRO"}' \
  --compressed
  ```
it returns HTTP/1.1 500 Internal Server Error with:
```
{"code":5000,"message":"500 Internal Server Error from POST http://schemaregistry0:8085/subjects/my.new","timestamp":1617110665799,"requestId":"adb03f44","fieldsErrors":null}
```
instead of 409

2. Schema Update should return list of invalid fields in 422 response
3. Topic create dup request returns correct message with incorrect code
```
{"code":5000,"message":"Topic '123' already exists.","timestamp":1617110943045,"requestId":"71474ae8","fieldsErrors":null}
```

Please update All existing api endpoints to return correct responses in format of ErrorResponse
kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/schemaregistry/ErrorResponse.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/schemaregistry/ErrorResponse.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/schemaregistry/ErrorResponse.java
new file mode 100644
index 00000000000..a7c86b01f19
--- /dev/null
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/schemaregistry/ErrorResponse.java
@@ -0,0 +1,14 @@
+package com.provectus.kafka.ui.model.schemaregistry;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
+
+@Data
+public class ErrorResponse {
+
+  @JsonProperty("error_code")
+  private int errorCode;
+
+  private String message;
+
+}
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 3f05ddb186f..41374120b81 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
@@ -14,6 +14,7 @@
 import com.provectus.kafka.ui.model.NewSchemaSubject;
 import com.provectus.kafka.ui.model.SchemaSubject;
 import com.provectus.kafka.ui.model.SchemaType;
+import com.provectus.kafka.ui.model.schemaregistry.ErrorResponse;
 import com.provectus.kafka.ui.model.schemaregistry.InternalCompatibilityCheck;
 import com.provectus.kafka.ui.model.schemaregistry.InternalCompatibilityLevel;
 import com.provectus.kafka.ui.model.schemaregistry.InternalNewSchema;
@@ -195,7 +196,8 @@ private Mono<SubjectIdResponse> submitNewSchema(String subject,
         .body(BodyInserters.fromPublisher(newSchemaSubject, InternalNewSchema.class))
         .retrieve()
         .onStatus(UNPROCESSABLE_ENTITY::equals,
-            r -> Mono.error(new UnprocessableEntityException("Invalid params")))
+            r -> r.bodyToMono(ErrorResponse.class)
+                .flatMap(x -> Mono.error(new UnprocessableEntityException(x.getMessage()))))
         .bodyToMono(SubjectIdResponse.class);
   }
 
@@ -210,7 +212,8 @@ private Mono<SchemaSubject> checkSchemaOnDuplicate(String subject,
         .retrieve()
         .onStatus(NOT_FOUND::equals, res -> Mono.empty())
         .onStatus(UNPROCESSABLE_ENTITY::equals,
-            r -> Mono.error(new UnprocessableEntityException("Invalid params")))
+            r -> r.bodyToMono(ErrorResponse.class)
+                .flatMap(x -> Mono.error(new UnprocessableEntityException(x.getMessage()))))
         .bodyToMono(SchemaSubject.class)
         .filter(s -> Objects.isNull(s.getId()))
         .switchIfEmpty(Mono.error(new DuplicateEntityException("Such schema already exists")));
