fix: do not allow WITH clause to be created with invalid WINDOW_SIZE (#…

…3432)
confluentinc · Sep 27, 2019 · 96bfc11 · 96bfc11
1 parent e09d6ad
commit 96bfc11
Show file tree

Hide file tree

Showing 5 changed files with 23 additions and 0 deletions.
diff --git a/...confluent/ksql/parser/DurationParser.java → ...confluent/ksql/parser/DurationParser.java b/...confluent/ksql/parser/DurationParser.java → ...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
@@ -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/...luent/ksql/parser/DurationParserTest.java → ...luent/ksql/parser/DurationParserTest.java b/...luent/ksql/parser/DurationParserTest.java → ...luent/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
@@ -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'"
+      }
     }
   ]
 }
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
@@ -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."
+      }
     }
   ]
 }