fix: fixed how wrapped KsqlTopicAuthorizationException error messages…

… are displayed (#3258)
confluentinc · Aug 23, 2019 · 63672ae · 63672ae
1 parent 4fb8298
commit 63672ae
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 7 deletions.
diff --git a/ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/Errors.java b/ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/Errors.java
@@ -72,10 +72,10 @@ public static Response accessDenied(final String msg) {
         .build();
   }
 
-  public static Response accessDeniedFromKafka(final Exception error) {
+  public static Response accessDeniedFromKafka(final Throwable t) {
     return Response
         .status(FORBIDDEN)
-        .entity(new KsqlErrorMessage(ERROR_CODE_FORBIDDEN_KAFKA_ACCESS, error.getMessage()))
+        .entity(new KsqlErrorMessage(ERROR_CODE_FORBIDDEN_KAFKA_ACCESS, t))
         .build();
   }
 

diff --git a/ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java b/ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java
@@ -663,9 +663,10 @@ public void shouldReturnForbiddenKafkaAccessIfKsqlTopicAuthorizationException()
   @Test
   public void shouldReturnForbiddenKafkaAccessIfRootCauseKsqlTopicAuthorizationException() {
     // Given:
-    doThrow(new KsqlException("", new KsqlTopicAuthorizationException(
-        AclOperation.DELETE,
-        Collections.singleton("topic")))).when(topicAccessValidator).validate(any(), any(), any());
+    doThrow(new KsqlException("Could not delete the corresponding kafka topic: topic",
+        new KsqlTopicAuthorizationException(
+          AclOperation.DELETE,
+          Collections.singleton("topic")))).when(topicAccessValidator).validate(any(), any(), any());
 
 
     // When:
@@ -676,6 +677,9 @@ public void shouldReturnForbiddenKafkaAccessIfRootCauseKsqlTopicAuthorizationExc
     // Then:
     assertThat(result, is(instanceOf(KsqlErrorMessage.class)));
     assertThat(result.getErrorCode(), is(Errors.ERROR_CODE_FORBIDDEN_KAFKA_ACCESS));
+    assertThat(result.getMessage(), is(
+        "Could not delete the corresponding kafka topic: topic\n" +
+              "Caused by: Authorization denied to Delete on topic(s): [topic]"));
   }
 
   @Test

diff --git a/...-app/src/test/java/io/confluent/ksql/rest/server/resources/StreamedQueryResourceTest.java b/...-app/src/test/java/io/confluent/ksql/rest/server/resources/StreamedQueryResourceTest.java
@@ -48,6 +48,7 @@
 import io.confluent.ksql.parser.tree.Statement;
 import io.confluent.ksql.planner.PlanSourceExtractorVisitor;
 import io.confluent.ksql.planner.plan.OutputNode;
+import io.confluent.ksql.rest.entity.KsqlErrorMessage;
 import io.confluent.ksql.rest.entity.KsqlRequest;
 import io.confluent.ksql.rest.entity.StreamedRow;
 import io.confluent.ksql.rest.server.StatementParser;
@@ -465,8 +466,10 @@ public void shouldReturnForbiddenKafkaAccessIfKsqlTopicAuthorizationException()
     final Response expected = Errors.accessDeniedFromKafka(
         new KsqlTopicAuthorizationException(AclOperation.READ, Collections.singleton(topicName)));
 
+    final KsqlErrorMessage responseEntity = (KsqlErrorMessage) response.getEntity();
+    final KsqlErrorMessage expectedEntity = (KsqlErrorMessage) expected.getEntity();
     assertEquals(response.getStatus(), expected.getStatus());
-    assertEquals(response.getEntity(), expected.getEntity());
+    assertEquals(responseEntity.getMessage(), expectedEntity.getMessage());
   }
 
   @Test
@@ -497,8 +500,10 @@ public void shouldReturnForbiddenKafkaAccessIfRootCauseKsqlTopicAuthorizationExc
             "",
             new KsqlTopicAuthorizationException(AclOperation.READ, Collections.singleton(topicName))));
 
+    final KsqlErrorMessage responseEntity = (KsqlErrorMessage) response.getEntity();
+    final KsqlErrorMessage expectedEntity = (KsqlErrorMessage) expected.getEntity();
     assertEquals(response.getStatus(), expected.getStatus());
-    assertEquals(response.getEntity(), expected.getEntity());
+    assertEquals(responseEntity.getMessage(), expectedEntity.getMessage());
   }
 
   @Test