[FLINK-10986][tests] Enable fine-grained configuration of required dbs

This enables toggling the setup of test dependencies, such as Hadooop, Mesos,
and, ZooKeeper through the --test-spec edn file. The type of the Flink cluster
can also be specified via: :flink-yarn-job :flink-yarn-session
:flink-mesos-session, and :flink-standalone-session.

Retryable operations that exhausted all attempts, now propagate the exception by
default. This is to fail fast if during the db/setup!, an operation, such as
Flink job submission, fails to complete.

flink-jepsen/docker/run-tests.sh

@@ -29,26 +29,25 @@ n3
 EOF
 
 common_jepsen_args+=(--ha-storage-dir hdfs:https:///flink
---test-spec "${dockerdir}/data-stream-test-program-parallelism-1.edn"
 --tarball ${2}
 --ssh-private-key ~/.ssh/id_rsa
 --nodes-file ${dockerdir}/nodes)
 
 for i in $(seq 1 ${1})
 do
  echo "Executing run #${i} of ${1}"
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --deployment-mode yarn-session
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --deployment-mode yarn-session
- lein run test "${common_jepsen_args[@]}" --nemesis-gen fail-name-node-during-recovery --deployment-mode yarn-session
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --test-spec "${dockerdir}/test-specs/yarn-session.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --test-spec "${dockerdir}/test-specs/yarn-session.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen fail-name-node-during-recovery --test-spec "${dockerdir}/test-specs/yarn-session.edn"
 
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --deployment-mode yarn-job
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --deployment-mode yarn-job
- lein run test "${common_jepsen_args[@]}" --nemesis-gen fail-name-node-during-recovery --deployment-mode yarn-job
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --test-spec "${dockerdir}/test-specs/yarn-job.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --test-spec "${dockerdir}/test-specs/yarn-job.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen fail-name-node-during-recovery --test-spec "${dockerdir}/test-specs/yarn-job.edn"
 
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --deployment-mode mesos-session
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --deployment-mode mesos-session
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-task-managers --test-spec "${dockerdir}/test-specs/mesos-session.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --test-spec "${dockerdir}/test-specs/mesos-session.edn"
 
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --deployment-mode standalone-session
- lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --client-gen cancel-jobs --deployment-mode standalone-session
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --test-spec "${dockerdir}/standalone-session.edn"
+ lein run test "${common_jepsen_args[@]}" --nemesis-gen kill-job-managers --client-gen cancel-jobs --test-spec "${dockerdir}/standalone-session.edn"
  echo
 done

flink-jepsen/docker/test-specs/mesos-session.edn

@@ -0,0 +1,19 @@
+;; 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:https://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.
+
+{:dbs [:hadoop :zookeeper :mesos :flink-mesos-session]
+ :jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
+ :job-args "--environment.parallelism 1 --state_backend.checkpoint_directory hdfs:https:///checkpoints --state_backend rocks --state_backend.rocks.incremental true"}]}

flink-jepsen/docker/test-specs/standalone-session.edn

@@ -0,0 +1,19 @@
+;; 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:https://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.
+
+{:dbs [:hadoop :zookeeper :flink-standalone-session]
+ :jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
+ :job-args "--environment.parallelism 1 --state_backend.checkpoint_directory hdfs:https:///checkpoints --state_backend rocks --state_backend.rocks.incremental true"}]}

flink-jepsen/docker/data-stream-test-program-parallelism-1.edn → flink-jepsen/docker/test-specs/yarn-job.edn

@@ -14,5 +14,6 @@
 ;; See the License for the specific language governing permissions and
 ;; limitations under the License.
 
-{:jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
+{:dbs [:hadoop :zookeeper :flink-yarn-job]
+ :jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
  :job-args "--environment.parallelism 1 --state_backend.checkpoint_directory hdfs:https:///checkpoints --state_backend rocks --state_backend.rocks.incremental true"}]}

flink-jepsen/docker/test-specs/yarn-session.edn

@@ -0,0 +1,19 @@
+;; 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:https://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.
+
+{:dbs [:hadoop :zookeeper :flink-yarn-session]
+ :jobs [{:job-jar "/jepsen/bin/DataStreamAllroundTestProgram.jar"
+ :job-args "--environment.parallelism 1 --state_backend.checkpoint_directory hdfs:https:///checkpoints --state_backend rocks --state_backend.rocks.incremental true"}]}

flink-jepsen/src/jepsen/flink/checker.clj

@@ -211,7 +211,7 @@
 (defn- compute-final-model
  [model history]
  (let [start-time (-> history first :time)]
- (reduce model/step
+ (reduce knossos.model/step
  (assoc model :last-failure start-time)
  history)))
 

flink-jepsen/src/jepsen/flink/client.clj

@@ -172,12 +172,11 @@
  :cancel-jobs (dispatch-operation-or-fatal op (cancel-jobs! rest-url job-ids)))))
 
 (defrecord Client
- [deploy-cluster! ; function that starts a non-standalone cluster and submits the jobs
- closer ; function that closes the ZK client
+ [closer ; function that closes the ZK client
  rest-url ; atom storing the current rest-url
  init-future ; future that completes if rest-url is set to an initial value
  job-ids ; atom storing the job-ids
- job-submitted?] ; Have the jobs already been submitted? Used to avoid re-submission if the client is re-opened.
+ client-setup?]  ; Has the client already been setup? Used to avoid running setup! again if the client is re-opened.
  client/Client
  (open! [this test _]
  (info "Open client.")
@@ -186,10 +185,9 @@
  :rest-url rest-url-atom
  :init-future init-future)))
 
- (setup! [_ test]
+ (setup! [_ _]
  (info "Setup client.")
- (when (compare-and-set! job-submitted? false true)
- (deploy-cluster! test)
+ (when (compare-and-set! client-setup? false true)
  (deref init-future)
  (let [jobs (fu/retry (fn [] (list-jobs! @rest-url))
  :fallback (fn [e]
@@ -209,5 +207,5 @@
  (closer)))
 
 (defn create-client
- [deploy-cluster!]
- (Client. deploy-cluster! nil nil nil (atom nil) (atom false)))
+ []
+ (Client. nil nil nil (atom nil) (atom false)))