Issue better error responses to UI (#1133)

* Do a look up for the superadmin or admin to run the queries for migration to 2.0.0 Klaw Signed-off-by: Aindriu Lavelle <[email protected]> * Initial commit for better error response handling Signed-off-by: Aindriu Lavelle <[email protected]> * Return the correct error messages Signed-off-by: Aindriu Lavelle <[email protected]> * Update to responses Signed-off-by: Aindriu Lavelle <[email protected]> * Update to responses Signed-off-by: Aindriu Lavelle <[email protected]> * Update to responses Signed-off-by: Aindriu Lavelle <[email protected]> * Update to reduce duplication Signed-off-by: Aindriu Lavelle <[email protected]> * Additional exception handling to ensure that errors do not escape Signed-off-by: Aindriu Lavelle <[email protected]> * Update to address PRs Signed-off-by: Aindriu Lavelle <[email protected]> * Update to address PRs Signed-off-by: Aindriu Lavelle <[email protected]> * Update to address PRs Signed-off-by: Aindriu Lavelle <[email protected]> * Special character introduced in error messages had to be removed. Signed-off-by: Aindriu Lavelle <[email protected]> * Special character introduced in error messages had to be removed. Signed-off-by: Aindriu Lavelle <[email protected]> * run spotless Signed-off-by: Aindriu Lavelle <[email protected]> --------- Signed-off-by: Aindriu Lavelle <[email protected]>
Aiven-Open · May 10, 2023 · 622c7bf · 622c7bf
1 parent 4a653d0
commit 622c7bf
Show file tree

Hide file tree

Showing 16 changed files with 415 additions and 21 deletions.
diff --git a/cluster-api/src/main/java/io/aiven/klaw/clusterapi/controller/KafkaConnectController.java b/cluster-api/src/main/java/io/aiven/klaw/clusterapi/controller/KafkaConnectController.java
@@ -63,7 +63,7 @@ public ResponseEntity<ApiResponse> postConnector(
       return new ResponseEntity<>(result, HttpStatus.OK);
     } catch (Exception e) {
       return new ResponseEntity<>(
-          ApiResponse.builder().success(false).message("Unable to register connector").build(),
+          ApiResponse.builder().success(false).message(e.getMessage()).build(),
           HttpStatus.INTERNAL_SERVER_ERROR);
     }
   }

diff --git a/cluster-api/src/main/java/io/aiven/klaw/clusterapi/models/error/ClusterApiErrorMessages.java b/cluster-api/src/main/java/io/aiven/klaw/clusterapi/models/error/ClusterApiErrorMessages.java
@@ -0,0 +1,10 @@
+package io.aiven.klaw.clusterapi.models.error;
+
+public class ClusterApiErrorMessages {
+
+  public static final String CLUSTER_API_ERR_1 = "Unable to create Connector on Cluster.";
+
+  public static final String CLUSTER_API_ERR_2 = "Unable to update Connector on Cluster";
+
+  public static final String CLUSTER_API_ERR_3 = "Unable To Delete Connector on Cluster.";
+}
diff --git a/cluster-api/src/main/java/io/aiven/klaw/clusterapi/models/error/RestErrorResponse.java b/cluster-api/src/main/java/io/aiven/klaw/clusterapi/models/error/RestErrorResponse.java
@@ -0,0 +1,13 @@
+package io.aiven.klaw.clusterapi.models.error;
+
+import com.fasterxml.jackson.annotation.JsonAlias;
+import lombok.Data;
+
+@Data
+public class RestErrorResponse {
+
+  private String message;
+
+  @JsonAlias("error_code")
+  private int errorCode;
+}
diff --git a/cluster-api/src/main/java/io/aiven/klaw/clusterapi/services/KafkaConnectService.java b/cluster-api/src/main/java/io/aiven/klaw/clusterapi/services/KafkaConnectService.java
@@ -1,11 +1,16 @@
 package io.aiven.klaw.clusterapi.services;
 
