flink-jepsen

A Clojure project based on the Jepsen framework to find bugs in the distributed coordination of Apache Flink®.

Test Coverage

Jepsen is a framework built to test the behavior of distributed systems under faults. The tests in this particular project deploy Flink on YARN, Mesos, or as a standalone cluster, submit one or multiple jobs, and examine the availability of the job(s) after injecting faults. Optionally, we can cancel the job(s) during the test. The faults that can be currently introduced to the Flink cluster include:

• Killing of TaskManager/JobManager processes
• Stopping HDFS NameNode
• Network partitions

Checking Correctness

We define a job to be available if all the tasks of the job are running. Our correctness model prescribes that:

• Jobs should become available within the job recovery grace period after the last injected fault. Note that some faults happen at a single point in time (e.g., killing of processes). Other faults, such as network splits, happen during a period of time, and can thus be interleaving. As long as there are active faults, jobs are allowed to be unavailable.
• If jobs are canceled, they must become unavailable within 10 seconds of the cancellation.

Usage

Setting up the Environment

A Jepsen test is started and orchestrated by a Clojure program running on a control node. That program logs into several specified DB nodes to set up a Flink cluster and submit test job(s). Afterwards, the control node will inject faults into the DB nodes and continue to interact with the Flink cluster (for example by probing the status of the test job). Tests in this project require at least 3 DB nodes; this is due to how we install software, such as HDFS, on the DB nodes. There is no upper limit on the number of DB nodes.

For details on how to set up the environment required to run the tests, see the Jepsen documentation. To simplify development, we have prepared Dockerfiles and a Docker Compose template so that you can run the tests locally in containers (see Section Docker).

Running Tests

Because tests are started from the control node, the contents of this directory must be copied to the control node first. An example for a command that will run a test is

lein run test --tarball http://             \
--ssh-private-key ~/.ssh/id_rsa             \
--nodes-file nodes                          \
--nemesis-gen kill-task-managers            \
--test-spec test-specs/yarn-session.edn

Here is a breakdown of the command line options and their arguments:

• --tarball A URL to the Flink distribution under test
• --ssh-private-key Path to the private key that will be used to connect to the DB nodes (from the control node)
• --nodes-file Path to a file containing newline separated IPs or hostnames of the DB nodes
• --nemesis-gen Specifies the faults that will be injected. kill-task-managers means for the duration of the test (default 60s), TaskManagers will be killed on all DB nodes with a uniform random timing noise with a mean delay of 3 seconds (delay ranging from 0 to 6 seconds).
• --test-spec Path to an edn formatted file containing a specification about which software to install and which jobs to submit. For example, by using the file below, we would request the framework to install Hadoop (YARN, HDFS), ZooKeeper, and deploy Flink as a session cluster on YARN. Furthermore, we instruct to submit the DataStreamAllroundTestProgram to Flink with the specified job arguments.

  {:dbs  [:hadoop :zookeeper :flink-yarn-session]
   :jobs [{:job-jar  "/path/to/DataStreamAllroundTestProgram.jar"
           :job-args "--environment.parallelism 1 --state_backend.checkpoint_directory hdfs:///checkpoints --state_backend rocks --state_backend.rocks.incremental true"}]}
  

For more details on the command line options, see Section Command Line Options & Configuration.

As you can see, this project does not comprise only a single test but rather a parameterizable test template. The script under docker/run-tests.sh shows more examples on how to parameterize and run tests. The example tests can be used as a starting point for testing releases or patches that are modifying critical components of Flink. Most examples submit the DataStreamAllroundTestProgram to the Flink cluster under test. The job must be build before running the tests, for example by running

$ mvn clean package -pl flink-end-to-end-tests/flink-datastream-allround-test -am -DskipTests -Dfast

from Flink's project root.

Also included in the provided test suite is a more complicated scenario with two jobs that share a Kafka topic. See the docker/run-tests.sh script for details on how to enable and run this test.

Docker

Firstly, build the test job located under flink-end-to-end-tests/flink-datastream-allround-test and copy the resulting jar (DataStreamAllroundTestProgram.jar) to the ./bin directory of this project's root.

To build the images and start the containers, simply run:

$ cd docker
$ ./up.sh

This should start one control node container and three containers that will be used as DB nodes. After the containers have started, open a new terminal window and run docker exec -it jepsen-control bash. This will allow you to run arbitrary commands on the control node. To start the tests, you can use the run-tests.sh script in the docker directory, which expects the number of test iterations, and a URI to a Flink distribution, e.g.,

./docker/run-tests.sh 1 https://example.com/flink-dist.tgz

The project‘s root is mounted as a volume to all containers under the path /jepsen. This means that changes to the test sources are immediately reflected in the control node container. Moreover, this allows you to test locally built Flink distributions by copying the tarball to the project’s root and passing a URI with the file:// scheme to the run-tests.sh script, e.g., file:///jepsen/flink-dist.tgz.

Memory Requirements

The tests have high memory demands due to the many processes that are started by the control node. For example, to test Flink on YARN in a HA setup, we require ZooKeeper, HDFS NameNode, HDFS DataNode, YARN NodeManager, and YARN ResourceManager, in addition to the Flink processes. We found that the tests can be run comfortably in Docker containers on a machine with 32 GiB RAM.

Checking the Output of Tests

Consult the jepsen.log file for the particular test run in the store folder. The final output of every test will be either

Everything looks good! ヽ('ー`)ノ

or

Analysis invalid! (ﾉಥ益ಥ）ﾉ ┻━┻

depending on whether the test passed or not. If neither output is generated, the test did not finish properly due to problems of the environment, bugs in Jepsen or in the test suite, etc.

In addition, the test directories contain all relevant log files, and the jstack output for all Flink JVMs aggregated from the DB nodes.

Command Line Options & Configuration

Below is a listing of the most important command line options that can be passed to lein run test.

Nemesis

The nemesis or more accurately the nemesis generator determines the order and timing of the injected faults.

Client

The client interacts with the Flink cluster during the test; its result history will be used to validate whether the Flink cluster behaved correctly.

Test Spec

The test spec is an edn formatted file containing a specification about which software to install on the DB nodes and which jobs to submit. Additionally the user can choose to override the default Flink configuration in the test spec. Below is a complete example for a test spec file.

{:dbs          [:hadoop :zookeeper :flink-yarn-session]
 :jobs         [{:job-jar    "/path/to/FlinkJobs.jar"
                 :job-args   "--parameter1 arg1 --parmeter2 arg2"
                 :main-class "org.apache.flink.streaming.tests.Job1"}

                 {:job-jar    "/path/to/FlinkJobs.jar"
                  :job-args   "--parameter1 arg1 --parmeter2 arg2"
                  :main-class "org.apache.flink.streaming.tests.Job2"}]

 :flink-config {:jobmanager.execution.failover-strategy "region"}}

The test spec is a Clojure map; its set of allowed keys is described by the table below.

The set of possible dbs is described by the table below. Note that when we speak of the first node, the second node, nth node, we refer to the first, second, nth item in the nodes file in lexicographically sorted order.