Improved logs on schemas timeout and CLI format (#207)

Loïc GREFFIER · web-flow · commit 1f2443c1aaaa · 2022-08-05T11:37:36.000+02:00
* Improved logs on schemas timeout and CLI format

* Fix unit tests

* Fix unit tests
diff --git a/api/src/main/java/com/michelin/ns4kafka/controllers/ExceptionHandlerController.java b/api/src/main/java/com/michelin/ns4kafka/controllers/ExceptionHandlerController.java
@@ -11,6 +11,7 @@
 import io.micronaut.http.annotation.Error;
 import io.micronaut.security.authentication.AuthenticationException;
 import io.micronaut.security.authentication.AuthorizationException;
+import lombok.extern.slf4j.Slf4j;
 
 import javax.validation.ConstraintViolation;
 import javax.validation.ConstraintViolationException;
@@ -20,6 +21,7 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
+@Slf4j
 @Controller("/errors")
 public class ExceptionHandlerController {
     @Error(global = true)
@@ -122,19 +124,22 @@ public HttpResponse<Status> error(HttpRequest<?> request, AuthorizationException
 
     }
 
-    // if we don't know the exception
     @Error(global = true)
     public HttpResponse<Status> error(HttpRequest<?> request, Exception exception) {
-        var status = Status.builder()
+        log.error("An error occurred on API endpoint {} {}: {}", request.getMethodName(), request.getUri(), exception.getMessage());
+
+        Status status = Status.builder()
                 .status(StatusPhase.Failed)
                 .message("Internal server error")
                 .reason(StatusReason.InternalError)
                 .details(StatusDetails.builder()
-                        .causes(List.of(exception.toString()))
+                        .causes(List.of(exception.getMessage() != null ? exception.getMessage() : exception.toString()))
                         .build())
                 .code(HttpStatus.INTERNAL_SERVER_ERROR.getCode())
                 .build();
-        return HttpResponse.status(HttpStatus.INTERNAL_SERVER_ERROR)
+
+        return HttpResponse
+                .status(HttpStatus.INTERNAL_SERVER_ERROR)
                 .body(status);
     }
 }
diff --git a/api/src/main/java/com/michelin/ns4kafka/services/SchemaService.java b/api/src/main/java/com/michelin/ns4kafka/services/SchemaService.java
@@ -10,7 +10,6 @@
 import com.michelin.ns4kafka.services.schema.client.entities.SchemaCompatibilityResponse;
 import com.michelin.ns4kafka.services.schema.client.entities.SchemaRequest;
 import com.michelin.ns4kafka.services.schema.client.entities.SchemaResponse;
-import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.reactivex.Maybe;
 import io.reactivex.Single;
 import lombok.extern.slf4j.Slf4j;
@@ -165,16 +164,19 @@ public Single<List<String>> validateSchemaCompatibility(String cluster, Schema s
                         .schema(schema.getSpec().getSchema())
                         .references(schema.getSpec().getReferences())
                         .build())
-                .flatMap(schemaCompatibilityCheckSuccess -> {
-                            if (!schemaCompatibilityCheckSuccess.isCompatible()) {
-                                return Maybe.just(schemaCompatibilityCheckSuccess.messages());
-                            } else {
-                                return Maybe.just(List.<String>of());
+                .map(Optional::of)
+                .defaultIfEmpty(Optional.empty())
+                .flatMapSingle(schemaCompatibilityCheckOptional -> {
+                            if (schemaCompatibilityCheckOptional.isEmpty()) {
+                                return Single.just(List.of());
                             }
-                        },
-                        schemaCompatibilityCheckError -> Maybe.just(List.of("An error occurred during the schema validation (status code: " + ((HttpClientResponseException) schemaCompatibilityCheckError).getStatus() + ")")),
-                        () -> Maybe.just(List.<String>of()))
-                .flatMapSingle(Single::just);
+
+                            if (!schemaCompatibilityCheckOptional.get().isCompatible()) {
+                                return Single.just(schemaCompatibilityCheckOptional.get().messages());
+                            } 
+
+                            return Single.just(List.of());
+                        });
     }
 
     /**
diff --git a/api/src/main/java/com/michelin/ns4kafka/services/executors/TopicAsyncExecutor.java b/api/src/main/java/com/michelin/ns4kafka/services/executors/TopicAsyncExecutor.java
@@ -86,11 +86,9 @@ public void synchronizeTopics(){
                     .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
 
             if(log.isDebugEnabled()){
-                log.debug("Topics to create : "+ toCreate.stream().map(t -> t.getMetadata().getName()).collect(Collectors.joining(", ")));
-                //TODO reenable
-                // LOG.debug("Topics to delete : "+String.join(", ", toDelete.stream().map(t -> t.getMetadata().getName()).collect(Collectors.toList())));
-                log.debug("Topics to delete : "+toDelete.size());
-                log.debug("Topic configs to update : "+toUpdate.size());
+                log.debug("Topics to create: " + toCreate.stream().map(t -> t.getMetadata().getName()).collect(Collectors.joining(", ")));
+                log.debug("Topics to delete: " + toDelete.size());
+                log.debug("Topics to update: " + toUpdate.size());
                 for (Map.Entry<ConfigResource,Collection<AlterConfigOp>> e : toUpdate.entrySet()) {
                     for (AlterConfigOp op : e.getValue()) {
                         log.debug(e.getKey().name()+" "+op.opType().toString()+" " +op.configEntry().name()+"("+op.configEntry().value()+")");
diff --git a/api/src/main/java/com/michelin/ns4kafka/services/schema/KafkaSchemaRegistryClientProxy.java b/api/src/main/java/com/michelin/ns4kafka/services/schema/KafkaSchemaRegistryClientProxy.java
@@ -10,7 +10,6 @@
 import io.micronaut.http.MutableHttpResponse;
 import io.micronaut.http.annotation.Filter;
 import io.micronaut.http.client.ProxyHttpClient;
-import io.micronaut.http.filter.HttpServerFilter;
 import io.micronaut.http.filter.OncePerRequestHttpServerFilter;
 import io.micronaut.http.filter.ServerFilterChain;
 import org.reactivestreams.Publisher;
@@ -24,7 +23,7 @@
 @Filter(KafkaSchemaRegistryClientProxy.SCHEMA_REGISTRY_PREFIX + "/**")
 public class KafkaSchemaRegistryClientProxy extends OncePerRequestHttpServerFilter {
     /**
-     * Schema registry prefix used to filter request to schema registries. It'll be replace by
+     * Schema registry prefix used to filter request to schema registries. It'll be replaced by
      * the schema registry URL of the
      */
     public static final String SCHEMA_REGISTRY_PREFIX = "/schema-registry-proxy";
@@ -94,7 +93,7 @@ public Publisher<MutableHttpResponse<?>> doFilterOnce(HttpRequest<?> request, Se
             return Publishers.just(new ResourceValidationException(List.of("Kafka Cluster [" + kafkaCluster + "] has no schema registry"),null,null));
         }
 
-        return this.client.proxy(mutateSchemaRegistryRequest(request, config.get()));
+        return client.proxy(mutateSchemaRegistryRequest(request, config.get()));
     }
 
     /**
diff --git a/api/src/test/java/com/michelin/ns4kafka/controllers/ExceptionHandlerControllerTest.java b/api/src/test/java/com/michelin/ns4kafka/controllers/ExceptionHandlerControllerTest.java
@@ -1,27 +1,25 @@
 package com.michelin.ns4kafka.controllers;
 
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Disabled;
-import org.junit.jupiter.api.Test;
-
 import io.micronaut.http.HttpMethod;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.HttpStatus;
 import io.micronaut.security.authentication.AuthenticationException;
 import io.micronaut.security.authentication.AuthorizationException;
 import io.micronaut.security.authentication.DefaultAuthentication;
-import javax.validation.ConstraintViolationException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
 
+import javax.validation.ConstraintViolationException;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
-public class ExceptionHandlerControllerTest {
+class ExceptionHandlerControllerTest {
 
     ExceptionHandlerController exceptionHandlerController = new ExceptionHandlerController();
 
     @Test
-    void ressourceValidationError() {
+    void resourceValidationError() {
         var response = exceptionHandlerController.error(HttpRequest.create(HttpMethod.POST, "local")
                                                       ,new ResourceValidationException(List.of("Error1", "Error2"),"Topic", "Name"));
         var status = response.body();
@@ -36,15 +34,13 @@ void ressourceValidationError() {
     }
 
     @Test
-    @Disabled
     void constraintViolationError() {
         var response = exceptionHandlerController.error(HttpRequest.create(HttpMethod.POST, "local")
                                                       ,new ConstraintViolationException(Set.of()));
         var status = response.body();
 
         Assertions.assertEquals(HttpStatus.UNPROCESSABLE_ENTITY, response.getStatus());
         Assertions.assertEquals(HttpStatus.UNPROCESSABLE_ENTITY.getCode(), status.getCode());
-        //Assertions.assertEquals("Error1", status.getDetails().getCauses().get(0));
     }
 
     @Test
diff --git a/api/src/test/java/com/michelin/ns4kafka/services/SchemaServiceTest.java b/api/src/test/java/com/michelin/ns4kafka/services/SchemaServiceTest.java
@@ -9,8 +9,6 @@
 import com.michelin.ns4kafka.services.schema.client.entities.SchemaCompatibilityCheckResponse;
 import com.michelin.ns4kafka.services.schema.client.entities.SchemaCompatibilityResponse;
 import com.michelin.ns4kafka.services.schema.client.entities.SchemaResponse;
-import io.micronaut.http.HttpResponse;
-import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.reactivex.Maybe;
 import io.reactivex.Single;
 import org.junit.jupiter.api.Assertions;
@@ -247,24 +245,6 @@ void validateSchemaCompatibility404NotFound() {
                 .assertValue(List::isEmpty);
     }
 
-    /**
-     * Test the schema compatibility validation when the Schema Registry throws an exception
-     */
-    @Test
-    void validateSchemaCompatibilityThrowsException() {
-        Namespace namespace = buildNamespace();
-        Schema schema = buildSchema();
-
-        when(kafkaSchemaRegistryClient.validateSchemaCompatibility(any(), any(), any(), any()))
-                .thenReturn(Maybe.error(new HttpClientResponseException("Error", HttpResponse.notFound())));
-
-        schemaService.validateSchemaCompatibility(namespace.getMetadata().getCluster(), schema)
-                .test()
-                .assertValue(validationErrors -> validationErrors.size() == 1L)
-                .assertValue(validationErrors -> validationErrors.get(0)
-                        .equals("An error occurred during the schema validation (status code: NOT_FOUND)"));
-    }
-
     /**
      * Test the schema compatibility update when reset to default is asked
      */
