[FLINK-11745][State TTL][E2E] Restore from the savepoint after the jo…

…b cancellation.

This closes apache#7824.

flink-end-to-end-tests/test-scripts/test_stream_state_ttl.sh

@@ -22,9 +22,8 @@ source "$(dirname "$0")"/common.sh
 STATE_BACKEND_TYPE="${1:-file}"
 STATE_BACKEND_FILE_ASYNC="${2:-true}"
 TTL="${3:-1000}"
-PRECISION="${4:-5}"
-PARALLELISM="${5-3}"
-UPDATE_NUM="${6-1000}"
+PARALLELISM="${4-3}"
+UPDATE_NUM="${5-1000}"
 
 CHECKPOINT_DIR="file:https://$TEST_DATA_DIR/savepoint-e2e-test-chckpt-dir"
 
@@ -48,15 +47,19 @@ trap test_cleanup EXIT
 set_conf "metrics.fetcher.update-interval" "2000"
 
 start_cluster
-start_taskmanagers $PARALLELISM
+if [ "${PARALLELISM}" -gt "1" ]; then
+ start_taskmanagers $(expr ${PARALLELISM} - 1)
+fi
 
 function job_id() {
- CMD="${FLINK_DIR}/bin/flink run -d -p ${PARALLELISM} ${TEST_PROGRAM_JAR} \
+ if [ -n "$1" ]; then
+ SP="-s $1"
+ fi
+ CMD="${FLINK_DIR}/bin/flink run -d ${SP} -p ${PARALLELISM} ${TEST_PROGRAM_JAR} \
  --test.semantics exactly-once \
  --environment.parallelism ${PARALLELISM} \
  --state_backend ${STATE_BACKEND_TYPE} \
  --state_ttl_verifier.ttl_milli ${TTL} \
- --state_ttl_verifier.precision_milli ${PRECISION} \
  --state_backend.checkpoint_directory ${CHECKPOINT_DIR} \
  --state_backend.file.async ${STATE_BACKEND_FILE_ASYNC} \
  --update_generator_source.sleep_time 10 \
@@ -75,7 +78,7 @@ SAVEPOINT_PATH=$(take_savepoint ${JOB} ${TEST_DATA_DIR} \
 
 cancel_job ${JOB}
 
-JOB_CMD=$(job_id)
+JOB_CMD=$(job_id ${SAVEPOINT_PATH})
 echo ${JOB_CMD}
 JOB=$(${JOB_CMD} | grep 'Job has been submitted with JobID' | sed 's/.* //g')
 wait_job_running ${JOB}