You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2018/06/12 09:36:59 UTC
[1/5] flink git commit: [FLINK-9564][tests] Expose flink-end-to-end
module directory to tests
Repository: flink
Updated Branches:
refs/heads/master 5f11df6ee -> 0f5cf7b15
[FLINK-9564][tests] Expose flink-end-to-end module directory to tests
This closes #6145.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/89ab475c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/89ab475c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/89ab475c
Branch: refs/heads/master
Commit: 89ab475ccc1fb3ae199236546b25f79a27685529
Parents: 5f11df6
Author: zentol <ch...@apache.org>
Authored: Mon Jun 11 09:52:28 2018 +0200
Committer: zentol <ch...@apache.org>
Committed: Tue Jun 12 11:34:37 2018 +0200
----------------------------------------------------------------------
flink-end-to-end-tests/run-nightly-tests.sh | 2 ++
flink-end-to-end-tests/run-pre-commit-tests.sh | 2 ++
flink-end-to-end-tests/test-scripts/test_batch_allround.sh | 2 +-
flink-end-to-end-tests/test-scripts/test_ha.sh | 2 +-
.../test-scripts/test_high_parallelism_iterations.sh | 2 +-
.../test-scripts/test_local_recovery_and_scheduling.sh | 2 +-
flink-end-to-end-tests/test-scripts/test_quickstarts.sh | 2 +-
.../test-scripts/test_resume_externalized_checkpoints.sh | 2 +-
flink-end-to-end-tests/test-scripts/test_resume_savepoint.sh | 2 +-
.../test-scripts/test_stateful_stream_job_upgrade.sh | 2 +-
flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh | 2 +-
flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh | 2 +-
.../test-scripts/test_streaming_distributed_cache_via_blob.sh | 2 +-
.../test-scripts/test_streaming_elasticsearch.sh | 2 +-
flink-end-to-end-tests/test-scripts/test_streaming_sql.sh | 2 +-
15 files changed, 17 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/run-nightly-tests.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh
index a79754c..39007f5 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -25,6 +25,8 @@ if [ -z "$END_TO_END_DIR" ] ; then
exit 1 # fail
fi
+export END_TO_END_DIR="${END_TO_END_DIR}"
+
if [ -z "$FLINK_DIR" ] ; then
echo "You have to export the Flink distribution directory as FLINK_DIR"
exit 1
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/run-pre-commit-tests.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/run-pre-commit-tests.sh b/flink-end-to-end-tests/run-pre-commit-tests.sh
index 8c82e2e..50b1430 100755
--- a/flink-end-to-end-tests/run-pre-commit-tests.sh
+++ b/flink-end-to-end-tests/run-pre-commit-tests.sh
@@ -25,6 +25,8 @@ if [ -z "$END_TO_END_DIR" ] ; then
exit 1 # fail
fi
+export END_TO_END_DIR="${END_TO_END_DIR}"
+
if [ -z "$FLINK_DIR" ] ; then
echo "You have to export the Flink distribution directory as FLINK_DIR"
exit 1
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_batch_allround.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_batch_allround.sh b/flink-end-to-end-tests/test-scripts/test_batch_allround.sh
index 834fbad..0cbaa21 100755
--- a/flink-end-to-end-tests/test-scripts/test_batch_allround.sh
+++ b/flink-end-to-end-tests/test-scripts/test_batch_allround.sh
@@ -19,7 +19,7 @@
source "$(dirname "$0")"/common.sh
-TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-dataset-allround-test/target/DataSetAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-dataset-allround-test/target/DataSetAllroundTestProgram.jar
echo "Run DataSet-Allround-Test Program"
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_ha.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_ha.sh b/flink-end-to-end-tests/test-scripts/test_ha.sh
index 6d94c03..5714161 100755
--- a/flink-end-to-end-tests/test-scripts/test_ha.sh
+++ b/flink-end-to-end-tests/test-scripts/test_ha.sh
@@ -19,7 +19,7 @@
source "$(dirname "$0")"/common.sh
-TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-datastream-allround-test/target/DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/DataStreamAllroundTestProgram.jar
JM_WATCHDOG_PID=0
TM_WATCHDOG_PID=0
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_high_parallelism_iterations.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_high_parallelism_iterations.sh b/flink-end-to-end-tests/test-scripts/test_high_parallelism_iterations.sh
index 3240f8a..99f8017 100755
--- a/flink-end-to-end-tests/test-scripts/test_high_parallelism_iterations.sh
+++ b/flink-end-to-end-tests/test-scripts/test_high_parallelism_iterations.sh
@@ -23,7 +23,7 @@ PARALLELISM="${PARALLELISM:-100}"
TEST=flink-high-parallelism-iterations-test
TEST_PROGRAM_NAME=HighParallelismIterationsTestProgram
-TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/$TEST/target/$TEST_PROGRAM_NAME.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/$TEST/target/$TEST_PROGRAM_NAME.jar
echo "Run Not So MiniCluster Iterations Graph Connected Components Program"
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_local_recovery_and_scheduling.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_local_recovery_and_scheduling.sh b/flink-end-to-end-tests/test-scripts/test_local_recovery_and_scheduling.sh
index 98ef01f..13e7616 100755
--- a/flink-end-to-end-tests/test-scripts/test_local_recovery_and_scheduling.sh
+++ b/flink-end-to-end-tests/test-scripts/test_local_recovery_and_scheduling.sh
@@ -72,7 +72,7 @@ function run_local_recovery_test {
local kill_jvm=$5
echo "Running local recovery test on ${backend} backend: incremental checkpoints = ${incremental}, kill JVM = ${kill_jvm}."
- TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-local-recovery-and-allocation-test/target/StickyAllocationAndLocalRecoveryTestJob.jar
+ TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-local-recovery-and-allocation-test/target/StickyAllocationAndLocalRecoveryTestJob.jar
# Backup conf and configure for HA
backup_config
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_quickstarts.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_quickstarts.sh b/flink-end-to-end-tests/test-scripts/test_quickstarts.sh
index 3f63226..3db64fc 100755
--- a/flink-end-to-end-tests/test-scripts/test_quickstarts.sh
+++ b/flink-end-to-end-tests/test-scripts/test_quickstarts.sh
@@ -39,7 +39,7 @@ mvn archetype:generate \
cd flink-java-project
# use the Flink Elasticsearch sink example job code in flink-elasticsearch5-tests to simulate modifications to contained job
-cp $TEST_INFRA_DIR/../flink-elasticsearch5-test/src/main/java/org/apache/flink/streaming/tests/Elasticsearch5SinkExample.java $TEST_DATA_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
+cp ${END_TO_END_DIR}/flink-elasticsearch5-test/src/main/java/org/apache/flink/streaming/tests/Elasticsearch5SinkExample.java $TEST_DATA_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
sed -i -e 's/package org.apache.flink.streaming.tests;/package org.apache.flink.quickstart;/' $TEST_DATA_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/Elasticsearch5SinkExample.java
position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_resume_externalized_checkpoints.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_resume_externalized_checkpoints.sh b/flink-end-to-end-tests/test-scripts/test_resume_externalized_checkpoints.sh
index f9ce9d6..5cb2d8e 100755
--- a/flink-end-to-end-tests/test-scripts/test_resume_externalized_checkpoints.sh
+++ b/flink-end-to-end-tests/test-scripts/test_resume_externalized_checkpoints.sh
@@ -48,7 +48,7 @@ CHECKPOINT_DIR_URI="file://$CHECKPOINT_DIR"
echo "Running externalized checkpoints test, with STATE_BACKEND_TYPE=$STATE_BACKEND_TYPE STATE_BACKEND_FILE_ASYNC=$STATE_BACKEND_FILE_ASYNC SIMULATE_FAILURE=$SIMULATE_FAILURE ..."
-TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-datastream-allround-test/target/DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/DataStreamAllroundTestProgram.jar
BASE_JOB_CMD="$FLINK_DIR/bin/flink run -d $TEST_PROGRAM_JAR \
--test.semantics exactly-once \
--environment.externalize_checkpoint true \
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_resume_savepoint.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_resume_savepoint.sh b/flink-end-to-end-tests/test-scripts/test_resume_savepoint.sh
index 2060a87..084cf51 100755
--- a/flink-end-to-end-tests/test-scripts/test_resume_savepoint.sh
+++ b/flink-end-to-end-tests/test-scripts/test_resume_savepoint.sh
@@ -79,7 +79,7 @@ trap test_cleanup EXIT
CHECKPOINT_DIR="file://$TEST_DATA_DIR/savepoint-e2e-test-chckpt-dir"
# run the DataStream allroundjob
-TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-datastream-allround-test/target/DataStreamAllroundTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/DataStreamAllroundTestProgram.jar
DATASTREAM_JOB=$($FLINK_DIR/bin/flink run -d -p $ORIGINAL_DOP $TEST_PROGRAM_JAR \
--test.semantics exactly-once \
--environment.parallelism $ORIGINAL_DOP \
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_stateful_stream_job_upgrade.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_stateful_stream_job_upgrade.sh b/flink-end-to-end-tests/test-scripts/test_stateful_stream_job_upgrade.sh
index 44d9df0..3ecda5a 100755
--- a/flink-end-to-end-tests/test-scripts/test_stateful_stream_job_upgrade.sh
+++ b/flink-end-to-end-tests/test-scripts/test_stateful_stream_job_upgrade.sh
@@ -52,7 +52,7 @@ trap test_cleanup EXIT
CHECKPOINT_DIR="file://${TEST_DATA_DIR}/savepoint-e2e-test-chckpt-dir"
-TEST_PROGRAM_JAR="${TEST_INFRA_DIR}/../../flink-end-to-end-tests/flink-stream-stateful-job-upgrade-test/target/StatefulStreamJobUpgradeTestProgram.jar"
+TEST_PROGRAM_JAR="${END_TO_END_DIR}/flink-stream-stateful-job-upgrade-test/target/StatefulStreamJobUpgradeTestProgram.jar"
function job() {
DOP=$1
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh b/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
index b4c3ce9..10548d8 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_bucketing.sh
@@ -19,7 +19,7 @@
source "$(dirname "$0")"/common.sh
-TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-bucketing-sink-test/target/BucketingSinkTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-bucketing-sink-test/target/BucketingSinkTestProgram.jar
# enable DEBUG logging level to retrieve truncate length later
sed -i -e 's/#log4j.logger.org.apache.flink=INFO/log4j.logger.org.apache.flink=DEBUG/g' $FLINK_DIR/conf/log4j.properties
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh b/flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh
index 34c55f7..ce259a0 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh
@@ -19,7 +19,7 @@
source "$(dirname "$0")"/common.sh
-TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-parent-child-classloading-test/target/ClassLoaderTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-parent-child-classloading-test/target/ClassLoaderTestProgram.jar
echo "Testing parent-first class loading"
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_streaming_distributed_cache_via_blob.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_distributed_cache_via_blob.sh b/flink-end-to-end-tests/test-scripts/test_streaming_distributed_cache_via_blob.sh
index 25ae7a0..1d814ed 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_distributed_cache_via_blob.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_distributed_cache_via_blob.sh
@@ -19,7 +19,7 @@
source "$(dirname "$0")"/common.sh
-TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-distributed-cache-via-blob-test/target/DistributedCacheViaBlobTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-distributed-cache-via-blob-test/target/DistributedCacheViaBlobTestProgram.jar
echo "Testing distributing files via DistributedCache & BlobServer"
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh b/flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh
index 78ea283..10e72e9 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh
@@ -40,7 +40,7 @@ function test_cleanup {
trap test_cleanup INT
trap test_cleanup EXIT
-TEST_ES_JAR=$TEST_DATA_DIR/../../flink-elasticsearch${ELASTICSEARCH_VERSION}-test/target/Elasticsearch${ELASTICSEARCH_VERSION}SinkExample.jar
+TEST_ES_JAR=${END_TO_END_DIR}/flink-elasticsearch${ELASTICSEARCH_VERSION}-test/target/Elasticsearch${ELASTICSEARCH_VERSION}SinkExample.jar
# run the Flink job
$FLINK_DIR/bin/flink run -p 1 $TEST_ES_JAR \
http://git-wip-us.apache.org/repos/asf/flink/blob/89ab475c/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh b/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
index b7af5fb..391b623 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_sql.sh
@@ -19,7 +19,7 @@
source "$(dirname "$0")"/common.sh
-TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-stream-sql-test/target/StreamSQLTestProgram.jar
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-stream-sql-test/target/StreamSQLTestProgram.jar
# copy flink-table jar into lib folder
cp $FLINK_DIR/opt/flink-table*jar $FLINK_DIR/lib
[5/5] flink git commit: [FLINK-8946][metrics] Do not close TM
MetricGroup on JobManager failover
Posted by ch...@apache.org.
[FLINK-8946][metrics] Do not close TM MetricGroup on JobManager failover
This closes #6060.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0f5cf7b1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0f5cf7b1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0f5cf7b1
Branch: refs/heads/master
Commit: 0f5cf7b15592dd5d9a9303bf7a806d553e679d17
Parents: 25d277e
Author: yanghua <ya...@gmail.com>
Authored: Wed May 23 17:42:04 2018 +0800
Committer: zentol <ch...@apache.org>
Committed: Tue Jun 12 11:34:39 2018 +0200
----------------------------------------------------------------------
.../org/apache/flink/runtime/taskmanager/TaskManager.scala | 7 -------
1 file changed, 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0f5cf7b1/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 2d48b31..9a057ab 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -1080,13 +1080,6 @@ class TaskManager(
// clear the key-value location oracle
proxy.updateKvStateLocationOracle(HighAvailabilityServices.DEFAULT_JOB_ID, null)
}
-
- // failsafe shutdown of the metrics registry
- try {
- taskManagerMetricGroup.close()
- } catch {
- case t: Exception => log.warn("TaskManagerMetricGroup could not be closed successfully.", t)
- }
}
protected def handleJobManagerDisconnect(msg: String): Unit = {
[4/5] flink git commit: [FLINK-9566][cli] Support listing all jobs
Posted by ch...@apache.org.
[FLINK-9566][cli] Support listing all jobs
This closes #6049.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/25d277e1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/25d277e1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/25d277e1
Branch: refs/heads/master
Commit: 25d277e16970c6a1755f5050f7bc419b158f17aa
Parents: a2a85bc
Author: Rong Rong <wa...@hotmail.com>
Authored: Mon Jun 11 10:36:59 2018 +0200
Committer: zentol <ch...@apache.org>
Committed: Tue Jun 12 11:34:38 2018 +0200
----------------------------------------------------------------------
docs/ops/cli.md | 4 ++
.../apache/flink/client/cli/CliFrontend.java | 73 ++++++++++++--------
.../flink/client/cli/CliFrontendParser.java | 4 ++
.../apache/flink/client/cli/ListOptions.java | 23 +++---
.../flink/client/cli/CliFrontendListTest.java | 65 ++++++++++++++---
5 files changed, 123 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/25d277e1/docs/ops/cli.md
----------------------------------------------------------------------
diff --git a/docs/ops/cli.md b/docs/ops/cli.md
index 95c3c45..6974a2d 100644
--- a/docs/ops/cli.md
+++ b/docs/ops/cli.md
@@ -106,6 +106,10 @@ The command line can be used to
./bin/flink list -r
+- List all existing jobs (including their JobIDs):
+
+ ./bin/flink list -a
+
- List running Flink jobs inside Flink YARN session:
./bin/flink list -m yarn-cluster -yid <yarnApplicationID> -r
http://git-wip-us.apache.org/repos/asf/flink/blob/25d277e1/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index f01f5b8..e9a1590 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -81,6 +81,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import scala.concurrent.duration.FiniteDuration;
@@ -392,16 +393,19 @@ public class CliFrontend {
return;
}
- final boolean running;
- final boolean scheduled;
+ final boolean showRunning;
+ final boolean showScheduled;
+ final boolean showAll;
// print running and scheduled jobs if not option supplied
- if (!listOptions.getRunning() && !listOptions.getScheduled()) {
- running = true;
- scheduled = true;
+ if (!listOptions.showRunning() && !listOptions.showScheduled() && !listOptions.showAll()) {
+ showRunning = true;
+ showScheduled = true;
+ showAll = false;
} else {
- running = listOptions.getRunning();
- scheduled = listOptions.getScheduled();
+ showRunning = listOptions.showRunning();
+ showScheduled = listOptions.showScheduled();
+ showAll = listOptions.showAll();
}
final CustomCommandLine<?> activeCommandLine = getActiveCustomCommandLine(commandLine);
@@ -409,14 +413,15 @@ public class CliFrontend {
runClusterAction(
activeCommandLine,
commandLine,
- clusterClient -> listJobs(clusterClient, running, scheduled));
+ clusterClient -> listJobs(clusterClient, showRunning, showScheduled, showAll));
}
private <T> void listJobs(
ClusterClient<T> clusterClient,
- boolean running,
- boolean scheduled) throws FlinkException {
+ boolean showRunning,
+ boolean showScheduled,
+ boolean showAll) throws FlinkException {
Collection<JobStatusMessage> jobDetails;
try {
CompletableFuture<Collection<JobStatusMessage>> jobDetailsFuture = clusterClient.listJobs();
@@ -431,49 +436,63 @@ public class CliFrontend {
LOG.info("Successfully retrieved list of jobs");
- SimpleDateFormat dateFormat = new SimpleDateFormat("dd.MM.yyyy HH:mm:ss");
- Comparator<JobStatusMessage> startTimeComparator = (o1, o2) -> (int) (o1.getStartTime() - o2.getStartTime());
-
final List<JobStatusMessage> runningJobs = new ArrayList<>();
final List<JobStatusMessage> scheduledJobs = new ArrayList<>();
+ final List<JobStatusMessage> terminatedJobs = new ArrayList<>();
jobDetails.forEach(details -> {
if (details.getJobState() == JobStatus.CREATED) {
scheduledJobs.add(details);
} else if (!details.getJobState().isGloballyTerminalState()) {
runningJobs.add(details);
+ } else {
+ terminatedJobs.add(details);
}
});
- if (running) {
+ if (showRunning || showAll) {
if (runningJobs.size() == 0) {
System.out.println("No running jobs.");
}
else {
- runningJobs.sort(startTimeComparator);
-
System.out.println("------------------ Running/Restarting Jobs -------------------");
- for (JobStatusMessage runningJob : runningJobs) {
- System.out.println(dateFormat.format(new Date(runningJob.getStartTime()))
- + " : " + runningJob.getJobId() + " : " + runningJob.getJobName() + " (" + runningJob.getJobState() + ")");
- }
+ printJobStatusMessages(runningJobs);
System.out.println("--------------------------------------------------------------");
}
}
- if (scheduled) {
+ if (showScheduled || showAll) {
if (scheduledJobs.size() == 0) {
System.out.println("No scheduled jobs.");
}
else {
- scheduledJobs.sort(startTimeComparator);
-
System.out.println("----------------------- Scheduled Jobs -----------------------");
- for (JobStatusMessage scheduledJob : scheduledJobs) {
- System.out.println(dateFormat.format(new Date(scheduledJob.getStartTime()))
- + " : " + scheduledJob.getJobId() + " : " + scheduledJob.getJobName());
- }
+ printJobStatusMessages(scheduledJobs);
System.out.println("--------------------------------------------------------------");
}
}
+ if (showAll) {
+ if (terminatedJobs.size() != 0) {
+ System.out.println("---------------------- Terminated Jobs -----------------------");
+ printJobStatusMessages(terminatedJobs);
+ System.out.println("--------------------------------------------------------------");
+ }
+ }
+ }
+
+ private static void printJobStatusMessages(List<JobStatusMessage> jobs) {
+ SimpleDateFormat dateFormat = new SimpleDateFormat("dd.MM.yyyy HH:mm:ss");
+ Comparator<JobStatusMessage> startTimeComparator = (o1, o2) -> (int) (o1.getStartTime() - o2.getStartTime());
+ Comparator<Map.Entry<JobStatus, List<JobStatusMessage>>> statusComparator =
+ (o1, o2) -> String.CASE_INSENSITIVE_ORDER.compare(o1.getKey().toString(), o2.getKey().toString());
+
+ Map<JobStatus, List<JobStatusMessage>> jobsByState = jobs.stream().collect(Collectors.groupingBy(JobStatusMessage::getJobState));
+ jobsByState.entrySet().stream()
+ .sorted(statusComparator)
+ .map(Map.Entry::getValue).flatMap(List::stream).sorted(startTimeComparator)
+ .forEachOrdered(job -> {
+ System.out.println(dateFormat.format(new Date(job.getStartTime()))
+ + " : " + job.getJobId() + " : " + job.getJobName()
+ + " (" + job.getJobState() + ")");
+ });
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/25d277e1/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
index 1588aac..0adb8cf 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
@@ -94,6 +94,9 @@ public class CliFrontendParser {
static final Option SCHEDULED_OPTION = new Option("s", "scheduled", false,
"Show only scheduled programs and their JobIDs");
+ static final Option ALL_OPTION = new Option("a", "all", false,
+ "Show all programs and their JobIDs");
+
static final Option ZOOKEEPER_NAMESPACE_OPTION = new Option("z", "zookeeperNamespace", true,
"Namespace to create the Zookeeper sub-paths for high availability mode");
@@ -193,6 +196,7 @@ public class CliFrontendParser {
static Options getListCommandOptions() {
Options options = buildGeneralOptions(new Options());
+ options.addOption(ALL_OPTION);
options.addOption(RUNNING_OPTION);
return options.addOption(SCHEDULED_OPTION);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/25d277e1/flink-clients/src/main/java/org/apache/flink/client/cli/ListOptions.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/ListOptions.java b/flink-clients/src/main/java/org/apache/flink/client/cli/ListOptions.java
index 7ae00cf..f4c1246 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/ListOptions.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/ListOptions.java
@@ -20,6 +20,7 @@ package org.apache.flink.client.cli;
import org.apache.commons.cli.CommandLine;
+import static org.apache.flink.client.cli.CliFrontendParser.ALL_OPTION;
import static org.apache.flink.client.cli.CliFrontendParser.RUNNING_OPTION;
import static org.apache.flink.client.cli.CliFrontendParser.SCHEDULED_OPTION;
@@ -28,20 +29,26 @@ import static org.apache.flink.client.cli.CliFrontendParser.SCHEDULED_OPTION;
*/
public class ListOptions extends CommandLineOptions {
- private final boolean running;
- private final boolean scheduled;
+ private final boolean showRunning;
+ private final boolean showScheduled;
+ private final boolean showAll;
public ListOptions(CommandLine line) {
super(line);
- this.running = line.hasOption(RUNNING_OPTION.getOpt());
- this.scheduled = line.hasOption(SCHEDULED_OPTION.getOpt());
+ this.showAll = line.hasOption(ALL_OPTION.getOpt());
+ this.showRunning = line.hasOption(RUNNING_OPTION.getOpt());
+ this.showScheduled = line.hasOption(SCHEDULED_OPTION.getOpt());
}
- public boolean getRunning() {
- return running;
+ public boolean showRunning() {
+ return showRunning;
}
- public boolean getScheduled() {
- return scheduled;
+ public boolean showScheduled() {
+ return showScheduled;
+ }
+
+ public boolean showAll() {
+ return showAll;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/25d277e1/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendListTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendListTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendListTest.java
index a8a7f00..be37ccc 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendListTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendListTest.java
@@ -18,18 +18,24 @@
package org.apache.flink.client.cli;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.client.cli.util.MockedCliFrontend;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.jobgraph.JobStatus;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.Mockito;
+import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.when;
@@ -50,15 +56,35 @@ public class CliFrontendListTest extends CliFrontendTestBase {
}
@Test
- public void testList() throws Exception {
- // test list properly
+ public void testListOptions() throws Exception {
+ // test configure all job
{
- String[] parameters = {"-r", "-s"};
- ClusterClient<String> clusterClient = createClusterClient();
- MockedCliFrontend testFrontend = new MockedCliFrontend(clusterClient);
- testFrontend.list(parameters);
- Mockito.verify(clusterClient, times(1))
- .listJobs();
+ String[] parameters = {"-a"};
+ ListOptions options = new ListOptions(CliFrontendParser.parse(
+ CliFrontendParser.getListCommandOptions(), parameters, true));
+ assertTrue(options.showAll());
+ assertFalse(options.showRunning());
+ assertFalse(options.showScheduled());
+ }
+
+ // test configure running job
+ {
+ String[] parameters = {"-r"};
+ ListOptions options = new ListOptions(CliFrontendParser.parse(
+ CliFrontendParser.getListCommandOptions(), parameters, true));
+ assertFalse(options.showAll());
+ assertTrue(options.showRunning());
+ assertFalse(options.showScheduled());
+ }
+
+ // test configure scheduled job
+ {
+ String[] parameters = {"-s"};
+ ListOptions options = new ListOptions(CliFrontendParser.parse(
+ CliFrontendParser.getListCommandOptions(), parameters, true));
+ assertFalse(options.showAll());
+ assertFalse(options.showRunning());
+ assertTrue(options.showScheduled());
}
}
@@ -72,11 +98,28 @@ public class CliFrontendListTest extends CliFrontendTestBase {
testFrontend.list(parameters);
}
+ @Test
+ public void testList() throws Exception {
+ // test list properly
+ {
+ String[] parameters = {"-r", "-s", "-a"};
+ ClusterClient<String> clusterClient = createClusterClient();
+ MockedCliFrontend testFrontend = new MockedCliFrontend(clusterClient);
+ testFrontend.list(parameters);
+ Mockito.verify(clusterClient, times(1))
+ .listJobs();
+ }
+ }
+
private static ClusterClient<String> createClusterClient() throws Exception {
final ClusterClient<String> clusterClient = mock(ClusterClient.class);
-
- when(clusterClient.listJobs()).thenReturn(CompletableFuture.completedFuture(Collections.emptyList()));
-
+ when(clusterClient.listJobs()).thenReturn(CompletableFuture.completedFuture(Arrays.asList(
+ new JobStatusMessage(new JobID(), "job1", JobStatus.RUNNING, 1L),
+ new JobStatusMessage(new JobID(), "job2", JobStatus.CREATED, 1L),
+ new JobStatusMessage(new JobID(), "job3", JobStatus.SUSPENDING, 3L),
+ new JobStatusMessage(new JobID(), "job4", JobStatus.SUSPENDING, 2L),
+ new JobStatusMessage(new JobID(), "job5", JobStatus.FINISHED, 3L)
+ )));
return clusterClient;
}
}
[2/5] flink git commit: [FLINK-9398][cli] Only list non-terminal jobs
in running section
Posted by ch...@apache.org.
[FLINK-9398][cli] Only list non-terminal jobs in running section
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a2a85bcb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a2a85bcb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a2a85bcb
Branch: refs/heads/master
Commit: a2a85bcb3cdb5202055984604f45809260787fd5
Parents: 42e90c7
Author: Rong Rong <wa...@hotmail.com>
Authored: Mon Jun 11 10:35:29 2018 +0200
Committer: zentol <ch...@apache.org>
Committed: Tue Jun 12 11:34:38 2018 +0200
----------------------------------------------------------------------
.../src/main/java/org/apache/flink/client/cli/CliFrontend.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a2a85bcb/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index fd12a73..f01f5b8 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -439,7 +439,7 @@ public class CliFrontend {
jobDetails.forEach(details -> {
if (details.getJobState() == JobStatus.CREATED) {
scheduledJobs.add(details);
- } else {
+ } else if (!details.getJobState().isGloballyTerminalState()) {
runningJobs.add(details);
}
});
[3/5] flink git commit: [FLINK-9451][tests] Add scala quickstart
end-to-end test
Posted by ch...@apache.org.
[FLINK-9451][tests] Add scala quickstart end-to-end test
This closes #6089.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/42e90c77
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/42e90c77
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/42e90c77
Branch: refs/heads/master
Commit: 42e90c77319eeb4c1a073a66449af3dde016d61d
Parents: 89ab475
Author: Yadan.JS <y_...@yahoo.com>
Authored: Mon May 28 12:16:40 2018 -0400
Committer: zentol <ch...@apache.org>
Committed: Tue Jun 12 11:34:38 2018 +0200
----------------------------------------------------------------------
.../flink-quickstart-test/pom.xml | 63 +++++++++++++
.../test/Elasticsearch5SinkExample.java | 94 ++++++++++++++++++++
.../test/Elasticsearch5SinkExample.scala | 77 ++++++++++++++++
flink-end-to-end-tests/pom.xml | 1 +
flink-end-to-end-tests/run-nightly-tests.sh | 7 +-
.../test-scripts/elasticsearch-common.sh | 5 +-
.../test-scripts/test_quickstarts.sh | 70 +++++++++------
.../test_streaming_elasticsearch.sh | 4 +-
8 files changed, 289 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/42e90c77/flink-end-to-end-tests/flink-quickstart-test/pom.xml
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-quickstart-test/pom.xml b/flink-end-to-end-tests/flink-quickstart-test/pom.xml
new file mode 100644
index 0000000..46b402d
--- /dev/null
+++ b/flink-end-to-end-tests/flink-quickstart-test/pom.xml
@@ -0,0 +1,63 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-end-to-end-tests</artifactId>
+ <version>1.6-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-quickstart-test</artifactId>
+ <name>flink-quickstart-test</name>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-quickstart-java</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-quickstart-scala</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-elasticsearch5_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/42e90c77/flink-end-to-end-tests/flink-quickstart-test/src/main/java/org/apache/flink/quickstarts/test/Elasticsearch5SinkExample.java
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-quickstart-test/src/main/java/org/apache/flink/quickstarts/test/Elasticsearch5SinkExample.java b/flink-end-to-end-tests/flink-quickstart-test/src/main/java/org/apache/flink/quickstarts/test/Elasticsearch5SinkExample.java
new file mode 100644
index 0000000..900e495
--- /dev/null
+++ b/flink-end-to-end-tests/flink-quickstart-test/src/main/java/org/apache/flink/quickstarts/test/Elasticsearch5SinkExample.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.quickstarts.test;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
+import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
+import org.apache.flink.streaming.connectors.elasticsearch5.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;
+
+/**
+ * End to end test for Elasticsearch5Sink.
+ */
+public class Elasticsearch5SinkExample {
+
+ public static void main(String[] args) throws Exception {
+
+ final ParameterTool parameterTool = ParameterTool.fromArgs(args);
+
+ if (parameterTool.getNumberOfParameters() < 3) {
+ System.out.println("Missing parameters!\n" +
+ "Usage: --numRecords <numRecords> --index <index> --type <type>");
+ return;
+ }
+
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.getConfig().disableSysoutLogging();
+ env.enableCheckpointing(5000);
+
+ DataStream<String> source = env.generateSequence(0, parameterTool.getInt("numRecords") - 1)
+ .map(new MapFunction<Long, String>() {
+ @Override
+ public String map(Long value) throws Exception {
+ return "message #" + value;
+ }
+ });
+
+ 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 ElasticsearchSinkFunction<String>() {
+ @Override
+ public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
+ indexer.add(createIndexRequest(element, parameterTool));
+ }
+ }));
+
+ env.execute("Elasticsearch5.x end to end sink test example");
+ }
+
+ private static IndexRequest createIndexRequest(String element, ParameterTool parameterTool) {
+ Map<String, Object> json = new HashMap<>();
+ json.put("data", element);
+
+ return Requests.indexRequest()
+ .index(parameterTool.getRequired("index"))
+ .type(parameterTool.getRequired("type"))
+ .id(element)
+ .source(json);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/42e90c77/flink-end-to-end-tests/flink-quickstart-test/src/main/scala/org/apache/flink/quickstarts/test/Elasticsearch5SinkExample.scala
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-quickstart-test/src/main/scala/org/apache/flink/quickstarts/test/Elasticsearch5SinkExample.scala b/flink-end-to-end-tests/flink-quickstart-test/src/main/scala/org/apache/flink/quickstarts/test/Elasticsearch5SinkExample.scala
new file mode 100644
index 0000000..9e2a115
--- /dev/null
+++ b/flink-end-to-end-tests/flink-quickstart-test/src/main/scala/org/apache/flink/quickstarts/test/Elasticsearch5SinkExample.scala
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.quickstarts.test
+
+import org.apache.flink.api.common.functions.RuntimeContext
+import org.apache.flink.api.java.utils.ParameterTool
+import org.apache.flink.streaming.api.scala._
+import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
+import org.apache.flink.streaming.connectors.elasticsearch5.ElasticsearchSink
+import org.elasticsearch.action.index.IndexRequest
+import org.elasticsearch.client.Requests
+
+import scala.collection.JavaConversions.mapAsJavaMap
+
+import java.net.{InetAddress, InetSocketAddress}
+import java.util
+
+
+object Elasticsearch5SinkExample {
+ def main(args: Array[String]) {
+
+ val parameterTool = ParameterTool.fromArgs(args)
+
+ if (parameterTool.getNumberOfParameters < 3) {
+ println("Missing parameters!\n" + "Usage:" +
+ " --numRecords <numRecords> --index <index> --type <type>")
+ return
+ }
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.getConfig.disableSysoutLogging
+ env.enableCheckpointing(5000)
+
+ val source: DataStream[(String)] = env.generateSequence(0, 20 - 1)
+ .map(v => "message #" + v.toString)
+
+ val config = new util.HashMap[String, String]
+ config.put("bulk.flush.max.actions", "1")
+ config.put("cluster.name", "elasticsearch") //default cluster name: elasticsearch
+
+ val transports = new util.ArrayList[InetSocketAddress]
+ transports.add(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9300))
+
+ source.addSink(new ElasticsearchSink(config, transports,
+ new ElasticsearchSinkFunction[(String)] {
+ def createIndexRequest(element: (String)): IndexRequest = {
+
+ val json2 = Map(
+ "data" -> element.asInstanceOf[AnyRef]
+ )
+
+ Requests.indexRequest.index(parameterTool.getRequired("index"))
+ .`type`(parameterTool.getRequired("type")).source(mapAsJavaMap(json2))
+ }
+
+ override def process(element: (String), ctx: RuntimeContext, indexer: RequestIndexer) {
+ indexer.add(createIndexRequest(element))
+ }
+ }))
+
+ env.execute("Elasticsearch5.x end to end sink test example")
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/42e90c77/flink-end-to-end-tests/pom.xml
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/pom.xml b/flink-end-to-end-tests/pom.xml
index 581abc8..8fb7eb8 100644
--- a/flink-end-to-end-tests/pom.xml
+++ b/flink-end-to-end-tests/pom.xml
@@ -47,6 +47,7 @@ under the License.
<module>flink-elasticsearch1-test</module>
<module>flink-elasticsearch2-test</module>
<module>flink-elasticsearch5-test</module>
+ <module>flink-quickstart-test</module>
</modules>
<build>
http://git-wip-us.apache.org/repos/asf/flink/blob/42e90c77/flink-end-to-end-tests/run-nightly-tests.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh
index 39007f5..46b2609 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -191,7 +191,12 @@ if [ $EXIT_CODE == 0 ]; then
fi
if [ $EXIT_CODE == 0 ]; then
- run_test "Quickstarts nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_quickstarts.sh"
+ run_test "Quickstarts Java nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_quickstarts.sh java"
+ EXIT_CODE=$?
+fi
+
+if [ $EXIT_CODE == 0 ]; then
+ run_test "Quickstarts Scala nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_quickstarts.sh scala"
EXIT_CODE=$?
fi
http://git-wip-us.apache.org/repos/asf/flink/blob/42e90c77/flink-end-to-end-tests/test-scripts/elasticsearch-common.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/elasticsearch-common.sh b/flink-end-to-end-tests/test-scripts/elasticsearch-common.sh
index 0ef6d55..87ffa82 100644
--- a/flink-end-to-end-tests/test-scripts/elasticsearch-common.sh
+++ b/flink-end-to-end-tests/test-scripts/elasticsearch-common.sh
@@ -56,13 +56,14 @@ function verify_elasticsearch_process_exist {
function verify_result {
local numRecords=$1
+ local index=$2
if [ -f "$TEST_DATA_DIR/output" ]; then
rm $TEST_DATA_DIR/output
fi
while : ; do
- curl 'localhost:9200/index/_search?q=*&pretty&size=21' > $TEST_DATA_DIR/output
+ curl "localhost:9200/${index}/_search?q=*&pretty&size=21" > $TEST_DATA_DIR/output
if [ -n "$(grep "\"total\" : $numRecords" $TEST_DATA_DIR/output)" ]; then
echo "Elasticsearch end to end test pass."
@@ -75,6 +76,8 @@ function verify_result {
}
function shutdown_elasticsearch_cluster {
+ local index=$1
+ curl -X DELETE "http://localhost:9200/${index}"
pid=$(jps | grep Elasticsearch | awk '{print $1}')
kill -9 $pid
}
http://git-wip-us.apache.org/repos/asf/flink/blob/42e90c77/flink-end-to-end-tests/test-scripts/test_quickstarts.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_quickstarts.sh b/flink-end-to-end-tests/test-scripts/test_quickstarts.sh
index 3db64fc..9657938 100755
--- a/flink-end-to-end-tests/test-scripts/test_quickstarts.sh
+++ b/flink-end-to-end-tests/test-scripts/test_quickstarts.sh
@@ -18,45 +18,59 @@
################################################################################
# End to end test for quick starts test.
+# Usage:
+# FLINK_DIR=<flink dir> flink-end-to-end-tests/test-scripts/test_quickstarts.sh <Type (java or scala)>
source "$(dirname "$0")"/common.sh
source "$(dirname "$0")"/elasticsearch-common.sh
-mkdir -p $TEST_DATA_DIR
+TEST_TYPE=$1
+TEST_CLASS_NAME=Elasticsearch5SinkExample
+TEST_FILE_PATH=flink-quickstart-test/src/main/${TEST_TYPE}/org/apache/flink/quickstarts/test/${TEST_CLASS_NAME}.${TEST_TYPE}
+QUICKSTARTS_FILE_PATH=${TEST_DATA_DIR}/flink-quickstart-${TEST_TYPE}/src/main/${TEST_TYPE}/org/apache/flink/quickstart/${TEST_CLASS_NAME}.${TEST_TYPE}
+ES_INDEX=index_${TEST_TYPE}
-cd $TEST_DATA_DIR
-
-mvn archetype:generate \
- -DarchetypeGroupId=org.apache.flink \
- -DarchetypeArtifactId=flink-quickstart-java \
- -DarchetypeVersion=1.6-SNAPSHOT \
- -DgroupId=org.apache.flink.quickstart \
- -DartifactId=flink-java-project \
- -Dversion=0.1 \
- -Dpackage=org.apache.flink.quickstart \
+# get the elasticsearch dependency from flink-quickstart-test
+ES_DEPENDENCY="<dependency>\
+<groupId>org.apache.flink</groupId>\
+$(awk '/flink-connector-elasticsearch/ {print $1}' ${END_TO_END_DIR}/flink-quickstart-test/target/dependency-reduced-pom.xml)\
+<version>\${flink.version}</version>\
+</dependency>"
+
+mkdir -p "${TEST_DATA_DIR}"
+cd "${TEST_DATA_DIR}"
+
+ARTIFACT_ID=flink-quickstart-${TEST_TYPE}
+ARTIFACT_VERSION=0.1
+
+mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-${TEST_TYPE} \
+ -DarchetypeVersion=1.6-SNAPSHOT \
+ -DgroupId=org.apache.flink.quickstart \
+ -DartifactId=${ARTIFACT_ID} \
+ -Dversion=${ARTIFACT_VERSION} \
+ -Dpackage=org.apache.flink.quickstart \
-DinteractiveMode=false
-cd flink-java-project
+cd "${ARTIFACT_ID}"
-# use the Flink Elasticsearch sink example job code in flink-elasticsearch5-tests to simulate modifications to contained job
-cp ${END_TO_END_DIR}/flink-elasticsearch5-test/src/main/java/org/apache/flink/streaming/tests/Elasticsearch5SinkExample.java $TEST_DATA_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/
-sed -i -e 's/package org.apache.flink.streaming.tests;/package org.apache.flink.quickstart;/' $TEST_DATA_DIR/flink-java-project/src/main/java/org/apache/flink/quickstart/Elasticsearch5SinkExample.java
+# use the Flink Elasticsearch sink example job code in flink-quickstart-test to simulate modifications to contained job
+cp ${END_TO_END_DIR}/${TEST_FILE_PATH} "$QUICKSTARTS_FILE_PATH"
+sed -i -e 's/package org.apache.flink.quickstarts.test/package org.apache.flink.quickstart/' "${QUICKSTARTS_FILE_PATH}"
position=$(awk '/<dependencies>/ {print NR}' pom.xml | head -1)
-sed -i -e ''"$(($position + 1))"'i\
-<dependency>\
-<groupId>org.apache.flink</groupId>\
-<artifactId>flink-connector-elasticsearch5_${scala.binary.version}</artifactId>\
-<version>${flink.version}</version>\
-</dependency>' pom.xml
+# Add ElasticSearch dependency to pom.xml
+sed -i -e ''$(($position + 1))'i\
+'${ES_DEPENDENCY}'' pom.xml
-sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.streaming.tests.Elasticsearch5SinkExample/" pom.xml
+sed -i -e "s/org.apache.flink.quickstart.StreamingJob/org.apache.flink.quickstart.$TEST_CLASS_NAME/" pom.xml
mvn clean package -nsu
cd target
-jar tvf flink-java-project-0.1.jar > contentsInJar.txt
+jar tvf flink-quickstart-${TEST_TYPE}-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' && \
@@ -86,21 +100,21 @@ setup_elasticsearch "https://artifacts.elastic.co/downloads/elasticsearch/elasti
verify_elasticsearch_process_exist
function shutdownAndCleanup {
- shutdown_elasticsearch_cluster
+ shutdown_elasticsearch_cluster "$ES_INDEX"
# make sure to run regular cleanup as well
cleanup
}
trap shutdownAndCleanup INT
trap shutdownAndCleanup EXIT
-TEST_PROGRAM_JAR=$TEST_DATA_DIR/flink-java-project/target/flink-java-project-0.1.jar
+TEST_PROGRAM_JAR=${TEST_DATA_DIR}/${ARTIFACT_ID}/target/${ARTIFACT_ID}-${ARTIFACT_VERSION}.jar
start_cluster
-$FLINK_DIR/bin/flink run -c org.apache.flink.quickstart.Elasticsearch5SinkExample $TEST_PROGRAM_JAR \
+${FLINK_DIR}/bin/flink run -c org.apache.flink.quickstart.Elasticsearch5SinkExample "$TEST_PROGRAM_JAR" \
--numRecords 20 \
- --index index \
+ --index "${ES_INDEX}" \
--type type
-verify_result 20
+verify_result 20 "${ES_INDEX}"
http://git-wip-us.apache.org/repos/asf/flink/blob/42e90c77/flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh b/flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh
index 10e72e9..7464409 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh
@@ -31,7 +31,7 @@ verify_elasticsearch_process_exist
start_cluster
function test_cleanup {
- shutdown_elasticsearch_cluster
+ shutdown_elasticsearch_cluster index
# make sure to run regular cleanup as well
cleanup
@@ -48,4 +48,4 @@ $FLINK_DIR/bin/flink run -p 1 $TEST_ES_JAR \
--index index \
--type type
-verify_result 20
+verify_result 20 index