We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] multistep_workflow "best-effort detection of 'already run'" fails to detect run_info has FINISHED status
Thank you for submitting an issue. Please refer to our [issue policy](https://www.github.com/mlflow/mlflow/blob/master/ISSUE_POLICY.md)
for information on what types of issues we address. For help with debugging your code, please refer to [Stack Overflow](https://stackoverflow.com/questions/tagged/mlflow).

Please fill in this template and do not delete it unless you are sure your issue is outside its scope.

### System information
- **Have I written custom code (as opposed to using a stock example script provided in MLflow)**: No
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Linux Ubuntu 16.04
- **MLflow installed from (source or binary)**: pip install mlflow
- **MLflow version (run ``mlflow --version``)**: mlflow, version 1.3.0
- **Python version**: Python 3.5.6 :: Anaconda, Inc.
- **npm version, if running the dev UI**:
- **Exact command to reproduce**: 
```from mlflow.entities import RunStatus; type(RunStatus.ACTIVE) == int```

### Describe the problem
Describe the problem clearly here. Include descriptions of the expected behavior and the actual behavior.

The check on the run_info.status is finished fails because run_info.status is of type `str` equal to `FINISHED` but `mlflow.entities.RunStatus` is actually of `int` type.

### Code to reproduce issue
Provide a reproducible test case that is the bare minimum necessary to generate the problem.

1. `mlflow run https://github.com/mlflow/mlflow#examples/multistep_workflow -e load_raw_data`
2. Allow to complete and verify it is actually FINISHED
3. `mlflow run https://github.com/mlflow/mlflow#examples/multistep_workflow`
4. Notice the print statement saying
> Run matched, but is not FINISHED, so skipping (run_id=91cab91e68cf4f45b3c88781fc7000fb, status=FINISHED)


### Other info / logs
Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks,
please include the full traceback. Large logs and files should be attached.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `examples/multistep_workflow/main.py`
Content:
```
1 """
2 Downloads the MovieLens dataset, ETLs it into Parquet, trains an
3 ALS model, and uses the ALS model to train a Keras neural network.
4 
5 See README.rst for more details.
6 """
7 
8 import click
9 import os
10 
11 
12 import mlflow
13 from mlflow.utils import mlflow_tags
14 from mlflow.entities import RunStatus
15 from mlflow.utils.logging_utils import eprint
16 import six
17 
18 from mlflow.tracking.fluent import _get_experiment_id
19 
20 
21 def _already_ran(entry_point_name, parameters, git_commit, experiment_id=None):
22     """Best-effort detection of if a run with the given entrypoint name,
23     parameters, and experiment id already ran. The run must have completed
24     successfully and have at least the parameters provided.
25     """
26     experiment_id = experiment_id if experiment_id is not None else _get_experiment_id()
27     client = mlflow.tracking.MlflowClient()
28     all_run_infos = reversed(client.list_run_infos(experiment_id))
29     for run_info in all_run_infos:
30         full_run = client.get_run(run_info.run_id)
31         tags = full_run.data.tags
32         if tags.get(mlflow_tags.MLFLOW_PROJECT_ENTRY_POINT, None) != entry_point_name:
33             continue
34         match_failed = False
35         for param_key, param_value in six.iteritems(parameters):
36             run_value = full_run.data.params.get(param_key)
37             if run_value != param_value:
38                 match_failed = True
39                 break
40         if match_failed:
41             continue
42 
43         if run_info.status != RunStatus.FINISHED:
44             eprint(("Run matched, but is not FINISHED, so skipping "
45                     "(run_id=%s, status=%s)") % (run_info.run_id, run_info.status))
46             continue
47 
48         previous_version = tags.get(mlflow_tags.MLFLOW_GIT_COMMIT, None)
49         if git_commit != previous_version:
50             eprint(("Run matched, but has a different source version, so skipping "
51                     "(found=%s, expected=%s)") % (previous_version, git_commit))
52             continue
53         return client.get_run(run_info.run_id)
54     eprint("No matching run has been found.")
55     return None
56 
57 
58 # TODO(aaron): This is not great because it doesn't account for:
59 # - changes in code
60 # - changes in dependant steps
61 def _get_or_run(entrypoint, parameters, git_commit, use_cache=True):
62     existing_run = _already_ran(entrypoint, parameters, git_commit)
63     if use_cache and existing_run:
64         print("Found existing run for entrypoint=%s and parameters=%s" % (entrypoint, parameters))
65         return existing_run
66     print("Launching new run for entrypoint=%s and parameters=%s" % (entrypoint, parameters))
67     submitted_run = mlflow.run(".", entrypoint, parameters=parameters)
68     return mlflow.tracking.MlflowClient().get_run(submitted_run.run_id)
69 
70 
71 @click.command()
72 @click.option("--als-max-iter", default=10, type=int)
73 @click.option("--keras-hidden-units", default=20, type=int)
74 @click.option("--max-row-limit", default=100000, type=int)
75 def workflow(als_max_iter, keras_hidden_units, max_row_limit):
76     # Note: The entrypoint names are defined in MLproject. The artifact directories
77     # are documented by each step's .py file.
78     with mlflow.start_run() as active_run:
79         os.environ['SPARK_CONF_DIR'] = os.path.abspath('.')
80         git_commit = active_run.data.tags.get(mlflow_tags.MLFLOW_GIT_COMMIT)
81         load_raw_data_run = _get_or_run("load_raw_data", {}, git_commit)
82         ratings_csv_uri = os.path.join(load_raw_data_run.info.artifact_uri, "ratings-csv-dir")
83         etl_data_run = _get_or_run("etl_data",
84                                    {"ratings_csv": ratings_csv_uri,
85                                     "max_row_limit": max_row_limit},
86                                    git_commit)
87         ratings_parquet_uri = os.path.join(etl_data_run.info.artifact_uri, "ratings-parquet-dir")
88 
89         # We specify a spark-defaults.conf to override the default driver memory. ALS requires
90         # significant memory. The driver memory property cannot be set by the application itself.
91         als_run = _get_or_run("als",
92                               {"ratings_data": ratings_parquet_uri, "max_iter": str(als_max_iter)},
93                               git_commit)
94         als_model_uri = os.path.join(als_run.info.artifact_uri, "als-model")
95 
96         keras_params = {
97             "ratings_data": ratings_parquet_uri,
98             "als_model_uri": als_model_uri,
99             "hidden_units": keras_hidden_units,
100         }
101         _get_or_run("train_keras", keras_params, git_commit, use_cache=False)
102 
103 
104 if __name__ == '__main__':
105     workflow()
106
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/examples/multistep_workflow/main.py b/examples/multistep_workflow/main.py
--- a/examples/multistep_workflow/main.py
+++ b/examples/multistep_workflow/main.py
@@ -40,7 +40,7 @@
         if match_failed:
             continue
 
-        if run_info.status != RunStatus.FINISHED:
+        if run_info.to_proto().status != RunStatus.FINISHED:
             eprint(("Run matched, but is not FINISHED, so skipping "
                     "(run_id=%s, status=%s)") % (run_info.run_id, run_info.status))
             continue

{"golden_diff": "diff --git a/examples/multistep_workflow/main.py b/examples/multistep_workflow/main.py\n--- a/examples/multistep_workflow/main.py\n+++ b/examples/multistep_workflow/main.py\n@@ -40,7 +40,7 @@\n         if match_failed:\n             continue\n \n-        if run_info.status != RunStatus.FINISHED:\n+        if run_info.to_proto().status != RunStatus.FINISHED:\n             eprint((\"Run matched, but is not FINISHED, so skipping \"\n                     \"(run_id=%s, status=%s)\") % (run_info.run_id, run_info.status))\n             continue\n", "issue": "[BUG] multistep_workflow \"best-effort detection of 'already run'\" fails to detect run_info has FINISHED status\nThank you for submitting an issue. Please refer to our [issue policy](https://www.github.com/mlflow/mlflow/blob/master/ISSUE_POLICY.md)\r\nfor information on what types of issues we address. For help with debugging your code, please refer to [Stack Overflow](https://stackoverflow.com/questions/tagged/mlflow).\r\n\r\nPlease fill in this template and do not delete it unless you are sure your issue is outside its scope.\r\n\r\n### System information\r\n- **Have I written custom code (as opposed to using a stock example script provided in MLflow)**: No\r\n- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Linux Ubuntu 16.04\r\n- **MLflow installed from (source or binary)**: pip install mlflow\r\n- **MLflow version (run ``mlflow --version``)**: mlflow, version 1.3.0\r\n- **Python version**: Python 3.5.6 :: Anaconda, Inc.\r\n- **npm version, if running the dev UI**:\r\n- **Exact command to reproduce**: \r\n```from mlflow.entities import RunStatus; type(RunStatus.ACTIVE) == int```\r\n\r\n### Describe the problem\r\nDescribe the problem clearly here. Include descriptions of the expected behavior and the actual behavior.\r\n\r\nThe check on the run_info.status is finished fails because run_info.status is of type `str` equal to `FINISHED` but `mlflow.entities.RunStatus` is actually of `int` type.\r\n\r\n### Code to reproduce issue\r\nProvide a reproducible test case that is the bare minimum necessary to generate the problem.\r\n\r\n1. `mlflow run https://github.com/mlflow/mlflow#examples/multistep_workflow -e load_raw_data`\r\n2. Allow to complete and verify it is actually FINISHED\r\n3. `mlflow run https://github.com/mlflow/mlflow#examples/multistep_workflow`\r\n4. Notice the print statement saying\r\n> Run matched, but is not FINISHED, so skipping (run_id=91cab91e68cf4f45b3c88781fc7000fb, status=FINISHED)\r\n\r\n\r\n### Other info / logs\r\nInclude any logs or source code that would be helpful to diagnose the problem. If including tracebacks,\r\nplease include the full traceback. Large logs and files should be attached.\r\n\n", "before_files": [{"content": "\"\"\"\nDownloads the MovieLens dataset, ETLs it into Parquet, trains an\nALS model, and uses the ALS model to train a Keras neural network.\n\nSee README.rst for more details.\n\"\"\"\n\nimport click\nimport os\n\n\nimport mlflow\nfrom mlflow.utils import mlflow_tags\nfrom mlflow.entities import RunStatus\nfrom mlflow.utils.logging_utils import eprint\nimport six\n\nfrom mlflow.tracking.fluent import _get_experiment_id\n\n\ndef _already_ran(entry_point_name, parameters, git_commit, experiment_id=None):\n    \"\"\"Best-effort detection of if a run with the given entrypoint name,\n    parameters, and experiment id already ran. The run must have completed\n    successfully and have at least the parameters provided.\n    \"\"\"\n    experiment_id = experiment_id if experiment_id is not None else _get_experiment_id()\n    client = mlflow.tracking.MlflowClient()\n    all_run_infos = reversed(client.list_run_infos(experiment_id))\n    for run_info in all_run_infos:\n        full_run = client.get_run(run_info.run_id)\n        tags = full_run.data.tags\n        if tags.get(mlflow_tags.MLFLOW_PROJECT_ENTRY_POINT, None) != entry_point_name:\n            continue\n        match_failed = False\n        for param_key, param_value in six.iteritems(parameters):\n            run_value = full_run.data.params.get(param_key)\n            if run_value != param_value:\n                match_failed = True\n                break\n        if match_failed:\n            continue\n\n        if run_info.status != RunStatus.FINISHED:\n            eprint((\"Run matched, but is not FINISHED, so skipping \"\n                    \"(run_id=%s, status=%s)\") % (run_info.run_id, run_info.status))\n            continue\n\n        previous_version = tags.get(mlflow_tags.MLFLOW_GIT_COMMIT, None)\n        if git_commit != previous_version:\n            eprint((\"Run matched, but has a different source version, so skipping \"\n                    \"(found=%s, expected=%s)\") % (previous_version, git_commit))\n            continue\n        return client.get_run(run_info.run_id)\n    eprint(\"No matching run has been found.\")\n    return None\n\n\n# TODO(aaron): This is not great because it doesn't account for:\n# - changes in code\n# - changes in dependant steps\ndef _get_or_run(entrypoint, parameters, git_commit, use_cache=True):\n    existing_run = _already_ran(entrypoint, parameters, git_commit)\n    if use_cache and existing_run:\n        print(\"Found existing run for entrypoint=%s and parameters=%s\" % (entrypoint, parameters))\n        return existing_run\n    print(\"Launching new run for entrypoint=%s and parameters=%s\" % (entrypoint, parameters))\n    submitted_run = mlflow.run(\".\", entrypoint, parameters=parameters)\n    return mlflow.tracking.MlflowClient().get_run(submitted_run.run_id)\n\n\n@click.command()\n@click.option(\"--als-max-iter\", default=10, type=int)\n@click.option(\"--keras-hidden-units\", default=20, type=int)\n@click.option(\"--max-row-limit\", default=100000, type=int)\ndef workflow(als_max_iter, keras_hidden_units, max_row_limit):\n    # Note: The entrypoint names are defined in MLproject. The artifact directories\n    # are documented by each step's .py file.\n    with mlflow.start_run() as active_run:\n        os.environ['SPARK_CONF_DIR'] = os.path.abspath('.')\n        git_commit = active_run.data.tags.get(mlflow_tags.MLFLOW_GIT_COMMIT)\n        load_raw_data_run = _get_or_run(\"load_raw_data\", {}, git_commit)\n        ratings_csv_uri = os.path.join(load_raw_data_run.info.artifact_uri, \"ratings-csv-dir\")\n        etl_data_run = _get_or_run(\"etl_data\",\n                                   {\"ratings_csv\": ratings_csv_uri,\n                                    \"max_row_limit\": max_row_limit},\n                                   git_commit)\n        ratings_parquet_uri = os.path.join(etl_data_run.info.artifact_uri, \"ratings-parquet-dir\")\n\n        # We specify a spark-defaults.conf to override the default driver memory. ALS requires\n        # significant memory. The driver memory property cannot be set by the application itself.\n        als_run = _get_or_run(\"als\",\n                              {\"ratings_data\": ratings_parquet_uri, \"max_iter\": str(als_max_iter)},\n                              git_commit)\n        als_model_uri = os.path.join(als_run.info.artifact_uri, \"als-model\")\n\n        keras_params = {\n            \"ratings_data\": ratings_parquet_uri,\n            \"als_model_uri\": als_model_uri,\n            \"hidden_units\": keras_hidden_units,\n        }\n        _get_or_run(\"train_keras\", keras_params, git_commit, use_cache=False)\n\n\nif __name__ == '__main__':\n    workflow()\n", "path": "examples/multistep_workflow/main.py"}], "after_files": [{"content": "\"\"\"\nDownloads the MovieLens dataset, ETLs it into Parquet, trains an\nALS model, and uses the ALS model to train a Keras neural network.\n\nSee README.rst for more details.\n\"\"\"\n\nimport click\nimport os\n\n\nimport mlflow\nfrom mlflow.utils import mlflow_tags\nfrom mlflow.entities import RunStatus\nfrom mlflow.utils.logging_utils import eprint\nimport six\n\nfrom mlflow.tracking.fluent import _get_experiment_id\n\n\ndef _already_ran(entry_point_name, parameters, git_commit, experiment_id=None):\n    \"\"\"Best-effort detection of if a run with the given entrypoint name,\n    parameters, and experiment id already ran. The run must have completed\n    successfully and have at least the parameters provided.\n    \"\"\"\n    experiment_id = experiment_id if experiment_id is not None else _get_experiment_id()\n    client = mlflow.tracking.MlflowClient()\n    all_run_infos = reversed(client.list_run_infos(experiment_id))\n    for run_info in all_run_infos:\n        full_run = client.get_run(run_info.run_id)\n        tags = full_run.data.tags\n        if tags.get(mlflow_tags.MLFLOW_PROJECT_ENTRY_POINT, None) != entry_point_name:\n            continue\n        match_failed = False\n        for param_key, param_value in six.iteritems(parameters):\n            run_value = full_run.data.params.get(param_key)\n            if run_value != param_value:\n                match_failed = True\n                break\n        if match_failed:\n            continue\n\n        if run_info.to_proto().status != RunStatus.FINISHED:\n            eprint((\"Run matched, but is not FINISHED, so skipping \"\n                    \"(run_id=%s, status=%s)\") % (run_info.run_id, run_info.status))\n            continue\n\n        previous_version = tags.get(mlflow_tags.MLFLOW_GIT_COMMIT, None)\n        if git_commit != previous_version:\n            eprint((\"Run matched, but has a different source version, so skipping \"\n                    \"(found=%s, expected=%s)\") % (previous_version, git_commit))\n            continue\n        return client.get_run(run_info.run_id)\n    eprint(\"No matching run has been found.\")\n    return None\n\n\n# TODO(aaron): This is not great because it doesn't account for:\n# - changes in code\n# - changes in dependant steps\ndef _get_or_run(entrypoint, parameters, git_commit, use_cache=True):\n    existing_run = _already_ran(entrypoint, parameters, git_commit)\n    if use_cache and existing_run:\n        print(\"Found existing run for entrypoint=%s and parameters=%s\" % (entrypoint, parameters))\n        return existing_run\n    print(\"Launching new run for entrypoint=%s and parameters=%s\" % (entrypoint, parameters))\n    submitted_run = mlflow.run(\".\", entrypoint, parameters=parameters)\n    return mlflow.tracking.MlflowClient().get_run(submitted_run.run_id)\n\n\n@click.command()\n@click.option(\"--als-max-iter\", default=10, type=int)\n@click.option(\"--keras-hidden-units\", default=20, type=int)\n@click.option(\"--max-row-limit\", default=100000, type=int)\ndef workflow(als_max_iter, keras_hidden_units, max_row_limit):\n    # Note: The entrypoint names are defined in MLproject. The artifact directories\n    # are documented by each step's .py file.\n    with mlflow.start_run() as active_run:\n        os.environ['SPARK_CONF_DIR'] = os.path.abspath('.')\n        git_commit = active_run.data.tags.get(mlflow_tags.MLFLOW_GIT_COMMIT)\n        load_raw_data_run = _get_or_run(\"load_raw_data\", {}, git_commit)\n        ratings_csv_uri = os.path.join(load_raw_data_run.info.artifact_uri, \"ratings-csv-dir\")\n        etl_data_run = _get_or_run(\"etl_data\",\n                                   {\"ratings_csv\": ratings_csv_uri,\n                                    \"max_row_limit\": max_row_limit},\n                                   git_commit)\n        ratings_parquet_uri = os.path.join(etl_data_run.info.artifact_uri, \"ratings-parquet-dir\")\n\n        # We specify a spark-defaults.conf to override the default driver memory. ALS requires\n        # significant memory. The driver memory property cannot be set by the application itself.\n        als_run = _get_or_run(\"als\",\n                              {\"ratings_data\": ratings_parquet_uri, \"max_iter\": str(als_max_iter)},\n                              git_commit)\n        als_model_uri = os.path.join(als_run.info.artifact_uri, \"als-model\")\n\n        keras_params = {\n            \"ratings_data\": ratings_parquet_uri,\n            \"als_model_uri\": als_model_uri,\n            \"hidden_units\": keras_hidden_units,\n        }\n        _get_or_run(\"train_keras\", keras_params, git_commit, use_cache=False)\n\n\nif __name__ == '__main__':\n    workflow()\n", "path": "examples/multistep_workflow/main.py"}]}