From 96bfc11ae69edf17a9473f364cf0c424ead60499 Mon Sep 17 00:00:00 2001
From: Andy Coates <8012398+big-andy-coates@users.noreply.github.com>
Date: Fri, 27 Sep 2019 19:01:59 +0100
Subject: [PATCH] fix: do not allow WITH clause to be created with invalid
 WINDOW_SIZE (#3432)

---
 .../java/io/confluent/ksql/parser/DurationParser.java  |  0
 .../confluent/ksql/properties/with/CreateConfigs.java  |  3 +++
 .../io/confluent/ksql/parser/DurationParserTest.java   |  0
 .../query-validation-tests/hopping-windows.json        | 10 ++++++++++
 .../query-validation-tests/session-windows.json        | 10 ++++++++++
 5 files changed, 23 insertions(+)
 rename {ksql-parser => ksql-common}/src/main/java/io/confluent/ksql/parser/DurationParser.java (100%)
 rename {ksql-parser => ksql-common}/src/test/java/io/confluent/ksql/parser/DurationParserTest.java (100%)

diff --git a/ksql-parser/src/main/java/io/confluent/ksql/parser/DurationParser.java b/ksql-common/src/main/java/io/confluent/ksql/parser/DurationParser.java
similarity index 100%
rename from ksql-parser/src/main/java/io/confluent/ksql/parser/DurationParser.java
rename to ksql-common/src/main/java/io/confluent/ksql/parser/DurationParser.java
diff --git a/ksql-common/src/main/java/io/confluent/ksql/properties/with/CreateConfigs.java b/ksql-common/src/main/java/io/confluent/ksql/properties/with/CreateConfigs.java
index 53f8b575feb9..a2e6836c7364 100644
--- a/ksql-common/src/main/java/io/confluent/ksql/properties/with/CreateConfigs.java
+++ b/ksql-common/src/main/java/io/confluent/ksql/properties/with/CreateConfigs.java
@@ -17,7 +17,9 @@
 
 import static io.confluent.ksql.configdef.ConfigValidators.enumValues;
 
+import io.confluent.ksql.configdef.ConfigValidators;
 import io.confluent.ksql.model.WindowType;
+import io.confluent.ksql.parser.DurationParser;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
@@ -54,6 +56,7 @@ public final class CreateConfigs {
           WINDOW_SIZE_PROPERTY,
           Type.STRING,
           null,
+          ConfigValidators.nullsAllowed(ConfigValidators.parses(DurationParser::parse)),
           Importance.LOW,
           "If the data is windowed, i.e., was created using KSQL via a query that "
               + "contains a ``WINDOW`` clause and the window is a HOPPING or TUMBLING window, "
diff --git a/ksql-parser/src/test/java/io/confluent/ksql/parser/DurationParserTest.java b/ksql-common/src/test/java/io/confluent/ksql/parser/DurationParserTest.java
similarity index 100%
rename from ksql-parser/src/test/java/io/confluent/ksql/parser/DurationParserTest.java
rename to ksql-common/src/test/java/io/confluent/ksql/parser/DurationParserTest.java
diff --git a/ksql-functional-tests/src/test/resources/query-validation-tests/hopping-windows.json b/ksql-functional-tests/src/test/resources/query-validation-tests/hopping-windows.json
index 6bdb4644013e..429c10e42943 100644
--- a/ksql-functional-tests/src/test/resources/query-validation-tests/hopping-windows.json
+++ b/ksql-functional-tests/src/test/resources/query-validation-tests/hopping-windows.json
@@ -153,6 +153,16 @@
         {"topic": "S2", "key": 100, "value": "100,100,100 : Window{start=20000 end=-}", "timestamp": 30000, "window": {"start": 20000, "end": 50000, "type": "time"}},
         {"topic": "S2", "key": 100, "value": "100,100,100 : Window{start=30000 end=-}", "timestamp": 30000, "window": {"start": 30000, "end": 60000, "type": "time"}}
       ]
+    },
+    {
+      "name": "import table with invalid window size",
+      "statements": [
+        "CREATE TABLE TEST (ID bigint, VALUE bigint) WITH (kafka_topic='test_topic', value_format='DELIMITED', key='ID', WINDOW_TYPE='Hopping', WINDOW_SIZE='30 bobs');"
+      ],
+      "expectedException": {
+        "type": "io.confluent.ksql.parser.exception.ParseFailedException",
+        "message": "Configuration WINDOW_SIZE is invalid: Invalid duration: '30 bobs'. Unknown time unit: 'BOBS'"
+      }
     }
   ]
 }
\ No newline at end of file
diff --git a/ksql-functional-tests/src/test/resources/query-validation-tests/session-windows.json b/ksql-functional-tests/src/test/resources/query-validation-tests/session-windows.json
index 8192aabd7db8..db682d7b4c87 100644
--- a/ksql-functional-tests/src/test/resources/query-validation-tests/session-windows.json
+++ b/ksql-functional-tests/src/test/resources/query-validation-tests/session-windows.json
@@ -134,6 +134,16 @@
           }
         ]
       }
+    },
+    {
+      "name": "import table with invalid window size",
+      "statements": [
+        "CREATE TABLE TEST (ID bigint, VALUE bigint) WITH (kafka_topic='test_topic', value_format='DELIMITED', key='ID', WINDOW_TYPE='Session', WINDOW_SIZE='30 seconds');"
+      ],
+      "expectedException": {
+        "type": "io.confluent.ksql.parser.exception.ParseFailedException",
+        "message": "'WINDOW_SIZE' should not be set for SESSION windows."
+      }
     }
   ]
 }
\ No newline at end of file