[FLINK-13193][tests] Enable custom flink cluster config per test

Enable users to set a cluster configuration in the test spec which will have
precedence over the default configuration.

Remove --ha-storage-dir parameter because the user can set this value in the
test spec cluster configuration now.

This closes apache#9070.

flink-jepsen/docker/run-tests.sh

@@ -28,7 +28,7 @@ n2
 n3
 EOF
 
-common_jepsen_args+=(--ha-storage-dir hdfs:https:///flink
+common_jepsen_args+=(
 --tarball ${2}
 --ssh-private-key ~/.ssh/id_rsa
 --nodes-file ${dockerdir}/nodes)

flink-jepsen/src/jepsen/flink/db.clj

@@ -36,13 +36,13 @@
 
 (def taskmanager-slots 3)
 
-(defn flink-configuration
+(defn- default-flink-configuration
  [test node]
  {:high-availability "zookeeper"
  :high-availability.zookeeper.quorum (zookeeper-quorum test)
- :high-availability.storageDir (str (:ha-storage-dir test) "/ha")
+ :high-availability.storageDir "hdfs:https:///flink/ha"
  :jobmanager.rpc.address node
- :state.savepoints.dir (str (:ha-storage-dir test) "/savepoints")
+ :state.savepoints.dir "hdfs:https:///flink/savepoints"
  :rest.address node
  :rest.port 8081
  :rest.bind-address "0.0.0.0"
@@ -52,6 +52,12 @@
  :state.backend.local-recovery "true"
  :taskmanager.registration.timeout "30 s"})
 
+(defn flink-configuration
+ [test node]
+ (let [additional-config (-> test :test-spec :flink-config)]
+ (merge (default-flink-configuration test node)
+ additional-config)))
+
 (defn write-configuration!
  "Writes the flink-conf.yaml to the flink conf directory"
  [test node]

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

@@ -120,7 +120,6 @@
  :parse-fn read-test-spec
  :validate [#(->> % :dbs (map dbs) (every? (complement nil?)))
  (str "Invalid :dbs specification. " (keys->allowed-values-help-text dbs))]]
- [nil "--ha-storage-dir DIR" "high-availability.storageDir"]
  [nil "--nemesis-gen GEN" (str "Which nemesis should be used?"
  (keys->allowed-values-help-text fn/nemesis-generator-factories))
  :parse-fn keyword

flink-jepsen/test/jepsen/flink/db_test.clj

@@ -0,0 +1,29 @@
+;; 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.
+
+(ns jepsen.flink.db-test
+ (:require [clojure.test :refer :all])
+ (:require [jepsen.flink.db :refer :all]))
+
+(deftest flink-configuration-test
+ (testing "High availability is zookeeper by default"
+ (is (= "zookeeper" (:high-availability (flink-configuration {} "n1")))))
+
+ (testing "Default configuration can be overridden"
+ (let [expected-config-value "NONE"
+ custom-flink-config {:high-availability expected-config-value}
+ test {:test-spec {:flink-config custom-flink-config}}]
+ (is (= expected-config-value (:high-availability (flink-configuration test "n1")))))))