+import static io.aiven.klaw.clusterapi.models.error.ClusterApiErrorMessages.CLUSTER_API_ERR_1;
+import static io.aiven.klaw.clusterapi.models.error.ClusterApiErrorMessages.CLUSTER_API_ERR_2;
+import static io.aiven.klaw.clusterapi.models.error.ClusterApiErrorMessages.CLUSTER_API_ERR_3;
+
 import io.aiven.klaw.clusterapi.models.ApiResponse;
 import io.aiven.klaw.clusterapi.models.ClusterConnectorRequest;
 import io.aiven.klaw.clusterapi.models.enums.ApiResultStatus;
 import io.aiven.klaw.clusterapi.models.enums.ClusterStatus;
 import io.aiven.klaw.clusterapi.models.enums.KafkaClustersType;
 import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
+import io.aiven.klaw.clusterapi.models.error.RestErrorResponse;
 import io.aiven.klaw.clusterapi.utils.ClusterApiUtils;
 import java.util.Collections;
 import java.util.HashMap;
@@ -19,6 +24,9 @@
 import org.springframework.http.HttpMethod;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
+import org.springframework.web.client.HttpClientErrorException;
+import org.springframework.web.client.HttpServerErrorException;
+import org.springframework.web.client.HttpStatusCodeException;
 import org.springframework.web.client.RestClientException;
 import org.springframework.web.client.RestTemplate;
 
@@ -58,9 +66,12 @@ public ApiResponse deleteConnector(ClusterConnectorRequest clusterConnectorReque
               HttpMethod.DELETE,
               request,
               new ParameterizedTypeReference<>() {});
-    } catch (RestClientException e) {
+    } catch (HttpServerErrorException | HttpClientErrorException e) {
       log.error("Error in deleting connector ", e);
-      return ApiResponse.builder().success(false).message(e.getMessage()).build();
+      return buildErrorResponseFromRestException(e, CLUSTER_API_ERR_3);
+    } catch (RestClientException ex) {
+      log.error("Error in deleting connector ", ex);
+      return ApiResponse.builder().success(false).message(CLUSTER_API_ERR_3).build();
     }
     return ApiResponse.builder().success(true).message(ApiResultStatus.SUCCESS.value).build();
   }
@@ -84,13 +95,30 @@ public ApiResponse updateConnector(ClusterConnectorRequest clusterConnectorReque
 
     try {
       reqDetails.getRight().put(reqDetails.getLeft(), request, String.class);
-    } catch (RestClientException e) {
+    } catch (HttpServerErrorException | HttpClientErrorException e) {
       log.error("Error in updating connector ", e);
-      return ApiResponse.builder().success(false).message(e.getMessage()).build();
+      return buildErrorResponseFromRestException(e, CLUSTER_API_ERR_2);
+    } catch (Exception ex) {
+      return ApiResponse.builder().success(false).message(CLUSTER_API_ERR_2).build();
     }
     return ApiResponse.builder().success(true).message(ApiResultStatus.SUCCESS.value).build();
   }
 
