[flink] 03/03: [FLINK-15793][e2e] Add E2E test to enable plugin for native K8s

[rm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

rmetzger pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 5db513770b5b3205b4948ecd8ffdc6327177ad52
Author: wangyang0918 <da...@alibaba-inc.com>
AuthorDate: Tue Aug 18 15:48:29 2020 +0800

    [FLINK-15793][e2e] Add E2E test to enable plugin for native K8s
---
 flink-end-to-end-tests/run-nightly-tests.sh        |  3 ++-
 .../test-scripts/common_dummy_fs.sh                | 12 +++++++--
 .../test-scripts/test_kubernetes_session.sh        | 29 +++++++++++++++++++---
 3 files changed, 38 insertions(+), 6 deletions(-)

diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh
index a2571e6..d879af3 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -125,7 +125,8 @@ if [[ ${PROFILE} != *"jdk11"* ]]; then
 	run_test "Wordcount on Docker test (custom fs plugin)" "$END_TO_END_DIR/test-scripts/test_docker_embedded_job.sh dummy-fs"
 
 	run_test "Run Kubernetes test" "$END_TO_END_DIR/test-scripts/test_kubernetes_embedded_job.sh"
-	run_test "Run kubernetes session test" "$END_TO_END_DIR/test-scripts/test_kubernetes_session.sh"
+	run_test "Run kubernetes session test (default input)" "$END_TO_END_DIR/test-scripts/test_kubernetes_session.sh"
+	run_test "Run kubernetes session test (custom fs plugin)" "$END_TO_END_DIR/test-scripts/test_kubernetes_session.sh dummy-fs"
 	run_test "Run kubernetes application test" "$END_TO_END_DIR/test-scripts/test_kubernetes_application.sh"
 
 	run_test "Running Flink over NAT end-to-end test" "$END_TO_END_DIR/test-scripts/test_nat.sh" "skip_check_exceptions"
diff --git a/flink-end-to-end-tests/test-scripts/common_dummy_fs.sh b/flink-end-to-end-tests/test-scripts/common_dummy_fs.sh
index aba7e51..d025619 100644
--- a/flink-end-to-end-tests/test-scripts/common_dummy_fs.sh
+++ b/flink-end-to-end-tests/test-scripts/common_dummy_fs.sh
@@ -17,9 +17,17 @@
 # limitations under the License.
 ################################################################################
 
+DUMMY_FS_PATH="${END_TO_END_DIR}/flink-plugins-test/dummy-fs/target/flink-dummy-fs.jar"
+ANOTHER_DUMMY_FS_PATH="${END_TO_END_DIR}/flink-plugins-test/another-dummy-fs/target/flink-another-dummy-fs.jar"
+
 function dummy_fs_setup() {
     mkdir -p "$FLINK_DIR/plugins/dummy-fs"
     mkdir -p "$FLINK_DIR/plugins/another-dummy-fs"
-    cp "${END_TO_END_DIR}/flink-plugins-test/dummy-fs/target/flink-dummy-fs.jar" "${FLINK_DIR}/plugins/dummy-fs/"
-    cp "${END_TO_END_DIR}/flink-plugins-test/another-dummy-fs/target/flink-another-dummy-fs.jar" "${FLINK_DIR}/plugins/another-dummy-fs/"
+    cp "$DUMMY_FS_PATH" "${FLINK_DIR}/plugins/dummy-fs/"
+    cp "$ANOTHER_DUMMY_FS_PATH" "${FLINK_DIR}/plugins/another-dummy-fs/"
+}
+
+function cp_dummy_fs_to_opt() {
+    cp "$DUMMY_FS_PATH" "${FLINK_DIR}/opt/"
+    cp "$ANOTHER_DUMMY_FS_PATH" "${FLINK_DIR}/opt/"
 }
diff --git a/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh b/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
index 7ccdc8d..f0a580f 100755
--- a/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
+++ b/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
@@ -24,7 +24,26 @@ CLUSTER_ID="flink-native-k8s-session-1"
 FLINK_IMAGE_NAME="test_kubernetes_session"
 LOCAL_OUTPUT_PATH="${TEST_DATA_DIR}/out/wc_out"
 OUTPUT_PATH="/tmp/wc_out"
-ARGS="--output ${OUTPUT_PATH}"
+OUTPUT_ARGS="--output ${OUTPUT_PATH}"
+
+INPUT_TYPE=${1:-embedded}
+case $INPUT_TYPE in
+    (embedded)
+        INPUT_ARGS=""
+    ;;
+    (dummy-fs)
+        source "$(dirname "$0")"/common_dummy_fs.sh
+        cp_dummy_fs_to_opt
+        INPUT_ARGS="--input dummy://localhost/words --input anotherDummy://localhost/words"
+        RESULT_HASH="0e5bd0a3dd7d5a7110aa85ff70adb54b"
+        ENABLE_DUMMPY_FS_ARGS="-Dcontainerized.master.env.ENABLE_BUILT_IN_PLUGINS=flink-dummy-fs.jar;flink-another-dummy-fs.jar \
+        -Dcontainerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS=flink-dummy-fs.jar;flink-another-dummy-fs.jar"
+    ;;
+    (*)
+        echo "Unknown input type $INPUT_TYPE"
+        exit 1
+    ;;
+esac
 
 function internal_cleanup {
     kubectl delete deployment ${CLUSTER_ID}
@@ -39,13 +58,17 @@ kubectl create clusterrolebinding ${CLUSTER_ROLE_BINDING} --clusterrole=edit --s
 
 mkdir -p "$(dirname $LOCAL_OUTPUT_PATH)"
 
+# Enable dummy fs for Flink client
+[[ $INPUT_TYPE = 'dummy-fs' ]] && dummy_fs_setup
+
 # Set the memory and cpu smaller than default, so that the jobmanager and taskmanager pods could be allocated in minikube.
 "$FLINK_DIR"/bin/kubernetes-session.sh -Dkubernetes.cluster-id=${CLUSTER_ID} \
     -Dkubernetes.container.image=${FLINK_IMAGE_NAME} \
     -Djobmanager.memory.process.size=1088m \
     -Dkubernetes.jobmanager.cpu=0.5 \
     -Dkubernetes.taskmanager.cpu=0.5 \
-    -Dkubernetes.rest-service.exposed.type=NodePort
+    -Dkubernetes.rest-service.exposed.type=NodePort \
+    $ENABLE_DUMMPY_FS_ARGS
 
 kubectl wait --for=condition=Available --timeout=30s deploy/${CLUSTER_ID} || exit 1
 jm_pod_name=$(kubectl get pods --selector="app=${CLUSTER_ID},component=jobmanager" -o jsonpath='{..metadata.name}')
@@ -53,7 +76,7 @@ wait_rest_endpoint_up_k8s $jm_pod_name
 
 "$FLINK_DIR"/bin/flink run -e kubernetes-session \
     -Dkubernetes.cluster-id=${CLUSTER_ID} \
-    ${FLINK_DIR}/examples/batch/WordCount.jar ${ARGS}
+    ${FLINK_DIR}/examples/batch/WordCount.jar ${INPUT_ARGS} ${OUTPUT_ARGS}
 
 if ! check_logs_output $jm_pod_name 'Starting KubernetesSessionClusterEntrypoint'; then
   echo "JobManager logs are not accessible via kubectl logs."