[FLINK-8985][e2etest] initial support for End-to-end CLI test, excluding YARN test

flink-end-to-end-tests/flink-api-test/pom.xml

@@ -0,0 +1,105 @@
+<?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-api-test</artifactId>
+	<name>flink-api-test</name>
+	<packaging>jar</packaging>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-core</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-filesystem_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<type>test-jar</type>
+		</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>
+							<finalName>PeriodicStreamingJob</finalName>
+							<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.runtime.tests.PeriodicStreamingJob</mainClass>
+								</transformer>
+							</transformers>
+						</configuration>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
+	</build>
+
+</project>
+

flink-end-to-end-tests/flink-api-test/src/main/java/org/apache/flink/runtime/tests/PeriodicStreamingJob.java

@@ -0,0 +1,127 @@
+/*
+ * 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.runtime.tests;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.core.fs.FileSystem;
+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.api.windowing.time.Time;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * This is a periodic streaming job that runs for API testing purpose.
+ *
+ * <p>The stream is bounded and will complete after the specified duration.
+ *
+ * <p>Parameters:
+ * -outputPath Sets the path to where the result data is written.
+ * -recordsPerSecond Sets the output record frequency.
+ * -durationInSecond Sets the running duration of the job.
+ * -offsetInSecond Sets the startup delay before the processing first message.
+ */
+public class PeriodicStreamingJob {
+
+	public static void main(String[] args) throws Exception {
+		ParameterTool params = ParameterTool.fromArgs(args);
+		String outputPath = params.getRequired("outputPath");
+		int recordsPerSecond = params.getInt("recordsPerSecond", 10);
+		int duration = params.getInt("durationInSecond", 60);
+		int offset = params.getInt("offsetInSecond", 0);
+
+		StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+		sEnv.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
+		sEnv.enableCheckpointing(4000);
+		sEnv.getConfig().setAutoWatermarkInterval(1000);
+
+		// execute a simple pass through program.
+		PeriodicSourceGenerator generator = new PeriodicSourceGenerator(
+			recordsPerSecond, duration, offset);
+		DataStream rows = sEnv.addSource(generator);
+
+		DataStream result = rows
+			.keyBy(1)
+			.timeWindow(Time.seconds(5))
+			.sum(0);
+
+		result.writeAsText(outputPath + "/result.txt", FileSystem.WriteMode.OVERWRITE)
+			.setParallelism(1);
+
+		sEnv.execute();
+	}
+
+	/**
+	 * Data-generating source function.
+	 */
+	public static class PeriodicSourceGenerator implements SourceFunction<Tuple>, ResultTypeQueryable<Tuple>, ListCheckpointed<Long> {
+		private final int sleepMs;
+		private final int durationMs;
+		private final int offsetSeconds;
+		private long ms = 0;
+
+		public PeriodicSourceGenerator(float rowsPerSecond, int durationSeconds, int offsetSeconds) {
+			this.durationMs = durationSeconds * 1000;
+			this.sleepMs = (int) (1000 / rowsPerSecond);
+			this.offsetSeconds = offsetSeconds;
+		}
+
+		@Override
+		public void run(SourceContext<Tuple> ctx) throws Exception {
+			long offsetMs = offsetSeconds * 1000L;
+
+			while (ms < durationMs) {
+				synchronized (ctx.getCheckpointLock()) {
+					ctx.collect(new Tuple2<>(ms + offsetMs, "key"));
+				}
+				ms += sleepMs;
+				Thread.sleep(sleepMs);
+			}
+		}
+
+		@Override
+		public void cancel() { }
+
+		@Override
+		public TypeInformation<Tuple> getProducedType() {
+			return Types.TUPLE(Types.LONG, Types.STRING);
+		}
+
+		@Override
+		public List<Long> snapshotState(long checkpointId, long timestamp) {
+			return Collections.singletonList(ms);
+		}
+
+		@Override
+		public void restoreState(List<Long> state) {
+			for (Long l : state) {
+				ms += l;
+			}
+		}
+	}
+}

flink-end-to-end-tests/pom.xml

@@ -35,6 +35,7 @@ under the License.
 	<name>flink-end-to-end-tests</name>
 
 	<modules>
+		<module>flink-api-test</module>
 		<module>flink-parent-child-classloading-test</module>
 		<module>flink-dataset-allround-test</module>
 		<module>flink-datastream-allround-test</module>