You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by zhangminglei <gi...@git.apache.org> on 2018/04/05 14:30:02 UTC
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
GitHub user zhangminglei opened a pull request:
https://github.com/apache/flink/pull/5823
[FLINK-9008] [e2e] Implements quickstarts end to end test
## What is the purpose of the change
Add an end-to-end test which verifies Flink's quickstarts.
## Brief change log
1. Create a new Flink project using the quickstarts archetype []
2. Add a new Flink dependency to the pom.xml (Flink elasticsearch connector)
3. Add a scripts verify that program
## Verifying this change
Run test_quickstarts.sh verify this change
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no )
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not documented)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zhangminglei/flink flink-9008
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5823.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5823
----
commit c7b86bb18358715c747e0fa12946b3cfaff86a4d
Author: zhangminglei <zm...@...>
Date: 2018-04-05T14:20:17Z
[FLINK-9008] [e2e] Implements quickstarts end to end test
----
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on the issue:
https://github.com/apache/flink/pull/5823
Thank you very much to @tzulitai and @zentol review this PR.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188664926
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
--- End diff --
Will do.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r183698160
--- Diff: flink-end-to-end-tests/run-pre-commit-tests.sh ---
@@ -53,6 +53,14 @@ if [ $EXIT_CODE == 0 ]; then
EXIT_CODE=$?
fi
+if [ $EXIT_CODE == 0 ]; then
+ printf "\n==============================================================================\n"
+ printf "Running Quickstarts nightly end-to-end test\n"
--- End diff --
this test should only be run nightly, i.e. must not show up in this file.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r181699430
--- Diff: flink-end-to-end-tests/test-scripts/test-class/ElasticsearchStreamingJob.java ---
@@ -0,0 +1,97 @@
+/*
+ * 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.quickstart;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
+
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.Requests;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Elasticsearch example for Flink Streaming Job.
+ *
+ * <p>In this streaming job, we generate a bunch of data from numbers, apply two operators map
+ * and filter to that data. Then we choose elasticsearch as its sink to storage these data.
+ *
+ * <p>Run test_quickstarts.sh to verify this program. Package this class to a jar, verify the jar,
+ * then deploy it on a flink cluster.
+ */
+public class ElasticsearchStreamingJob {
+
+ public static void main(String[] args) throws Exception {
+ // set up the streaming execution environment
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<String> source = env.generateSequence(0, 20)
+ // map the data from 1,2,3... to the form of message #1, message #2, message #3...
+ .map(new MapFunction<Long, String>() {
--- End diff --
these operations aren't necessary. The only interesting part is that the sink is actually working.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188196219
--- Diff: flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala ---
@@ -34,7 +34,6 @@ import org.apache.flink.runtime.instance.InstanceManager
import org.apache.flink.runtime.jobmanager.scheduler.{Scheduler => FlinkScheduler}
import org.apache.flink.runtime.jobmanager.{JobManager, SubmittedJobGraphStore}
import org.apache.flink.runtime.leaderelection.LeaderElectionService
-import org.apache.flink.runtime.metrics.MetricRegistryImpl
--- End diff --
Is this irrelevant? Would prefer this as a separate hotfix.
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on the issue:
https://github.com/apache/flink/pull/5823
Hi, @zentol I have updated the code and the patch is available to have another review now.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/5823
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188194072
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
--- End diff --
I would suggest to generate the mvn project under `$TEST_DATA_DIR`. Then, it will be properly cleaned up after the test completes.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188645244
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/../flink-quickstart-test/src/main/java/org/apache/flink/quickstart/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
--- End diff --
Will change.
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on the issue:
https://github.com/apache/flink/pull/5823
Hello, @zentol. I have updated the code. CI run without error. Could you please take another look ?
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188486423
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/../flink-quickstart-test/src/main/java/org/apache/flink/quickstart/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package -nsu
+
+cd target
+jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+
+if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Success: There are no flink core classes are contained in the jar."
+else
+ echo "Failure: There are flink core classes are contained in the jar."
+ exit 1
+fi
+
+if [[ `grep -c "org/apache/flink/quickstart/StreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/quickstart/ElasticsearchStreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/connectors/elasticsearch2" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Failure: Since ElasticsearchStreamingJob.class and other user classes are not included in the jar. "
+ exit 1
+else
+ echo "Success: ElasticsearchStreamingJob.class and other user classes are included in the jar."
+fi
+
+cd $CURRENT_DIR
+
+source "$(dirname "$0")"/common.sh
+
+start_cluster
+
+mkdir -p $TEST_DATA_DIR
+
+ELASTICSEARCH_URL="https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.5/elasticsearch-2.3.5.tar.gz"
+
+curl "$ELASTICSEARCH_URL" > $TEST_DATA_DIR/elasticsearch.tar.gz
+tar xzf $TEST_DATA_DIR/elasticsearch.tar.gz -C $TEST_DATA_DIR/
+ELASTICSEARCH_DIR=$TEST_DATA_DIR/elasticsearch-2.3.5
+
+# start elasticsearch cluster
+nohup $ELASTICSEARCH_DIR/bin/elasticsearch &
+
+TEST_PROGRAM_JAR=$CURRENT_DIR/flink-java-project/target/flink-java-project-0.1.jar
+
+# run the Flink job
--- End diff --
Will do.
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on the issue:
https://github.com/apache/flink/pull/5823
Thank you very much @zentol . I have updated the PR. Not sure whether to meet the demand 100%. For verify the job was successfully run, I used Elasticsearch2 as a sink. As well add a elasticsearch2 dependency to archetype pom.xml. Could you take another look ?
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r183970394
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/test-class/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package
+
+cd target
+jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+
+if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Success: There are no flink core classes are contained in the jar."
+else
+ echo "Failure: There are flink core classes are contained in the jar."
+ exit 1
+fi
+
+if [[ `grep -c "org/apache/flink/quickstart/StreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/quickstart/ElasticsearchStreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/connectors/elasticsearch2" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Failure: Since ElasticsearchStreamingJob.class and other user classes are not included in the jar. "
+ exit 1
+else
+ echo "Success: ElasticsearchStreamingJob.class and other user classes are included in the jar."
+fi
+
+cd $CURRENT_DIR
+
+source "$(dirname "$0")"/common.sh
+
+start_cluster
+
+mkdir -p $TEST_DATA_DIR
+
+ELASTICSEARCH_URL="https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.5/elasticsearch-2.3.5.tar.gz"
+
+curl "$ELASTICSEARCH_URL" > $TEST_DATA_DIR/elasticsearch.tar.gz
+tar xzf $TEST_DATA_DIR/elasticsearch.tar.gz -C $TEST_DATA_DIR/
+ELASTICSEARCH_DIR=$TEST_DATA_DIR/elasticsearch-2.3.5
+
+# start elasticsearch cluster
+nohup $ELASTICSEARCH_DIR/bin/elasticsearch &
--- End diff --
what happens if the ports 9200/9300 are already taken?
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188638639
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
--- End diff --
Yes. will do.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r183970197
--- Diff: flink-end-to-end-tests/test-scripts/test-class/ElasticsearchStreamingJob.java ---
@@ -0,0 +1,90 @@
+/*
+ * 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.quickstart;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
+
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.Requests;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Elasticsearch example for Flink Streaming Job.
+ *
+ * <p>In this streaming job, we generate a bunch of data from numbers, apply operator map
+ * made a type conversion. Then we choose elasticsearch as its sink to storage these data.
+ *
+ * <p>Run test_quickstarts.sh to verify this program. Package this class to a jar, verify the jar,
+ * then deploy it on a flink cluster.
+ */
+public class ElasticsearchStreamingJob {
--- End diff --
Let's move this class into a separate `flink-quickstart-test` module. It currently is not part of any module's source. The module should also depend on the ES connector.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r183753921
--- Diff: flink-end-to-end-tests/test-scripts/test-class/ElasticsearchStreamingJob.java ---
@@ -0,0 +1,97 @@
+/*
+ * 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.quickstart;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
+
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.Requests;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Elasticsearch example for Flink Streaming Job.
+ *
+ * <p>In this streaming job, we generate a bunch of data from numbers, apply two operators map
+ * and filter to that data. Then we choose elasticsearch as its sink to storage these data.
+ *
+ * <p>Run test_quickstarts.sh to verify this program. Package this class to a jar, verify the jar,
+ * then deploy it on a flink cluster.
+ */
+public class ElasticsearchStreamingJob {
+
+ public static void main(String[] args) throws Exception {
+ // set up the streaming execution environment
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<String> source = env.generateSequence(0, 20)
+ // map the data from 1,2,3... to the form of message #1, message #2, message #3...
+ .map(new MapFunction<Long, String>() {
+ @Override
+ public String map(Long value) throws Exception {
+ return "message #" + value;
+ }})
+ // filter out the data that contains message #11 and message #17
+ .filter(new FilterFunction<String>() {
+ @Override
+ public boolean filter(String value) throws Exception {
+ return !value.equals("message #11") && !value.equals("message #17");
+ }
+ });
+
+ Map<String, String> userConfig = new HashMap<>();
+ userConfig.put("cluster.name", "elasticsearch");
+ // This instructs the sink to emit after every element, otherwise they would be buffered
--- End diff --
The code just force write the data to Elasticsearch once there is one data was generated. The default number of actions to flush a bulk is 1000. If we do not set it to 1 for flush, then we can not see the data to sink.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r183742072
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.4.2 \
--- End diff --
will update.
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by medcv <gi...@git.apache.org>.
Github user medcv commented on the issue:
https://github.com/apache/flink/pull/5823
@tzulitai I agree it should be separate of this PR.
Main challenge is to upload a compiled Quickstart zip file contain all dependencies for each changes (minor) we do automatically. I was thinking after this PR got merged we can re-use this e2e test to build a Quickstart .zip file ready to upload on each merged PR.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r183741490
--- Diff: flink-end-to-end-tests/run-pre-commit-tests.sh ---
@@ -53,6 +53,14 @@ if [ $EXIT_CODE == 0 ]; then
EXIT_CODE=$?
fi
+if [ $EXIT_CODE == 0 ]; then
+ printf "\n==============================================================================\n"
+ printf "Running Quickstarts nightly end-to-end test\n"
--- End diff --
Yes. I also feel strange put this in this file, but I watched the same like ```test_streaming_kafka010.sh``` in this file. And this confused me. I do not think ```test_streaming_kafka010.sh``` should put here since ```kafka-common.sh``` will execute a download also.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188195999
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/../flink-quickstart-test/src/main/java/org/apache/flink/quickstart/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package -nsu
+
+cd target
+jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+
+if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Success: There are no flink core classes are contained in the jar."
+else
+ echo "Failure: There are flink core classes are contained in the jar."
+ exit 1
+fi
+
+if [[ `grep -c "org/apache/flink/quickstart/StreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/quickstart/ElasticsearchStreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/connectors/elasticsearch2" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Failure: Since ElasticsearchStreamingJob.class and other user classes are not included in the jar. "
+ exit 1
+else
+ echo "Success: ElasticsearchStreamingJob.class and other user classes are included in the jar."
+fi
+
+cd $CURRENT_DIR
+
+source "$(dirname "$0")"/common.sh
+
+start_cluster
+
+mkdir -p $TEST_DATA_DIR
+
+ELASTICSEARCH_URL="https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.5/elasticsearch-2.3.5.tar.gz"
+
+curl "$ELASTICSEARCH_URL" > $TEST_DATA_DIR/elasticsearch.tar.gz
+tar xzf $TEST_DATA_DIR/elasticsearch.tar.gz -C $TEST_DATA_DIR/
+ELASTICSEARCH_DIR=$TEST_DATA_DIR/elasticsearch-2.3.5
+
+# start elasticsearch cluster
+nohup $ELASTICSEARCH_DIR/bin/elasticsearch &
+
+TEST_PROGRAM_JAR=$CURRENT_DIR/flink-java-project/target/flink-java-project-0.1.jar
+
+# run the Flink job
+$FLINK_DIR/bin/flink run -c org.apache.flink.quickstart.ElasticsearchStreamingJob $TEST_PROGRAM_JAR
+
+touch $TEST_DATA_DIR/output
+
+curl 'localhost:9200/my-index/_search?q=*&pretty&size=21' > $TEST_DATA_DIR/output
+
+if [ -n "$(grep '\"total\" : 21' $TEST_DATA_DIR/output)" ]; then
+ echo "Quickstarts end to end test pass."
+else
+ echo "Quickstarts end to end test failed."
+ exit 1
--- End diff --
`PASS=""`
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/5823
Thanks @zhangminglei. This looks good to merge, will proceed to merge this.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r183745076
--- Diff: flink-end-to-end-tests/test-scripts/test-class/ElasticsearchStreamingJob.java ---
@@ -0,0 +1,97 @@
+/*
+ * 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.quickstart;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
+
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.Requests;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Elasticsearch example for Flink Streaming Job.
+ *
+ * <p>In this streaming job, we generate a bunch of data from numbers, apply two operators map
+ * and filter to that data. Then we choose elasticsearch as its sink to storage these data.
+ *
+ * <p>Run test_quickstarts.sh to verify this program. Package this class to a jar, verify the jar,
+ * then deploy it on a flink cluster.
+ */
+public class ElasticsearchStreamingJob {
+
+ public static void main(String[] args) throws Exception {
+ // set up the streaming execution environment
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<String> source = env.generateSequence(0, 20)
+ // map the data from 1,2,3... to the form of message #1, message #2, message #3...
+ .map(new MapFunction<Long, String>() {
+ @Override
+ public String map(Long value) throws Exception {
+ return "message #" + value;
+ }})
+ // filter out the data that contains message #11 and message #17
+ .filter(new FilterFunction<String>() {
+ @Override
+ public boolean filter(String value) throws Exception {
+ return !value.equals("message #11") && !value.equals("message #17");
+ }
+ });
+
+ Map<String, String> userConfig = new HashMap<>();
+ userConfig.put("cluster.name", "elasticsearch");
+ // This instructs the sink to emit after every element, otherwise they would be buffered
+ userConfig.put(ElasticsearchSink.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, "1");
+
+ List<InetSocketAddress> transports = new ArrayList<>();
+ transports.add(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9300));
+
+ source.addSink(new ElasticsearchSink<>(userConfig, transports, new org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction<String>(){
--- End diff --
Okay, Sir ~
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/5823
1. You should create a project _within the test_. This project must **not** be based on `https://flink.apache.org/q/quickstart.sh`, is has to be created with the archetype that is installed locally by `flink-quickstarts`. For this you will have to use the maven commands as outlined in the [documentation](https://ci.apache.org/projects/flink/flink-docs-master/quickstart/java_api_quickstart.html).
2. Correct, you should verify that none of the core flink classes are contained in the jar. The only classes that should be contained are those of the project.
3. No, this is not related to checkpointing. The point of this test is to ensure that job-jars created by a quickstart project actually work when submitted to a flink cluster. For example, they have to contain the job classes, like `StreamingJob`, as otherwise the job will fail outright since there's nothing to run. So you have to verify that the job was successfully run, the easiest way being to write some data to some file (like the WordCount tests), and verifying the contents.
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on the issue:
https://github.com/apache/flink/pull/5823
Hi, @tzulitai I have updated the code according to your suggestions. It is the time for another review, could you take a look ? Thank you very much.
And the below is the test result on my local machine. Tests looks good.
```
Elasticsearch node is running.
Starting cluster.
Starting standalonesession daemon on host zhangmingleideMacBook-Pro.local.
Starting taskexecutor daemon on host zhangmingleideMacBook-Pro.local.
Dispatcher/TaskManagers are not yet up
Waiting for dispatcher REST endpoint to come up...
Dispatcher/TaskManagers are not yet up
Waiting for dispatcher REST endpoint to come up...
Dispatcher/TaskManagers are not yet up
Waiting for dispatcher REST endpoint to come up...
Waiting for dispatcher REST endpoint to come up...
Dispatcher REST endpoint is up.
Starting execution of program
Program execution finished
Job with JobID b527720074fcb49665ab03216abbcd5b has finished.
Job Runtime: 1843 ms
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
0 0 0 0 0 0 0 0 --:--:-- --:--:-- --:100 3470 100 3470 0 0 37739 0 --:--:-- --:--:-- --:--:-- 38131
Quickstarts end to end test pass.
Stopping taskexecutor daemon (pid: 67722) on host zhangmingleideMacBook-Pro.local.
Stopping standalonesession daemon (pid: 67428) on host zhangmingleideMacBook-Pro.local.
No zookeeper daemon to stop on host zhangmingleideMacBook-Pro.local.
All tests PASS
```
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r184013633
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/test-class/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package
--- End diff --
add `-nsu` options to prevent the download of snapshot versions. This will force maven to use the dependencies that are installed in the local repository.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188486521
--- Diff: flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala ---
@@ -34,7 +34,6 @@ import org.apache.flink.runtime.instance.InstanceManager
import org.apache.flink.runtime.jobmanager.scheduler.{Scheduler => FlinkScheduler}
import org.apache.flink.runtime.jobmanager.{JobManager, SubmittedJobGraphStore}
import org.apache.flink.runtime.leaderelection.LeaderElectionService
-import org.apache.flink.runtime.metrics.MetricRegistryImpl
--- End diff --
It already separate a hotfix. We have two commits here.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188195349
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/../flink-quickstart-test/src/main/java/org/apache/flink/quickstart/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
--- End diff --
The `$CURRENT_DIR` can be anywhere, so this path is basically invalid.
In the e2e test scripts we should be referencing paths to built jar / files relative to `$TEST_INFRA_DIR`.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r180938483
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,117 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.4.2 \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/test-class/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+sed -i -e '80i\
--- End diff --
Yes. very right! I will fix it.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188486224
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/../flink-quickstart-test/src/main/java/org/apache/flink/quickstart/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package -nsu
+
+cd target
+jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+
+if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Success: There are no flink core classes are contained in the jar."
+else
+ echo "Failure: There are flink core classes are contained in the jar."
+ exit 1
--- End diff --
Will do.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r181699682
--- Diff: flink-end-to-end-tests/test-scripts/test-class/ElasticsearchStreamingJob.java ---
@@ -0,0 +1,97 @@
+/*
+ * 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.quickstart;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
+
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.Requests;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Elasticsearch example for Flink Streaming Job.
+ *
+ * <p>In this streaming job, we generate a bunch of data from numbers, apply two operators map
+ * and filter to that data. Then we choose elasticsearch as its sink to storage these data.
+ *
+ * <p>Run test_quickstarts.sh to verify this program. Package this class to a jar, verify the jar,
+ * then deploy it on a flink cluster.
+ */
+public class ElasticsearchStreamingJob {
+
+ public static void main(String[] args) throws Exception {
+ // set up the streaming execution environment
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<String> source = env.generateSequence(0, 20)
+ // map the data from 1,2,3... to the form of message #1, message #2, message #3...
+ .map(new MapFunction<Long, String>() {
+ @Override
+ public String map(Long value) throws Exception {
+ return "message #" + value;
+ }})
+ // filter out the data that contains message #11 and message #17
+ .filter(new FilterFunction<String>() {
+ @Override
+ public boolean filter(String value) throws Exception {
+ return !value.equals("message #11") && !value.equals("message #17");
+ }
+ });
+
+ Map<String, String> userConfig = new HashMap<>();
+ userConfig.put("cluster.name", "elasticsearch");
+ // This instructs the sink to emit after every element, otherwise they would be buffered
+ userConfig.put(ElasticsearchSink.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, "1");
+
+ List<InetSocketAddress> transports = new ArrayList<>();
+ transports.add(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9300));
+
+ source.addSink(new ElasticsearchSink<>(userConfig, transports, new org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction<String>(){
--- End diff --
add an import for ElasticsearchSinkFunction
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r180810121
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,117 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.4.2 \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/test-class/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+sed -i -e '80i\
--- End diff --
This is quite brittle. What you have to realize is that any change to the original pom may now break this test, even if it is just reorganizing the pom.
A better alternative would be to search for the `<dependencies>` tag and insert the dependency after that.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188196115
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/../flink-quickstart-test/src/main/java/org/apache/flink/quickstart/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package -nsu
+
+cd target
+jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+
+if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Success: There are no flink core classes are contained in the jar."
+else
+ echo "Failure: There are flink core classes are contained in the jar."
+ exit 1
+fi
+
+if [[ `grep -c "org/apache/flink/quickstart/StreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/quickstart/ElasticsearchStreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/connectors/elasticsearch2" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Failure: Since ElasticsearchStreamingJob.class and other user classes are not included in the jar. "
+ exit 1
+else
+ echo "Success: ElasticsearchStreamingJob.class and other user classes are included in the jar."
+fi
+
+cd $CURRENT_DIR
+
+source "$(dirname "$0")"/common.sh
+
+start_cluster
+
+mkdir -p $TEST_DATA_DIR
+
+ELASTICSEARCH_URL="https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.5/elasticsearch-2.3.5.tar.gz"
+
+curl "$ELASTICSEARCH_URL" > $TEST_DATA_DIR/elasticsearch.tar.gz
+tar xzf $TEST_DATA_DIR/elasticsearch.tar.gz -C $TEST_DATA_DIR/
+ELASTICSEARCH_DIR=$TEST_DATA_DIR/elasticsearch-2.3.5
+
+# start elasticsearch cluster
+nohup $ELASTICSEARCH_DIR/bin/elasticsearch &
+
+TEST_PROGRAM_JAR=$CURRENT_DIR/flink-java-project/target/flink-java-project-0.1.jar
+
+# run the Flink job
--- End diff --
Before running the Flink job, we should verify that the Elasticsearch node really is running.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r183970230
--- Diff: flink-end-to-end-tests/test-scripts/test-class/ElasticsearchStreamingJob.java ---
@@ -0,0 +1,90 @@
+/*
+ * 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.quickstart;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
+
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.Requests;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Elasticsearch example for Flink Streaming Job.
+ *
+ * <p>In this streaming job, we generate a bunch of data from numbers, apply operator map
+ * made a type conversion. Then we choose elasticsearch as its sink to storage these data.
+ *
+ * <p>Run test_quickstarts.sh to verify this program. Package this class to a jar, verify the jar,
+ * then deploy it on a flink cluster.
+ */
+public class ElasticsearchStreamingJob {
+
+ public static void main(String[] args) throws Exception {
+ // set up the streaming execution environment
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<String> source = env.generateSequence(0, 20)
+ .map(new MapFunction<Long, String>() {
+ @Override
+ public String map(Long value) throws Exception {
+ return value.toString();
+ }});
+
+
--- End diff --
remove empty line (checkstyle)
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by medcv <gi...@git.apache.org>.
Github user medcv commented on the issue:
https://github.com/apache/flink/pull/5823
@zhangminglei @zentol
I was looking at the ticket [FLINK-1044](https://issues.apache.org/jira/browse/FLINK-1044) and seems we can use the output of this test to offer a zip archive with a pre-setup user project, upload it to `https://flink.apache.org/q/quickstart.zip` and update the document to point to this .zip file. This can help for users that have no experience with Maven/SBT setups.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r183756061
--- Diff: flink-end-to-end-tests/test-scripts/test-class/ElasticsearchStreamingJob.java ---
@@ -0,0 +1,97 @@
+/*
+ * 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.quickstart;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
+
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.Requests;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Elasticsearch example for Flink Streaming Job.
+ *
+ * <p>In this streaming job, we generate a bunch of data from numbers, apply two operators map
+ * and filter to that data. Then we choose elasticsearch as its sink to storage these data.
+ *
+ * <p>Run test_quickstarts.sh to verify this program. Package this class to a jar, verify the jar,
+ * then deploy it on a flink cluster.
+ */
+public class ElasticsearchStreamingJob {
+
+ public static void main(String[] args) throws Exception {
+ // set up the streaming execution environment
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<String> source = env.generateSequence(0, 20)
+ // map the data from 1,2,3... to the form of message #1, message #2, message #3...
+ .map(new MapFunction<Long, String>() {
--- End diff --
Will change
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188195971
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/../flink-quickstart-test/src/main/java/org/apache/flink/quickstart/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package -nsu
+
+cd target
+jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+
+if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Success: There are no flink core classes are contained in the jar."
+else
+ echo "Failure: There are flink core classes are contained in the jar."
+ exit 1
+fi
+
+if [[ `grep -c "org/apache/flink/quickstart/StreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/quickstart/ElasticsearchStreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/connectors/elasticsearch2" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Failure: Since ElasticsearchStreamingJob.class and other user classes are not included in the jar. "
+ exit 1
+else
+ echo "Success: ElasticsearchStreamingJob.class and other user classes are included in the jar."
+fi
+
+cd $CURRENT_DIR
+
+source "$(dirname "$0")"/common.sh
+
+start_cluster
+
+mkdir -p $TEST_DATA_DIR
+
+ELASTICSEARCH_URL="https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.5/elasticsearch-2.3.5.tar.gz"
+
+curl "$ELASTICSEARCH_URL" > $TEST_DATA_DIR/elasticsearch.tar.gz
+tar xzf $TEST_DATA_DIR/elasticsearch.tar.gz -C $TEST_DATA_DIR/
+ELASTICSEARCH_DIR=$TEST_DATA_DIR/elasticsearch-2.3.5
+
+# start elasticsearch cluster
+nohup $ELASTICSEARCH_DIR/bin/elasticsearch &
+
+TEST_PROGRAM_JAR=$CURRENT_DIR/flink-java-project/target/flink-java-project-0.1.jar
+
+# run the Flink job
+$FLINK_DIR/bin/flink run -c org.apache.flink.quickstart.ElasticsearchStreamingJob $TEST_PROGRAM_JAR
+
+touch $TEST_DATA_DIR/output
+
+curl 'localhost:9200/my-index/_search?q=*&pretty&size=21' > $TEST_DATA_DIR/output
+
+if [ -n "$(grep '\"total\" : 21' $TEST_DATA_DIR/output)" ]; then
+ echo "Quickstarts end to end test pass."
+else
+ echo "Quickstarts end to end test failed."
+ exit 1
+fi
+
+function shutdownAndCleanup {
+ pid=$(jps | grep Elasticsearch | awk '{print $1}')
+ kill -SIGTERM $pid
+
+ # make sure to run regular cleanup as well
+ cleanup
+
+ rm -r $CURRENT_DIR/flink-java-project
+ rm $CURRENT_DIR/*nohup.out*
--- End diff --
These `rm`s can actually be avoided if ALL test resources (the quickstart project, Elasticsearch binaries) are all placed under `$TEST_DATA_DIR`. The `cleanup` call makes sure to delete that directory after the test ends.
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on the issue:
https://github.com/apache/flink/pull/5823
Since Saturday and Sunday, sorry for later.
Thanks @zentol . I need confirm some points here with you before I getting start again with this issue.
1. I have to create an actual quickstart project. Hmm, Is this a project or a maven module ? And where do I created for that ? If it is a project(sounds weird because all in flink belongs to maven module), I think I should use command in bash to create this project. And the bash command is ```curl https://flink.apache.org/q/quickstart.sh | bash```. As the position for this project, Is it suitable put it to ```flink-end-to-end-tests``` folder ?
2. Verification issue. verify that no core dependencies are contained in the jar file. I do not understand very well here. Is that mean we should check the jar file that no flink-core dependencies in the jar file since there is already have that in flink cluster set up. If we still have flink-core dependency in the jar file, that would cause the jar file very big size. I understand correct ? So, we need have a check here.
3. The job could fail outright, yet the test will still succeed. I also do not understand very well. This means I need make the job ```StreamExecutionEnvironment#enableCheckpointing``` in code ?
If I am wrong, please helps me out here. Thank you in advance. @zentol
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r183999361
--- Diff: flink-end-to-end-tests/test-scripts/test-class/ElasticsearchStreamingJob.java ---
@@ -0,0 +1,90 @@
+/*
+ * 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.quickstart;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
+
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.Requests;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Elasticsearch example for Flink Streaming Job.
+ *
+ * <p>In this streaming job, we generate a bunch of data from numbers, apply operator map
+ * made a type conversion. Then we choose elasticsearch as its sink to storage these data.
+ *
+ * <p>Run test_quickstarts.sh to verify this program. Package this class to a jar, verify the jar,
+ * then deploy it on a flink cluster.
+ */
+public class ElasticsearchStreamingJob {
+
+ public static void main(String[] args) throws Exception {
+ // set up the streaming execution environment
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<String> source = env.generateSequence(0, 20)
+ .map(new MapFunction<Long, String>() {
+ @Override
+ public String map(Long value) throws Exception {
+ return value.toString();
+ }});
+
+
--- End diff --
will change.
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/5823
@medcv That sounds like something we should do as part of the release process (if we are going to do it), orthogonal to this PR here.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188486209
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/../flink-quickstart-test/src/main/java/org/apache/flink/quickstart/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package -nsu
+
+cd target
+jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+
+if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Success: There are no flink core classes are contained in the jar."
+else
+ echo "Failure: There are flink core classes are contained in the jar."
+ exit 1
+fi
+
+if [[ `grep -c "org/apache/flink/quickstart/StreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/quickstart/ElasticsearchStreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/connectors/elasticsearch2" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Failure: Since ElasticsearchStreamingJob.class and other user classes are not included in the jar. "
+ exit 1
+else
+ echo "Success: ElasticsearchStreamingJob.class and other user classes are included in the jar."
+fi
+
+cd $CURRENT_DIR
+
+source "$(dirname "$0")"/common.sh
+
+start_cluster
+
+mkdir -p $TEST_DATA_DIR
+
+ELASTICSEARCH_URL="https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.5/elasticsearch-2.3.5.tar.gz"
+
+curl "$ELASTICSEARCH_URL" > $TEST_DATA_DIR/elasticsearch.tar.gz
+tar xzf $TEST_DATA_DIR/elasticsearch.tar.gz -C $TEST_DATA_DIR/
+ELASTICSEARCH_DIR=$TEST_DATA_DIR/elasticsearch-2.3.5
+
+# start elasticsearch cluster
+nohup $ELASTICSEARCH_DIR/bin/elasticsearch &
+
+TEST_PROGRAM_JAR=$CURRENT_DIR/flink-java-project/target/flink-java-project-0.1.jar
+
+# run the Flink job
+$FLINK_DIR/bin/flink run -c org.apache.flink.quickstart.ElasticsearchStreamingJob $TEST_PROGRAM_JAR
+
+touch $TEST_DATA_DIR/output
+
+curl 'localhost:9200/my-index/_search?q=*&pretty&size=21' > $TEST_DATA_DIR/output
+
+if [ -n "$(grep '\"total\" : 21' $TEST_DATA_DIR/output)" ]; then
+ echo "Quickstarts end to end test pass."
+else
+ echo "Quickstarts end to end test failed."
+ exit 1
+fi
+
+function shutdownAndCleanup {
+ pid=$(jps | grep Elasticsearch | awk '{print $1}')
+ kill -SIGTERM $pid
+
+ # make sure to run regular cleanup as well
+ cleanup
+
+ rm -r $CURRENT_DIR/flink-java-project
+ rm $CURRENT_DIR/*nohup.out*
--- End diff --
Yes. will do.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r184074206
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/test-class/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package
--- End diff --
Will update.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r184072485
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/test-class/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package
+
+cd target
+jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+
+if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Success: There are no flink core classes are contained in the jar."
+else
+ echo "Failure: There are flink core classes are contained in the jar."
+ exit 1
+fi
+
+if [[ `grep -c "org/apache/flink/quickstart/StreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/quickstart/ElasticsearchStreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/connectors/elasticsearch2" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Failure: Since ElasticsearchStreamingJob.class and other user classes are not included in the jar. "
+ exit 1
+else
+ echo "Success: ElasticsearchStreamingJob.class and other user classes are included in the jar."
+fi
+
+cd $CURRENT_DIR
+
+source "$(dirname "$0")"/common.sh
+
+start_cluster
+
+mkdir -p $TEST_DATA_DIR
+
+ELASTICSEARCH_URL="https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.5/elasticsearch-2.3.5.tar.gz"
+
+curl "$ELASTICSEARCH_URL" > $TEST_DATA_DIR/elasticsearch.tar.gz
+tar xzf $TEST_DATA_DIR/elasticsearch.tar.gz -C $TEST_DATA_DIR/
+ELASTICSEARCH_DIR=$TEST_DATA_DIR/elasticsearch-2.3.5
+
+# start elasticsearch cluster
+nohup $ELASTICSEARCH_DIR/bin/elasticsearch &
--- End diff --
I think the elasticsearch can not set up if 9200 or 9300 already taken. But I do a stop elasticsearch process by
```pid=$(jps | grep Elasticsearch | awk '{print $1}')```
```kill -SIGTERM $pid```.
when ```trap INIT and EXIT```
So, That would not happen.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r181699493
--- Diff: flink-end-to-end-tests/test-scripts/test-class/ElasticsearchStreamingJob.java ---
@@ -0,0 +1,97 @@
+/*
+ * 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.quickstart;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
+
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.Requests;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Elasticsearch example for Flink Streaming Job.
+ *
+ * <p>In this streaming job, we generate a bunch of data from numbers, apply two operators map
+ * and filter to that data. Then we choose elasticsearch as its sink to storage these data.
+ *
+ * <p>Run test_quickstarts.sh to verify this program. Package this class to a jar, verify the jar,
+ * then deploy it on a flink cluster.
+ */
+public class ElasticsearchStreamingJob {
+
+ public static void main(String[] args) throws Exception {
+ // set up the streaming execution environment
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<String> source = env.generateSequence(0, 20)
+ // map the data from 1,2,3... to the form of message #1, message #2, message #3...
+ .map(new MapFunction<Long, String>() {
+ @Override
+ public String map(Long value) throws Exception {
+ return "message #" + value;
+ }})
+ // filter out the data that contains message #11 and message #17
+ .filter(new FilterFunction<String>() {
+ @Override
+ public boolean filter(String value) throws Exception {
+ return !value.equals("message #11") && !value.equals("message #17");
+ }
+ });
+
+ Map<String, String> userConfig = new HashMap<>();
+ userConfig.put("cluster.name", "elasticsearch");
+ // This instructs the sink to emit after every element, otherwise they would be buffered
--- End diff --
why would buffering be a problem?
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188195534
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/../flink-quickstart-test/src/main/java/org/apache/flink/quickstart/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package -nsu
+
+cd target
+jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+
+if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Success: There are no flink core classes are contained in the jar."
+else
+ echo "Failure: There are flink core classes are contained in the jar."
+ exit 1
--- End diff --
also set `PASS=""` if you want to fail the e2e test
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on the issue:
https://github.com/apache/flink/pull/5823
Fixing the CI build error...
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r181700170
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.4.2 \
--- End diff --
the version is outdated and must point to 1.5-SNAPSHOT.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188194545
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
--- End diff --
This should be `1.6-SNAPSHOT` now?
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r184000003
--- Diff: flink-end-to-end-tests/test-scripts/test-class/ElasticsearchStreamingJob.java ---
@@ -0,0 +1,90 @@
+/*
+ * 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.quickstart;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
+
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.Requests;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Elasticsearch example for Flink Streaming Job.
+ *
+ * <p>In this streaming job, we generate a bunch of data from numbers, apply operator map
+ * made a type conversion. Then we choose elasticsearch as its sink to storage these data.
+ *
+ * <p>Run test_quickstarts.sh to verify this program. Package this class to a jar, verify the jar,
+ * then deploy it on a flink cluster.
+ */
+public class ElasticsearchStreamingJob {
--- End diff --
I will create a module named ```flink-quickstart-test```.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188486404
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=flink-java-project \
+ -Dversion=0.1 \
+ -Dpackage=org.apache.flink.quickstart \
+ -DinteractiveMode=false
+
+cd flink-java-project
+
+cp $CURRENT_DIR/../flink-quickstart-test/src/main/java/org/apache/flink/quickstart/ElasticsearchStreamingJob.java $CURRENT_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+
+position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
+
+sed -i -e ''"$(($position + 1))"'i\
+<dependency>\
+<groupId>org.apache.flink</groupId>\
+<artifactId>flink-connector-elasticsearch2_${scala.binary.version}</artifactId>\
+<version>${flink.version}</version>\
+</dependency>' pom.xml
+
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.ElasticsearchStreamingJob/" pom.xml
+
+mvn clean package -nsu
+
+cd target
+jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+
+if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Success: There are no flink core classes are contained in the jar."
+else
+ echo "Failure: There are flink core classes are contained in the jar."
+ exit 1
+fi
+
+if [[ `grep -c "org/apache/flink/quickstart/StreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/quickstart/ElasticsearchStreamingJob.class" contentsInJar.txt` -eq '0' && \
+ `grep -c "org/apache/flink/streaming/connectors/elasticsearch2" contentsInJar.txt` -eq '0' ]]; then
+
+ echo "Failure: Since ElasticsearchStreamingJob.class and other user classes are not included in the jar. "
+ exit 1
+else
+ echo "Success: ElasticsearchStreamingJob.class and other user classes are included in the jar."
+fi
+
+cd $CURRENT_DIR
+
+source "$(dirname "$0")"/common.sh
+
+start_cluster
+
+mkdir -p $TEST_DATA_DIR
+
+ELASTICSEARCH_URL="https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.5/elasticsearch-2.3.5.tar.gz"
+
+curl "$ELASTICSEARCH_URL" > $TEST_DATA_DIR/elasticsearch.tar.gz
+tar xzf $TEST_DATA_DIR/elasticsearch.tar.gz -C $TEST_DATA_DIR/
+ELASTICSEARCH_DIR=$TEST_DATA_DIR/elasticsearch-2.3.5
+
+# start elasticsearch cluster
+nohup $ELASTICSEARCH_DIR/bin/elasticsearch &
+
+TEST_PROGRAM_JAR=$CURRENT_DIR/flink-java-project/target/flink-java-project-0.1.jar
+
+# run the Flink job
+$FLINK_DIR/bin/flink run -c org.apache.flink.quickstart.ElasticsearchStreamingJob $TEST_PROGRAM_JAR
+
+touch $TEST_DATA_DIR/output
+
+curl 'localhost:9200/my-index/_search?q=*&pretty&size=21' > $TEST_DATA_DIR/output
+
+if [ -n "$(grep '\"total\" : 21' $TEST_DATA_DIR/output)" ]; then
+ echo "Quickstarts end to end test pass."
+else
+ echo "Quickstarts end to end test failed."
+ exit 1
--- End diff --
Will change.
---
[GitHub] flink pull request #5823: [FLINK-9008] [e2e] Implements quickstarts end to e...
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on a diff in the pull request:
https://github.com/apache/flink/pull/5823#discussion_r188486388
--- Diff: flink-end-to-end-tests/test-scripts/test_quickstarts.sh ---
@@ -0,0 +1,123 @@
+#!/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.
+################################################################################
+
+# End to end test for quick starts test.
+
+CURRENT_DIR=$(cd "$( dirname "$0" )" && pwd )
+
+cd $CURRENT_DIR
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.5-SNAPSHOT \
--- End diff --
Yes. Will change.
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on the issue:
https://github.com/apache/flink/pull/5823
This commit seems has an issue. not suitable to review now. I will fix it soon.
---
[GitHub] flink issue #5823: [FLINK-9008] [e2e] Implements quickstarts end to end test
Posted by zhangminglei <gi...@git.apache.org>.
Github user zhangminglei commented on the issue:
https://github.com/apache/flink/pull/5823
Hi, @zentol . What is the state of this PR ? Thanks.
---