You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2018/03/27 13:56:18 UTC
[1/2] flink git commit: [FLINK-9067] [e2eTests] Fix test and simplify
code
Repository: flink
Updated Branches:
refs/heads/master c28096a32 -> e04a3bf77
[FLINK-9067] [e2eTests] Fix test and simplify code
This closes #5759.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e04a3bf7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e04a3bf7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e04a3bf7
Branch: refs/heads/master
Commit: e04a3bf7703544cf5012eb2f55626af8e0a2cb8c
Parents: 6c37b15
Author: Timo Walther <tw...@apache.org>
Authored: Tue Mar 27 14:20:47 2018 +0200
Committer: Timo Walther <tw...@apache.org>
Committed: Tue Mar 27 15:52:04 2018 +0200
----------------------------------------------------------------------
.../flink-stream-sql-test/pom.xml | 21 +----------------
.../flink/sql/tests/StreamSQLTestProgram.java | 12 +++++-----
.../test-scripts/test_streaming_sql.sh | 24 +++++++++++++++-----
3 files changed, 25 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e04a3bf7/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-stream-sql-test/pom.xml b/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
index 860432c..1b8e4f7 100644
--- a/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
+++ b/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
@@ -73,6 +73,7 @@
<goal>shade</goal>
</goals>
<configuration>
+ <finalName>StreamSQLTestProgram</finalName>
<artifactSet>
<excludes>
<exclude>com.google.code.findbugs:jsr305</exclude>
@@ -99,26 +100,6 @@
</execution>
</executions>
</plugin>
- <!--simplify the name of the testing JARs for referring to them in the end-to-end test scripts-->
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-antrun-plugin</artifactId>
- <version>1.7</version>
- <executions>
- <execution>
- <id>rename</id>
- <phase>package</phase>
- <goals>
- <goal>run</goal>
- </goals>
- <configuration>
- <target>
- <copy file="${project.basedir}/target/flink-stream-sql-test_${scala.binary.version}-${project.version}.jar" tofile="${project.basedir}/target/StreamSQLTestProgram.jar" />
- </target>
- </configuration>
- </execution>
- </executions>
- </plugin>
</plugins>
</build>
http://git-wip-us.apache.org/repos/asf/flink/blob/e04a3bf7/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java b/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
index 91fb4c1..e9e79ac 100644
--- a/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
+++ b/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
@@ -58,7 +58,7 @@ import java.util.Map;
*
* <p>The stream is bounded and will complete after about a minute.
* The result is always constant.
- * The job is killed on the first attemped and restarted.
+ * The job is killed on the first attempt and restarted.
*
* <p>Parameters:
* -outputPath Sets the path to where the result data is written.
@@ -249,12 +249,12 @@ public class StreamSQLTestProgram {
}
@Override
- public List<Long> snapshotState(long checkpointId, long timestamp) throws Exception {
+ public List<Long> snapshotState(long checkpointId, long timestamp) {
return Collections.singletonList(ms);
}
@Override
- public void restoreState(List<Long> state) throws Exception {
+ public void restoreState(List<Long> state) {
for (Long l : state) {
ms += l;
}
@@ -272,7 +272,7 @@ public class StreamSQLTestProgram {
private int lostRecordCnt = 0;
@Override
- public Row map(Row value) throws Exception {
+ public Row map(Row value) {
// the both counts are the same only in the first execution attempt
if (saveRecordCnt == 1 && lostRecordCnt == 1) {
@@ -294,12 +294,12 @@ public class StreamSQLTestProgram {
}
@Override
- public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
+ public List<Integer> snapshotState(long checkpointId, long timestamp) {
return Collections.singletonList(saveRecordCnt);
}
@Override
- public void restoreState(List<Integer> state) throws Exception {
+ public void restoreState(List<Integer> state) {
for (Integer i : state) {
saveRecordCnt += i;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e04a3bf7/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh b/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
index ed7b16c..21c64a1 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
@@ -31,13 +31,25 @@ $FLINK_DIR/bin/taskmanager.sh start
$FLINK_DIR/bin/flink run -p 4 $TEST_PROGRAM_JAR -outputPath $TEST_DATA_DIR/out/result
-stop_cluster
-$FLINK_DIR/bin/taskmanager.sh stop-all
+function sql_cleanup() {
-# remove flink-table from lib folder
-rm $FLINK_DIR/lib/flink-table*jar
+ stop_cluster
+ $FLINK_DIR/bin/taskmanager.sh stop-all
+
+ # remove flink-table from lib folder
+ rm $FLINK_DIR/lib/flink-table*jar
+
+ # make sure to run regular cleanup as well
+ cleanup
+}
+trap sql_cleanup INT
+trap sql_cleanup EXIT
# collect results from files
-cat /tmp/xxx/part-0-0 /tmp/xxx/_part-0-1.pending > $TEST_DATA_DIR/out/result-complete
-# check result
+cat $TEST_DATA_DIR/out/result/part-0-0 $TEST_DATA_DIR/out/result/_part-0-1.pending > $TEST_DATA_DIR/out/result-complete
+
+# check result:
+# 20,1970-01-01 00:00:00.0
+# 20,1970-01-01 00:00:20.0
+# 20,1970-01-01 00:00:40.0
check_result_hash "StreamSQL" $TEST_DATA_DIR/out/result-complete "b29f14ed221a936211202ff65b51ee26"
[2/2] flink git commit: [FLINK-9067] [e2eTests] Add
StreamSQLTestProgram and test run script.
Posted by tw...@apache.org.
[FLINK-9067] [e2eTests] Add StreamSQLTestProgram and test run script.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6c37b157
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6c37b157
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6c37b157
Branch: refs/heads/master
Commit: 6c37b15721af3a84fb3ac0de14486a9e98433215
Parents: c28096a
Author: Fabian Hueske <fh...@apache.org>
Authored: Fri Mar 23 21:42:53 2018 +0100
Committer: Timo Walther <tw...@apache.org>
Committed: Tue Mar 27 15:52:04 2018 +0200
----------------------------------------------------------------------
.../flink-stream-sql-test/pom.xml | 126 ++++++++
.../flink/sql/tests/StreamSQLTestProgram.java | 309 +++++++++++++++++++
flink-end-to-end-tests/pom.xml | 1 +
flink-end-to-end-tests/run-nightly-tests.sh | 8 +
.../test-scripts/test_streaming_sql.sh | 43 +++
5 files changed, 487 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6c37b157/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-stream-sql-test/pom.xml b/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
new file mode 100644
index 0000000..860432c
--- /dev/null
+++ b/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
@@ -0,0 +1,126 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <parent>
+ <artifactId>flink-end-to-end-tests</artifactId>
+ <groupId>org.apache.flink</groupId>
+ <version>1.6-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-stream-sql-test_${scala.binary.version}</artifactId>
+ <name>flink-stream-sql-test</name>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-filesystem_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>3.0.0</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <artifactSet>
+ <excludes>
+ <exclude>com.google.code.findbugs:jsr305</exclude>
+ <exclude>org.slf4j:*</exclude>
+ <exclude>log4j:*</exclude>
+ </excludes>
+ </artifactSet>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>org.apache.flink.sql.tests.StreamSQLTestProgram</mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <!--simplify the name of the testing JARs for referring to them in the end-to-end test scripts-->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-antrun-plugin</artifactId>
+ <version>1.7</version>
+ <executions>
+ <execution>
+ <id>rename</id>
+ <phase>package</phase>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ <configuration>
+ <target>
+ <copy file="${project.basedir}/target/flink-stream-sql-test_${scala.binary.version}-${project.version}.jar" tofile="${project.basedir}/target/StreamSQLTestProgram.jar" />
+ </target>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
+
http://git-wip-us.apache.org/repos/asf/flink/blob/6c37b157/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java b/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
new file mode 100644
index 0000000..91fb4c1
--- /dev/null
+++ b/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
@@ -0,0 +1,309 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.sql.tests;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.fs.bucketing.BasePathBucketer;
+import org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.sources.DefinedFieldMapping;
+import org.apache.flink.table.sources.DefinedRowtimeAttributes;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.BoundedOutOfOrderTimestamps;
+import org.apache.flink.types.Row;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * End-to-end test for Stream SQL queries.
+ *
+ * <p>Includes the following SQL features:
+ * - OVER window aggregation
+ * - keyed and non-keyed GROUP BY TUMBLE aggregation
+ * - windowed INNER JOIN
+ * - TableSource with event-time attribute
+ *
+ * <p>The stream is bounded and will complete after about a minute.
+ * The result is always constant.
+ * The job is killed on the first attemped and restarted.
+ *
+ * <p>Parameters:
+ * -outputPath Sets the path to where the result data is written.
+ */
+public class StreamSQLTestProgram {
+
+ public static void main(String[] args) throws Exception {
+
+ ParameterTool params = ParameterTool.fromArgs(args);
+ String outputPath = params.getRequired("outputPath");
+
+ StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ sEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+ sEnv.enableCheckpointing(4000);
+ sEnv.getConfig().setAutoWatermarkInterval(1000);
+
+ StreamTableEnvironment tEnv = TableEnvironment.getTableEnvironment(sEnv);
+
+ tEnv.registerTableSource("table1", new GeneratorTableSource(10, 100, 60, 0));
+ tEnv.registerTableSource("table2", new GeneratorTableSource(5, 0.2f, 60, 5));
+
+ int overWindowSizeSeconds = 1;
+ int tumbleWindowSizeSeconds = 10;
+
+ String overQuery = String.format(
+ "SELECT " +
+ " key, " +
+ " rowtime, " +
+ " COUNT(*) OVER (PARTITION BY key ORDER BY rowtime RANGE BETWEEN INTERVAL '%d' SECOND PRECEDING AND CURRENT ROW) AS cnt " +
+ "FROM table1",
+ overWindowSizeSeconds);
+
+ String tumbleQuery = String.format(
+ "SELECT " +
+ " key, " +
+ " CASE SUM(cnt) / COUNT(*) WHEN 101 THEN 1 ELSE 99 END AS correct, " +
+ " TUMBLE_START(rowtime, INTERVAL '%d' SECOND) AS wStart, " +
+ " TUMBLE_ROWTIME(rowtime, INTERVAL '%d' SECOND) AS rowtime " +
+ "FROM (%s) " +
+ "WHERE rowtime > TIMESTAMP '1970-01-01 00:00:01' " +
+ "GROUP BY key, TUMBLE(rowtime, INTERVAL '%d' SECOND)",
+ tumbleWindowSizeSeconds,
+ tumbleWindowSizeSeconds,
+ overQuery,
+ tumbleWindowSizeSeconds);
+
+ String joinQuery = String.format(
+ "SELECT " +
+ " t1.key, " +
+ " t2.rowtime AS rowtime, " +
+ " t2.correct," +
+ " t2.wStart " +
+ "FROM table2 t1, (%s) t2 " +
+ "WHERE " +
+ " t1.key = t2.key AND " +
+ " t1.rowtime BETWEEN t2.rowtime AND t2.rowtime + INTERVAL '%d' SECOND",
+ tumbleQuery,
+ tumbleWindowSizeSeconds);
+
+ String finalAgg = String.format(
+ "SELECT " +
+ " SUM(correct) AS correct, " +
+ " TUMBLE_START(rowtime, INTERVAL '20' SECOND) AS rowtime " +
+ "FROM (%s) " +
+ "GROUP BY TUMBLE(rowtime, INTERVAL '20' SECOND)",
+ joinQuery);
+
+ // get Table for SQL query
+ Table result = tEnv.sqlQuery(finalAgg);
+ // convert Table into append-only DataStream
+ DataStream<Row> resultStream =
+ tEnv.toAppendStream(result, Types.ROW(Types.INT, Types.SQL_TIMESTAMP));
+
+ // define bucketing sink to emit the result
+ BucketingSink<Row> sink = new BucketingSink<Row>(outputPath)
+ .setBucketer(new BasePathBucketer<>());
+
+ resultStream
+ // inject a KillMapper that forwards all records but terminates the first execution attempt
+ .map(new KillMapper()).setParallelism(1)
+ // add sink function
+ .addSink(sink).setParallelism(1);
+
+ sEnv.execute();
+ }
+
+ /**
+ * TableSource for generated data.
+ */
+ public static class GeneratorTableSource
+ implements StreamTableSource<Row>, DefinedRowtimeAttributes, DefinedFieldMapping {
+
+ private final int numKeys;
+ private final float recordsPerKeyAndSecond;
+ private final int durationSeconds;
+ private final int offsetSeconds;
+
+ public GeneratorTableSource(int numKeys, float recordsPerKeyAndSecond, int durationSeconds, int offsetSeconds) {
+ this.numKeys = numKeys;
+ this.recordsPerKeyAndSecond = recordsPerKeyAndSecond;
+ this.durationSeconds = durationSeconds;
+ this.offsetSeconds = offsetSeconds;
+ }
+
+ @Override
+ public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) {
+ return execEnv.addSource(new Generator(numKeys, recordsPerKeyAndSecond, durationSeconds, offsetSeconds));
+ }
+
+ @Override
+ public TypeInformation<Row> getReturnType() {
+ return Types.ROW(Types.INT, Types.LONG, Types.STRING);
+ }
+
+ @Override
+ public TableSchema getTableSchema() {
+ return new TableSchema(
+ new String[] {"key", "rowtime", "payload"},
+ new TypeInformation[] {Types.INT, Types.SQL_TIMESTAMP, Types.STRING});
+ }
+
+ @Override
+ public String explainSource() {
+ return "GeneratorTableSource";
+ }
+
+ @Override
+ public List<RowtimeAttributeDescriptor> getRowtimeAttributeDescriptors() {
+ return Collections.singletonList(
+ new RowtimeAttributeDescriptor(
+ "rowtime",
+ new ExistingField("ts"),
+ new BoundedOutOfOrderTimestamps(100)));
+ }
+
+ @Override
+ public Map<String, String> getFieldMapping() {
+ Map<String, String> mapping = new HashMap<>();
+ mapping.put("key", "f0");
+ mapping.put("ts", "f1");
+ mapping.put("payload", "f2");
+ return mapping;
+ }
+ }
+
+ /**
+ * Data-generating source function.
+ */
+ public static class Generator implements SourceFunction<Row>, ResultTypeQueryable<Row>, ListCheckpointed<Long> {
+
+ private final int numKeys;
+ private final int offsetSeconds;
+
+ private final int sleepMs;
+ private final int durationMs;
+
+ private long ms = 0;
+
+ public Generator(int numKeys, float rowsPerKeyAndSecond, int durationSeconds, int offsetSeconds) {
+ this.numKeys = numKeys;
+ this.durationMs = durationSeconds * 1000;
+ this.offsetSeconds = offsetSeconds;
+
+ this.sleepMs = (int) (1000 / rowsPerKeyAndSecond);
+ }
+
+ @Override
+ public void run(SourceContext<Row> ctx) throws Exception {
+ long offsetMS = offsetSeconds * 2000L;
+
+ while (ms < durationMs) {
+ synchronized (ctx.getCheckpointLock()) {
+ for (int i = 0; i < numKeys; i++) {
+ ctx.collect(Row.of(i, ms + offsetMS, "Some payload..."));
+ }
+ ms += sleepMs;
+ }
+ Thread.sleep(sleepMs);
+ }
+ }
+
+ @Override
+ public void cancel() { }
+
+ @Override
+ public TypeInformation<Row> getProducedType() {
+ return Types.ROW(Types.INT, Types.LONG, Types.STRING);
+ }
+
+ @Override
+ public List<Long> snapshotState(long checkpointId, long timestamp) throws Exception {
+ return Collections.singletonList(ms);
+ }
+
+ @Override
+ public void restoreState(List<Long> state) throws Exception {
+ for (Long l : state) {
+ ms += l;
+ }
+ }
+ }
+
+ /**
+ * Kills the first execution attempt of an application when it receives the second record.
+ */
+ public static class KillMapper implements MapFunction<Row, Row>, ListCheckpointed<Integer>, ResultTypeQueryable {
+
+ // counts all processed records of all previous execution attempts
+ private int saveRecordCnt = 0;
+ // counts all processed records of this execution attempt
+ private int lostRecordCnt = 0;
+
+ @Override
+ public Row map(Row value) throws Exception {
+
+ // the both counts are the same only in the first execution attempt
+ if (saveRecordCnt == 1 && lostRecordCnt == 1) {
+ throw new RuntimeException("Kill this Job!");
+ }
+
+ // update checkpointed counter
+ saveRecordCnt++;
+ // update non-checkpointed counter
+ lostRecordCnt++;
+
+ // forward record
+ return value;
+ }
+
+ @Override
+ public TypeInformation getProducedType() {
+ return Types.ROW(Types.INT, Types.SQL_TIMESTAMP);
+ }
+
+ @Override
+ public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
+ return Collections.singletonList(saveRecordCnt);
+ }
+
+ @Override
+ public void restoreState(List<Integer> state) throws Exception {
+ for (Integer i : state) {
+ saveRecordCnt += i;
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c37b157/flink-end-to-end-tests/pom.xml
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/pom.xml b/flink-end-to-end-tests/pom.xml
index f8913d4..35002dd 100644
--- a/flink-end-to-end-tests/pom.xml
+++ b/flink-end-to-end-tests/pom.xml
@@ -37,6 +37,7 @@ under the License.
<modules>
<module>flink-parent-child-classloading-test</module>
<module>flink-dataset-allround-test</module>
+ <module>flink-stream-sql-test</module>
</modules>
</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/6c37b157/flink-end-to-end-tests/run-nightly-tests.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh
index 71224e0..eca2e4d 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -55,5 +55,13 @@ if [ $EXIT_CODE == 0 ]; then
EXIT_CODE=$?
fi
+if [ $EXIT_CODE == 0 ]; then
+ printf "\n==============================================================================\n"
+ printf "Running Streaming SQL nightly end-to-end test\n"
+ printf "==============================================================================\n"
+ $END_TO_END_DIR/test-scripts/test_streaming_sql.sh
+ EXIT_CODE=$?
+fi
+
# Exit code for Travis build success/failure
exit $EXIT_CODE
http://git-wip-us.apache.org/repos/asf/flink/blob/6c37b157/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh b/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
new file mode 100755
index 0000000..ed7b16c
--- /dev/null
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
@@ -0,0 +1,43 @@
+#!/usr/bin/env bash
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+source "$(dirname "$0")"/common.sh
+
+TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-stream-sql-test/target/StreamSQLTestProgram.jar
+
+# copy flink-table jar into lib folder
+cp $FLINK_DIR/opt/flink-table*jar $FLINK_DIR/lib
+
+start_cluster
+$FLINK_DIR/bin/taskmanager.sh start
+$FLINK_DIR/bin/taskmanager.sh start
+$FLINK_DIR/bin/taskmanager.sh start
+
+$FLINK_DIR/bin/flink run -p 4 $TEST_PROGRAM_JAR -outputPath $TEST_DATA_DIR/out/result
+
+stop_cluster
+$FLINK_DIR/bin/taskmanager.sh stop-all
+
+# remove flink-table from lib folder
+rm $FLINK_DIR/lib/flink-table*jar
+
+# collect results from files
+cat /tmp/xxx/part-0-0 /tmp/xxx/_part-0-1.pending > $TEST_DATA_DIR/out/result-complete
+# check result
+check_result_hash "StreamSQL" $TEST_DATA_DIR/out/result-complete "b29f14ed221a936211202ff65b51ee26"