prompt: 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()
  
  ```
  
golden_diff: 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
  
verification_info: {"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"}]}