Add runtime failing test jobs (#87)

Author: ferenc-csaky

.github/workflows/build.yml

@@ -117,6 +117,8 @@ jobs:
               compile -c aggregation-query-test-package.json
               compile -c aggregation-query-test-iceberg-package.json
               compile -c aggregation-query-subscription-test-package.json
+              compile -c runtime-failing-connector-package.json
+              compile -c runtime-failing-udf-package.json
 
           - example: udf-jbang
             path: user-defined-function/jbang

test-jobs/runtime-failing-connector-package.json

@@ -0,0 +1,7 @@
+{
+  "version": "1",
+  "enabled-engines": ["flink"],
+  "script": {
+    "main": "runtime-failing-connector.sqrl"
+  }
+}

test-jobs/runtime-failing-connector.sqrl

@@ -0,0 +1,20 @@
+CREATE TABLE _InputData (
+  uniqueId STRING NOT NULL,
+  measure DOUBLE NOT NULL,
+  partitionId STRING NOT NULL,
+  event_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
+  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
+) WITH (
+  'connector' = 'kafka',
+  'topic' = 'failing-test-source',
+  'format' = 'json',
+  'properties.bootstrap.servers' = 'non-existent-broker:9092',
+  'properties.request.timeout.ms' = '5000', -- Fail after 5sec
+  'properties.retry.backoff.ms' = '500',
+  'properties.metadata.max.age.ms' = '1000',
+  'scan.startup.mode' = 'earliest-offset'
+);
+
+_Res := SELECT * FROM _InputData;
+
+EXPORT _Res TO logger.res;

test-jobs/runtime-failing-udf-package.json

@@ -0,0 +1,7 @@
+{
+  "version": "1",
+  "enabled-engines": ["flink"],
+  "script": {
+    "main": "runtime-failing-udf.sqrl"
+  }
+}

test-jobs/runtime-failing-udf.sqrl

@@ -0,0 +1,5 @@
+IMPORT udfs.FailingFunction;
+
+Res := SELECT FailingFunction('dummy text') AS text;
+
+EXPORT Res TO logger.res;

test-jobs/udfs/FailingFunction.java

@@ -0,0 +1,16 @@
+//DEPS org.apache.flink:flink-table-common:1.19.3
+
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.ScalarFunction;
+
+public class FailingFunction extends ScalarFunction {
+
+  @Override
+  public void open(FunctionContext ctx) {
+    throw new RuntimeException("UDF init failed");
+  }
+
+  public String eval(String s) {
+    return s;
+  }
+}