fix: error message with DROP DELETE TOPIC is invalid (#3279)

confluentinc · Aug 28, 2019 · 4284b8c · 4284b8c
1 parent 2ea6a2c
commit 4284b8c
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 1 deletion.
diff --git a/ksql-engine/src/main/java/io/confluent/ksql/topic/TopicDeleteInjector.java b/ksql-engine/src/main/java/io/confluent/ksql/topic/TopicDeleteInjector.java
@@ -103,7 +103,7 @@ public <T extends Statement> ConfiguredStatement<T> inject(
             ExecutorUtil.RetryBehaviour.ALWAYS);
       } catch (Exception e) {
         throw new RuntimeException("Could not delete the corresponding kafka topic: "
-                + sourceName, e);
+                + source.getKafkaTopicName(), e);
       }
 
       try {

diff --git a/ksql-engine/src/test/java/io/confluent/ksql/topic/TopicDeleteInjectorTest.java b/ksql-engine/src/test/java/io/confluent/ksql/topic/TopicDeleteInjectorTest.java
@@ -49,6 +49,8 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+
+import org.apache.kafka.common.errors.TopicAuthorizationException;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -219,6 +221,30 @@ public void shouldThrowExceptionIfOtherSourcesUsingTopic() {
     deleteInjector.inject(dropStatement);
   }
 
+  @Test
+  public void shouldThrowIfTopicDoesNotExist() {
+    // Given:
+    final String STREAM_1 = "stream1";
+    final DataSource<?> other1 = givenSource(STREAM_1, "topicName");
+    when(metaStore.getSource(STREAM_1)).thenAnswer(inv -> other1);
+    when(other1.getKafkaTopicName()).thenReturn("topicName");
+    final ConfiguredStatement<DropStream> dropStatement = givenStatement(
+        "DROP stream1 DELETE TOPIC;",
+        new DropStream(QualifiedName.of("stream1"),
+            true,
+            true)
+    );
+    doThrow(RuntimeException.class).when(topicClient).deleteTopics(ImmutableList.of("topicName"));
+
+    // Expect:
+    expectedException.expect(RuntimeException.class);
+    expectedException.expectMessage("" +
+        "Could not delete the corresponding kafka topic: topicName");
+
+    // When:
+    deleteInjector.inject(dropStatement);
+  }
+
   @Test
   public void shouldNotThrowIfSchemaIsMissing() throws IOException, RestClientException {
     // Given: