Add information in error message when kcql topic parsing fails

lensesio · Aug 26, 2019 · 51eeb8d · 51eeb8d
1 parent 64d1903
commit 51eeb8d
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 2 deletions.
diff --git a/src/main/scala/com/datamountaineer/streamreactor/connect/config/Helpers.scala b/src/main/scala/com/datamountaineer/streamreactor/connect/config/Helpers.scala
@@ -104,14 +104,14 @@ object Helpers extends StrictLogging {
 
     if (!res) {
       val missing = topics.diff(sources)
-      throw new ConfigException(s"Mandatory `topics` configuration contains topics not set in $kcqlConstant: ${missing}")
+      throw new ConfigException(s"Mandatory `topics` configuration contains topics not set in $kcqlConstant: ${missing}, kcql contains $sources")
     }
 
     val res1 = sources.subsetOf(topics)
 
     if (!res1) {
       val missing = topics.diff(sources)
-      throw new ConfigException(s"$kcqlConstant configuration contains topics not set in mandatory `topic` configuration: ${missing}")
+      throw new ConfigException(s"$kcqlConstant configuration contains topics not set in mandatory `topic` configuration: ${missing}, kcql contains $sources")
     }
 
     true

diff --git a/src/test/scala/com/datamountaineer/streamreactor/connect/config/TestHelpers.scala b/src/test/scala/com/datamountaineer/streamreactor/connect/config/TestHelpers.scala
@@ -56,4 +56,16 @@ class TestHelpers extends TestUtilsBase  {
     val res = Helpers.checkInputTopics(kcqlConstant, props)
     res shouldBe true
   }
+
+  "should add topics involved in kcql error to message" in {
+    val props = Map("topics" -> "topic1",
+      s"$kcqlConstant" -> "insert into table select time,c1,c2 from topic1 WITH TIMESTAMP time"
+    )
+
+    val e = intercept[ConfigException] {
+      Helpers.checkInputTopics(kcqlConstant, props)
+    }
+
+    e.getMessage.contains("topic1WITHTIMESTAMPtime") shouldBe true
+  }
 }