+  private static ApiResponse buildErrorResponseFromRestException(
+      HttpStatusCodeException e, String defaultErrorMsg) {
+    RestErrorResponse errorResponse = null;
+    try {
+      errorResponse = e.getResponseBodyAs(RestErrorResponse.class);
+    } catch (Exception ex) {
+      log.error("Error caught trying to process the error response. ", ex);
+    }
+    if (errorResponse != null) {
+      return ApiResponse.builder().success(false).message(errorResponse.getMessage()).build();
+    } else {
+      return ApiResponse.builder().success(false).message(defaultErrorMsg).build();
+    }
+  }
+
   public ApiResponse postNewConnector(ClusterConnectorRequest clusterConnectorRequest)
       throws Exception {
     log.info("Into postNewConnector clusterConnectorRequest {} ", clusterConnectorRequest);
@@ -110,9 +138,11 @@ public ApiResponse postNewConnector(ClusterConnectorRequest clusterConnectorRequ
     try {
       responseNew =
           reqDetails.getRight().postForEntity(reqDetails.getLeft(), request, String.class);
-    } catch (RestClientException e) {
-      log.error("Error in registering new connector ", e);
-      throw new Exception(e.toString());
+    } catch (HttpServerErrorException | HttpClientErrorException e) {
+
+      return buildErrorResponseFromRestException(e, CLUSTER_API_ERR_1);
+    } catch (Exception ex) {
+      return ApiResponse.builder().success(false).message(CLUSTER_API_ERR_1).build();
     }
     if (responseNew.getStatusCodeValue() == 201) {
       return ApiResponse.builder().success(true).message(ApiResultStatus.SUCCESS.value).build();

diff --git a/cluster-api/src/test/java/io/aiven/klaw/clusterapi/services/KafkaConnectServiceTest.java b/cluster-api/src/test/java/io/aiven/klaw/clusterapi/services/KafkaConnectServiceTest.java
@@ -5,11 +5,18 @@
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.when;
 import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo;
+import static org.springframework.test.web.client.response.MockRestResponseCreators.withBadRequest;
+import static org.springframework.test.web.client.response.MockRestResponseCreators.withRawStatus;
 import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.aiven.klaw.clusterapi.models.ApiResponse;
+import io.aiven.klaw.clusterapi.models.ClusterConnectorRequest;
+import io.aiven.klaw.clusterapi.models.enums.ApiResultStatus;
+import io.aiven.klaw.clusterapi.models.enums.KafkaClustersType;
 import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
+import io.aiven.klaw.clusterapi.models.error.RestErrorResponse;
 import io.aiven.klaw.clusterapi.utils.ClusterApiUtils;
 import java.util.Collections;
 import org.apache.commons.lang3.tuple.Pair;
@@ -19,13 +26,16 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.autoconfigure.web.client.RestClientTest;
 import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.test.web.client.MockRestServiceServer;
 import org.springframework.web.client.RestTemplate;
 
 @RestClientTest(KafkaConnectService.class)
 class KafkaConnectServiceTest {
 
+  public static final String THIS_IS_A_MISCONFIGURED_CONNECTOR =
+      "This is a misconfigured connector";
   @Autowired KafkaConnectService kafkaConnectService;
 
   RestTemplate restTemplate;
@@ -73,4 +83,153 @@ public void getConnectorDetails_returnMap() throws JsonProcessingException {
                 "conn1", "env", KafkaSupportedProtocol.PLAINTEXT, "CLID1"))
         .isNotEmpty();
   }
+
+  @Test
+  public void createConnector_bad_request() throws Exception {
+    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();
+
+    this.mockRestServiceServer
+        .expect(requestTo("/env/connectors/conn1"))
+        .andRespond(
+            withBadRequest()
+                .contentType(MediaType.APPLICATION_JSON)
+                .body(getRestErrorResponse(THIS_IS_A_MISCONFIGURED_CONNECTOR)));
+    ApiResponse connectorResponse = kafkaConnectService.postNewConnector(connectorRequest);
+
+    assertThat(connectorResponse.isSuccess()).isFalse();
+    assertThat(connectorResponse.getMessage()).isEqualTo(THIS_IS_A_MISCONFIGURED_CONNECTOR);
+  }
+
+  private String getRestErrorResponse(String resp) throws JsonProcessingException {
+    RestErrorResponse restErrorResponse = new RestErrorResponse();
+    restErrorResponse.setErrorCode(400);
+    restErrorResponse.setMessage(resp);
+    return objectMapper.writeValueAsString(restErrorResponse);
+  }
+
+  @Test
+  public void createConnector_success() throws Exception {
+    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();
+
+    this.mockRestServiceServer
+        .expect(requestTo("/env/connectors/conn1"))
+        .andRespond(withRawStatus(201).contentType(MediaType.APPLICATION_JSON));
+    ApiResponse connectorResponse = kafkaConnectService.postNewConnector(connectorRequest);
+    assertThat(connectorResponse.isSuccess()).isTrue();
+    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
+  }
+
+  @Test
+  public void createConnector_fail() throws Exception {
+    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();
+
+    this.mockRestServiceServer
+        .expect(requestTo("/env/connectors/conn1"))
+        .andRespond(withRawStatus(207).contentType(MediaType.APPLICATION_JSON));
+    ApiResponse connectorResponse = kafkaConnectService.postNewConnector(connectorRequest);
+    assertThat(connectorResponse.isSuccess()).isFalse();
+    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.FAILURE.value);
+  }
+
+  @Test
+  public void updateConnector_success() {
+    ClusterConnectorRequest connectorRequest = stubUpdateConnector();
+
+    this.mockRestServiceServer
+        .expect(requestTo("/env/connectors/conn1/config"))
+        .andRespond(withRawStatus(201).contentType(MediaType.APPLICATION_JSON));
+    ApiResponse connectorResponse = kafkaConnectService.updateConnector(connectorRequest);
+    assertThat(connectorResponse.isSuccess()).isTrue();
+    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
+  }
+
+  @Test
+  public void updateConnector_badRequest() throws JsonProcessingException {
+    ClusterConnectorRequest connectorRequest = stubUpdateConnector();
+
+    this.mockRestServiceServer
+        .expect(requestTo("/env/connectors/conn1/config"))
+        .andRespond(
+            withRawStatus(400)
+                .contentType(MediaType.APPLICATION_JSON)
+                .body(getRestErrorResponse(THIS_IS_A_MISCONFIGURED_CONNECTOR)));
+    ApiResponse connectorResponse = kafkaConnectService.updateConnector(connectorRequest);
+
+    assertThat(connectorResponse.isSuccess()).isFalse();
+    assertThat(connectorResponse.getMessage()).isEqualTo(THIS_IS_A_MISCONFIGURED_CONNECTOR);
+  }
+
+  @Test
+  public void deleteConnector_badRequest() throws JsonProcessingException {
+    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();
+
+    this.mockRestServiceServer
+        .expect(requestTo("/env/connectors/conn1"))
+        .andRespond(
+            withRawStatus(400)
+                .contentType(MediaType.APPLICATION_JSON)
+                .body(getRestErrorResponse(THIS_IS_A_MISCONFIGURED_CONNECTOR)));
+    ApiResponse connectorResponse = kafkaConnectService.deleteConnector(connectorRequest);
+
+    assertThat(connectorResponse.isSuccess()).isFalse();
+    assertThat(connectorResponse.getMessage()).isEqualTo(THIS_IS_A_MISCONFIGURED_CONNECTOR);
+  }
+
+  @Test
+  public void deleteConnector_badRequest_undetermined_response() throws JsonProcessingException {
+    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();
+
+    this.mockRestServiceServer
+        .expect(requestTo("/env/connectors/conn1"))
+        .andRespond(
+            withRawStatus(400)
+                .contentType(MediaType.APPLICATION_JSON)
+                .body(THIS_IS_A_MISCONFIGURED_CONNECTOR));
+    ApiResponse connectorResponse = kafkaConnectService.deleteConnector(connectorRequest);
+
+    assertThat(connectorResponse.isSuccess()).isFalse();
+    assertThat(connectorResponse.getMessage()).isEqualTo("Unable To Delete Connector on Cluster.");
+  }
+
+  @Test
+  public void deleteConnector_success() {
+    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();
+
+    this.mockRestServiceServer
+        .expect(requestTo("/env/connectors/conn1"))
+        .andRespond(withRawStatus(201).contentType(MediaType.APPLICATION_JSON));
+    ApiResponse connectorResponse = kafkaConnectService.deleteConnector(connectorRequest);
+    assertThat(connectorResponse.isSuccess()).isTrue();
+    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
+  }
+
+  private ClusterConnectorRequest stubCreateOrDeleteConnector() {
+    when(getAdminClient.getRequestDetails(any(), eq(KafkaSupportedProtocol.PLAINTEXT)))
+        .thenReturn(Pair.of("/env/connectors/conn1", restTemplate));
+    when(getAdminClient.createHeaders(eq("1"), eq(KafkaClustersType.KAFKA_CONNECT)))
+        .thenReturn(new HttpHeaders());
+    ClusterConnectorRequest connectorRequest =
+        ClusterConnectorRequest.builder()
+            .connectorName("conn1")
+            .clusterIdentification("1")
+            .env("env")
+            .protocol(KafkaSupportedProtocol.PLAINTEXT)
+            .build();
+    return connectorRequest;
+  }
+
+  private ClusterConnectorRequest stubUpdateConnector() {
+    when(getAdminClient.getRequestDetails(any(), eq(KafkaSupportedProtocol.PLAINTEXT)))
+        .thenReturn(Pair.of("/env/connectors/conn1/config", restTemplate));
+    when(getAdminClient.createHeaders(eq("1"), eq(KafkaClustersType.KAFKA_CONNECT)))
+        .thenReturn(new HttpHeaders());
+    ClusterConnectorRequest connectorRequest =
+        ClusterConnectorRequest.builder()
+            .connectorName("conn1")
+            .clusterIdentification("1")
+            .env("env")
+            .protocol(KafkaSupportedProtocol.PLAINTEXT)
+            .build();
+    return connectorRequest;
+  }
 }
