Add global error handler to response with exception details

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/exception/CustomBaseException.java

@@ -0,0 +1,26 @@
+package com.provectus.kafka.ui.cluster.exception;
+
+import org.springframework.http.HttpStatus;
+
+public abstract class CustomBaseException extends RuntimeException {
+    public CustomBaseException() {
+    }
+
+    public CustomBaseException(String message) {
+        super(message);
+    }
+
+    public CustomBaseException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public CustomBaseException(Throwable cause) {
+        super(cause);
+    }
+
+    public CustomBaseException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+        super(message, cause, enableSuppression, writableStackTrace);
+    }
+
+    public abstract HttpStatus getResponseStatusCode();
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/exception/GlobalErrorAttributes.java

@@ -0,0 +1,32 @@
+package com.provectus.kafka.ui.cluster.exception;
+
+import org.springframework.boot.web.reactive.error.DefaultErrorAttributes;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.client.WebClientResponseException;
+import org.springframework.web.reactive.function.server.ServerRequest;
+
+import java.util.Map;
+
+@Component
+public class GlobalErrorAttributes extends DefaultErrorAttributes {
+
+    public static final String STATUS = "status";
+
+    @Override
+    public Map<String, Object> getErrorAttributes(ServerRequest request, boolean includeStackTrace) {
+        Map<String, Object> errorAttrs = super.getErrorAttributes(request, includeStackTrace);
+        includeCustomErrorAttributes(request, errorAttrs);
+        return errorAttrs;
+    }
+
+    private void includeCustomErrorAttributes(ServerRequest request, Map<String, Object> errorAttrs) {
+        Throwable error = getError(request);
+        if (error instanceof WebClientResponseException) {
+            var webClientError = (WebClientResponseException) error;
+            errorAttrs.put(STATUS, webClientError.getStatusCode());
+        } else if (error instanceof CustomBaseException) {
+            var customBaseError = (CustomBaseException) error;
+            errorAttrs.put(STATUS, customBaseError.getResponseStatusCode());
+        }
+    }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/exception/GlobalErrorWebExceptionHandler.java

@@ -0,0 +1,48 @@
+package com.provectus.kafka.ui.cluster.exception;
+
+import org.springframework.boot.autoconfigure.web.ResourceProperties;
+import org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler;
+import org.springframework.boot.web.reactive.error.ErrorAttributes;
+import org.springframework.context.ApplicationContext;
+import org.springframework.core.annotation.Order;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.ServerCodecConfigurer;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.server.*;
+import reactor.core.publisher.Mono;
+
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * The order of our global error handler is -2 to give it a higher priority than the default {@link org.springframework.boot.autoconfigure.web.reactive.error.DefaultErrorWebExceptionHandler}
+ * which is registered at <code>@Order(-1)</code>.
+ */
+@Component
+@Order(-2)
+public class GlobalErrorWebExceptionHandler extends AbstractErrorWebExceptionHandler {
+
+    public GlobalErrorWebExceptionHandler(GlobalErrorAttributes errorAttributes, ResourceProperties resourceProperties, ApplicationContext applicationContext,
+                                          ServerCodecConfigurer codecConfigurer) {
+        super(errorAttributes, resourceProperties, applicationContext);
+        this.setMessageWriters(codecConfigurer.getWriters());
+    }
+
+    @Override
+    protected RouterFunction<ServerResponse> getRoutingFunction(ErrorAttributes errorAttributes) {
+        return RouterFunctions.route(RequestPredicates.all(), this::renderErrorResponse);
+    }
+
+    private Mono<ServerResponse> renderErrorResponse(ServerRequest request) {
+        Map<String, Object> errorAttributes = getErrorAttributes(request, false);
+        HttpStatus statusCode = Optional.ofNullable(errorAttributes.get(GlobalErrorAttributes.STATUS))
+                .map(code -> (HttpStatus) code)
+                .orElse(HttpStatus.BAD_REQUEST);
+        return ServerResponse
+                .status(statusCode)
+                .contentType(MediaType.APPLICATION_JSON)
+                .body(BodyInserters.fromValue(errorAttributes));
+    }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/exception/NotFoundException.java

@@ -1,16 +1,15 @@
 package com.provectus.kafka.ui.cluster.exception;
 
 import org.springframework.http.HttpStatus;
-import org.springframework.web.bind.annotation.ResponseStatus;
 
-@ResponseStatus(HttpStatus.NOT_FOUND)
-public class NotFoundException extends RuntimeException {
-
-    public NotFoundException() {
-        super();
-    }
+public class NotFoundException extends CustomBaseException {
 
     public NotFoundException(String message) {
         super(message);
     }
+
+    @Override
+    public HttpStatus getResponseStatusCode() {
+        return HttpStatus.NOT_FOUND;
+    }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/SchemaRegistryService.java

@@ -16,7 +16,6 @@ import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 import org.springframework.web.reactive.function.BodyInserters;
-import org.springframework.web.reactive.function.client.ClientResponse;
 import org.springframework.web.reactive.function.client.WebClient;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -42,7 +41,6 @@ public class SchemaRegistryService {
                 .map(cluster -> webClient.get()
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECTS)
                         .retrieve()
-                        .onStatus(HttpStatus::is5xxServerError, ClientResponse::createException)
                         .bodyToFlux(String.class)
                         .doOnError(log::error))
                 .orElse(Flux.error(new NotFoundException("No such cluster")));
@@ -53,7 +51,7 @@ public class SchemaRegistryService {
                 .map(cluster -> webClient.get()
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECT_VERSIONS, schemaName)
                         .retrieve()
-                        .onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such subject")))
+                        .onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such schema %s".formatted(schemaName))))
                         .bodyToFlux(Integer.class))
                 .orElse(Flux.error(new NotFoundException("No such cluster")));
     }
@@ -71,7 +69,8 @@ public class SchemaRegistryService {
                 .map(cluster -> webClient.get()
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECT_BY_VERSION, schemaName, version)
                         .retrieve()
-                        .onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such subject or version")))
+                        .onStatus(HttpStatus.NOT_FOUND::equals,
+                                resp -> Mono.error(new NotFoundException("No such schema %s with version %s".formatted(schemaName, version))))
                         .bodyToMono(SchemaSubject.class)
                         .zipWith(getSchemaCompatibilityInfoOrGlobal(clusterName, schemaName))
                         .map(tuple -> {
@@ -81,7 +80,7 @@ public class SchemaRegistryService {
                             return schema;
                         })
                 )
-                .orElse(Mono.error(new NotFoundException()));
+                .orElseThrow();
     }
 
     public Mono<ResponseEntity<Void>> deleteSchemaSubjectByVersion(String clusterName, String schemaName, Integer version) {
@@ -97,7 +96,8 @@ public class SchemaRegistryService {
                 .map(cluster -> webClient.delete()
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECT_BY_VERSION, schemaName, version)
                         .retrieve()
-                        .onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such subject or version")))
+                        .onStatus(HttpStatus.NOT_FOUND::equals,
+                                resp -> Mono.error(new NotFoundException("No such schema %s with version %s".formatted(schemaName, version))))
                         .toBodilessEntity())
                 .orElse(Mono.error(new NotFoundException("No such cluster")));
     }
@@ -107,19 +107,20 @@ public class SchemaRegistryService {
                 .map(cluster -> webClient.delete()
                         .uri(cluster.getSchemaRegistry() + URL_SUBJECT, schemaName)
                         .retrieve()
-                        .onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such subject or version")))
+                        .onStatus(HttpStatus.NOT_FOUND::equals, resp -> Mono.error(new NotFoundException("No such schema %s".formatted(schemaName))))
                         .toBodilessEntity())
                 .orElse(Mono.error(new NotFoundException("No such cluster")));
     }
 
-    public Mono<ResponseEntity<SchemaSubject>> createNewSubject(String clusterName, String subjectSchema, Mono<NewSchemaSubject> newSchemaSubject) {
+    public Mono<ResponseEntity<SchemaSubject>> createNewSubject(String clusterName, String schemaName, Mono<NewSchemaSubject> newSchemaSubject) {
         return clustersStorage.getClusterByName(clusterName)
                 .map(cluster -> webClient.post()
-                        .uri(cluster.getSchemaRegistry() + URL_SUBJECT_VERSIONS, subjectSchema)
+                        .uri(cluster.getSchemaRegistry() + URL_SUBJECT_VERSIONS, schemaName)
                         .contentType(MediaType.APPLICATION_JSON)
                         .body(BodyInserters.fromPublisher(newSchemaSubject, NewSchemaSubject.class))
                         .retrieve()
-                        .onStatus(HttpStatus::isError, ClientResponse::createException)
+                        .onStatus(HttpStatus.NOT_FOUND::equals,
+                                resp -> Mono.error(new NotFoundException("No such schema %s".formatted(schemaName))))
                         .toEntity(SchemaSubject.class)
                         .log())
                 .orElse(Mono.error(new NotFoundException("No such cluster")));
@@ -140,6 +141,8 @@ public class SchemaRegistryService {
                             .contentType(MediaType.APPLICATION_JSON)
                             .body(BodyInserters.fromPublisher(compatibilityLevel, CompatibilityLevel.class))
                             .retrieve()
+                            .onStatus(HttpStatus.NOT_FOUND::equals,
+                                    resp -> Mono.error(new NotFoundException("No such schema %s".formatted(schemaName))))
                             .bodyToMono(Void.class);
                 }).orElse(Mono.error(new NotFoundException("No such cluster")));
     }
@@ -177,6 +180,8 @@ public class SchemaRegistryService {
                         .contentType(MediaType.APPLICATION_JSON)
                         .body(BodyInserters.fromPublisher(newSchemaSubject, NewSchemaSubject.class))
                         .retrieve()
+                        .onStatus(HttpStatus.NOT_FOUND::equals,
+                                resp -> Mono.error(new NotFoundException("No such schema %s".formatted(schemaName))))
                         .bodyToMono(InternalCompatibilityCheck.class)
                         .map(mapper::toCompatibilityCheckResponse)
                         .log()