[FLINK-29282][tests] Decouple Quickstart E2E test from Elasticsearch

flink-end-to-end-tests/flink-quickstart-test-dummy-dependency/pom.xml

@@ -0,0 +1,33 @@
+<?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>
+		<groupId>org.apache.flink</groupId>
+		<artifactId>flink-end-to-end-tests</artifactId>
+		<version>1.17-SNAPSHOT</version>
+	</parent>
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>flink-quickstart-test-dummy-dependency</artifactId>
+	<name>Flink : E2E Tests : Quickstart : Dummy-Dependency</name>
+	<packaging>jar</packaging>
+</project>

flink-end-to-end-tests/flink-quickstart-test-dummy-dependency/src/main/java/org/apache/flink/quickstarts/test/utils/Utils.java

@@ -0,0 +1,26 @@
+/*
+ * 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.quickstarts.test.utils;
+
+/** Dummy util to test packaging of Flink dependencies in quickstarts. */
+public class Utils {
+    public static String prefix(Long message) {
+        return "message #" + message;
+    }
+}

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

@@ -56,7 +56,7 @@ under the License.
 		</dependency>
 		<dependency>
 			<groupId>org.apache.flink</groupId>
-			<artifactId>flink-connector-elasticsearch7</artifactId>
+			<artifactId>flink-quickstart-test-dummy-dependency</artifactId>
 			<version>${project.version}</version>
 		</dependency>
 	</dependencies>

flink-end-to-end-tests/flink-quickstart-test/src/main/java/org/apache/flink/quickstarts/test/QuickstartExample.java

@@ -0,0 +1,65 @@
+/*
+ * 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.quickstarts.test;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.quickstarts.test.utils.Utils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.CloseableIterator;
+
+/** End to end test for quickstarts. */
+public class QuickstartExample {
+
+    public static void main(String[] args) throws Exception {
+
+        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
+
+        if (parameterTool.getNumberOfParameters() < 1) {
+            System.out.println("Missing parameters!\nUsage: --numRecords <numRecords>");
+            return;
+        }
+
+        int numRecordsToEmit = parameterTool.getInt("numRecords");
+
+        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        env.enableCheckpointing(5000);
+
+        DataStream<String> source =
+                env.fromSequence(0, numRecordsToEmit - 1)
+                        .map((MapFunction<Long, String>) Utils::prefix);
+
+        try (CloseableIterator<String> data = source.collectAsync()) {
+            env.execute("Quickstart example");
+
+            int count = 0;
+            while (data.hasNext()) {
+                data.next();
+                count++;
+            }
+            if (count != numRecordsToEmit) {
+                throw new RuntimeException(
+                        String.format(
+                                "Unexpected number of records; expected :%s actual: %s",
+                                numRecordsToEmit, count));
+            }
+        }
+    }
+}

flink-end-to-end-tests/flink-quickstart-test/src/main/scala/org/apache/flink/quickstarts/test/QuickstartExample.scala

@@ -0,0 +1,58 @@
+/*
+ * 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.quickstarts.test
+
+import org.apache.flink.api.java.utils.ParameterTool
+import org.apache.flink.quickstarts.test.utils.Utils
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
+
+object QuickstartExample {
+  def main(args: Array[String]) {
+
+    val parameterTool = ParameterTool.fromArgs(args)
+
+    if (parameterTool.getNumberOfParameters < 1) {
+      println("Missing parameters!\nUsage: --numRecords <numRecords>")
+      return
+    }
+
+    val numRecordsToEmit = parameterTool.getInt("numRecords")
+
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    env.enableCheckpointing(5000)
+
+    val source: DataStream[(String)] = env
+      .fromSequence(0, numRecordsToEmit - 1)
+      .map(v => Utils.prefix(v))
+
+    val data = source.collectAsync()
+
+    env.execute("Elasticsearch7.x end to end sink test example")
+
+    var count = 0
+    while (data.hasNext) {
+      data.next()
+      count += 1
+    }
+    if (count != numRecordsToEmit) {
+      throw new RuntimeException(
+        s"Unexpected number of records; expected :$numRecordsToEmit actual: $count")
+    }
+  }
+}

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

@@ -53,6 +53,7 @@ under the License.
 		<module>flink-queryable-state-test</module>
 		<module>flink-local-recovery-and-allocation-test</module>
 		<module>flink-quickstart-test</module>
+		<module>flink-quickstart-test-dummy-dependency</module>
 		<module>flink-confluent-schema-registry</module>
 		<module>flink-stream-state-ttl-test</module>
 		<module>flink-sql-client-test</module>