diff --git a/cli/src/main/java/com/michelin/ns4kafka/cli/services/FormatService.java b/cli/src/main/java/com/michelin/ns4kafka/cli/services/FormatService.java
@@ -26,10 +26,6 @@
 
 @Singleton
 public class FormatService {
-
-    @Inject
-    public KafkactlConfig kafkactlConfig;
-
     private static final String YAML = "yaml";
     private static final String TABLE = "table";
     private final List<String> defaults = List.of(
@@ -38,6 +34,9 @@ public class FormatService {
             "AGE:/metadata/creationTimestamp%AGO"
     );
 
+    @Inject
+    public KafkactlConfig kafkactlConfig;
+
     public void displayList(String kind, List<Resource> resources, String output) {
         if (output.equals(TABLE)) {
             printTable(kind, resources);
@@ -60,11 +59,12 @@ public void displayError(HttpClientResponseException e, String kind, String name
         Optional<Status> statusOptional = e.getResponse().getBody(Status.class);
         if (statusOptional.isPresent() && statusOptional.get().getDetails() != null && !statusOptional.get().getDetails().getCauses().isEmpty()) {
             Status status = statusOptional.get();
-            String causes = String.join("\n - ", status.getDetails().getCauses());
+            String causes = status.getDetails().getCauses().size() > 1 ?
+                    "\n - " + String.join("\n - ", status.getDetails().getCauses()) : status.getDetails().getCauses().get(0);
 
-            System.out.printf("Failed %s/%s: %s for causes: %n - %s%n", kind, name, status.getMessage(), causes);
+            System.out.printf("Failed %s/%s %s for causes: %s", kind, name, status.getMessage(), causes);
         } else {
-            System.out.printf("Failed %s/%s: %s%n", kind, name, e.getMessage());
+            System.out.printf("Failed %s/%s %s%n", kind, name, e.getMessage());
         }
     }
 