diff --git a/core/src/main/java/io/aiven/klaw/controller/KafkaConnectController.java b/core/src/main/java/io/aiven/klaw/controller/KafkaConnectController.java
@@ -1,6 +1,7 @@
 package io.aiven.klaw.controller;
 
 import io.aiven.klaw.error.KlawException;
+import io.aiven.klaw.error.KlawRestException;
 import io.aiven.klaw.model.ApiResponse;
 import io.aiven.klaw.model.enums.Order;
 import io.aiven.klaw.model.enums.RequestOperationType;
@@ -88,7 +89,7 @@ public ResponseEntity<ApiResponse> deleteConnectorRequests(
       value = "/execConnectorRequests",
       produces = {MediaType.APPLICATION_JSON_VALUE})
   public ResponseEntity<ApiResponse> approveTopicRequests(
-      @RequestParam("connectorId") String connectorId) throws KlawException {
+      @RequestParam("connectorId") String connectorId) throws KlawException, KlawRestException {
     return new ResponseEntity<>(
         kafkaConnectControllerService.approveConnectorRequests(connectorId), HttpStatus.OK);
   }

diff --git a/core/src/main/java/io/aiven/klaw/error/KlawErrorMessages.java b/core/src/main/java/io/aiven/klaw/error/KlawErrorMessages.java
@@ -129,6 +129,8 @@ public class KlawErrorMessages {
 
   public static final String CLUSTER_API_ERR_121 = "Connection refused";
 
+  public static final String CLUSTER_API_ERR_122 = "doesn't match connector name in the URL";
+
   // Env clusters tenants service
   public static final String ENV_CLUSTER_TNT_ERR_101 =
       "Failure. Please choose a different name. This environment name already exists.";

diff --git a/core/src/main/java/io/aiven/klaw/error/RestErrorResponse.java b/core/src/main/java/io/aiven/klaw/error/RestErrorResponse.java
@@ -0,0 +1,13 @@
+package io.aiven.klaw.error;
+
+import com.fasterxml.jackson.annotation.JsonAlias;
+import lombok.Data;
+
+@Data
+public class RestErrorResponse {
+
+  private String message;
+
+  @JsonAlias({"errorCode", "error_code"})
+  private int errorCode;
+}
diff --git a/core/src/main/java/io/aiven/klaw/helpers/KwConstants.java b/core/src/main/java/io/aiven/klaw/helpers/KwConstants.java
@@ -81,6 +81,8 @@ public class KwConstants {
 
   public static final String ORDER_OF_TOPIC_ENVS = "ORDER_OF_ENVS";
 
+  public static final String ORDER_OF_KAFKA_CONNECT_ENVS = "ORDER_OF_KAFKA_CONNECT_ENVS";
+
   public static final int DAYS_EXPIRY_DEFAULT_TENANT = 365 * 10;
   public static final int DAYS_TRIAL_PERIOD = 7;