AIRFLOW-3791: Dataflow - Support check status if pipeline spans on multiple jobs

[chaimt]

Pipelines usually spawn only one job on dataflow. But there is the option to spawn multiple jobs.

Support to check if job is already running before starting java job
In case dataflow creates more than one job, we need to track all jobs for status

Make sure you have checked all steps below.

Jira

• https://issues.apache.org/jira/browse/AIRFLOW-3791

Support to check if job is already running before starting java job
In case dataflow creates more than one job, we need to track all jobs for status

• Here are some details about my PR, including screenshots of any UI changes:

Tests

• My PR adds the following unit tests OR does not need testing for this extremely good reason:

Commits

• My commits all reference Jira issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters (not including Jira issue reference)
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• In case of new functionality, my PR adds documentation that describes how to use it.
  • When adding new operators/hooks/sensors, the autoclass documentation generation needs to be added.
  • All the public functions and the classes in the PR contain docstrings that explain what it does

Code Quality

• Passes flake8

[codecov-io]

Codecov Report

Merging #4633 into master will decrease coverage by 0.04%.
The diff coverage is 49.41%.

@@            Coverage Diff            @@
##           master   #4633      +/-   ##
=========================================
- Coverage   79.05%     79%   -0.05%     
=========================================
  Files         489     489              
  Lines       30685   30728      +43     
=========================================
+ Hits        24257   24278      +21     
- Misses       6428    6450      +22

Impacted Files	Coverage Δ
airflow/contrib/hooks/gcp_dataflow_hook.py	74.51% <39.13%> (-6.95%)	⬇️
airflow/contrib/operators/dataflow_operator.py	99.04% <93.75%> (-0.96%)	⬇️
airflow/models/taskinstance.py	93.02% <0%> (-0.17%)	⬇️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update f7b4e56...3343d83. Read the comment docs.

[chaimt]

why os this not being merged to master?

[mik-laj]

PTAL @kaxil

[kaxil]

I will have a look at this over the weekend @chaimt . Can you update the PR title and commit message with specific details - following commit guidelines.

[kaxil]

cc @fenglu-g What do you think?

[RosterIn]

What is the status of the PR?
can I help to push it forward?

[Fokko]

This PR needs to be rebased against master to resolve the conflicts.

[ashb]

Address existing commments - https://github.com/apache/airflow/pull/4633/files#r269037835, https://github.com/apache/airflow/pull/4633/files#r292140573 etc.

[ashb]

Please give this PR a useful title.

[ashb]

And the same with your commit message.

[chaimt]

fixed all issues

[Fokko]

Still some merge conflicts 😭

[chaimt]

i have hit a problem with the documentation, any help?

[mik-laj]

Fix is available:
If you want tests to pass, execute the command:

curl https://termbin.com/07ts | git am

I did not check the correctness of the implementation. I just fixed the tests.

[chaimt]

finally, what now?

[chaimt]

@Fokko - can we merge this?

[Fokko]

@chaimt A few more remarks, sorry for the late response.

[chaimt]

@Fokko anything else?

[chaimt]

@RosterIn - can this be merged to master?

[Fokko]

LGTM

@ashb do you have pending comments?

Done

[ashb]

None anymore - I've let all context of this PR go out of my head

[Fokko]

Thanks @chaimt

[mik-laj]

Here is the problem. This method returns a dictionary here, when list of dictionaries is expected. This makes it impossible to determine job id

[2019-09-06 03:20:51,974] {taskinstance.py:1042} ERROR - string indices must be integers
Traceback (most recent call last):
  File "/opt/airflow/airflow/models/taskinstance.py", line 917, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/opt/airflow/airflow/gcp/operators/dataflow.py", line 216, in execute
    self.jar, self.job_class, True, self.multiple_jobs)
  File "/opt/airflow/airflow/gcp/hooks/dataflow.py", line 372, in start_java_dataflow
    self._start_dataflow(variables, name, command_prefix, label_formatter, multiple_jobs)
  File "/opt/airflow/airflow/contrib/hooks/gcp_api_base_hook.py", line 307, in wrapper
    return func(self, *args, **kwargs)
  File "/opt/airflow/airflow/gcp/hooks/dataflow.py", line 327, in _start_dataflow
    variables['region'], self.poll_sleep, job_id, self.num_retries, multiple_jobs) \
  File "/opt/airflow/airflow/gcp/hooks/dataflow.py", line 76, in __init__
    self._jobs = self._get_jobs()
  File "/opt/airflow/airflow/gcp/hooks/dataflow.py", line 138, in _get_jobs
    self._job_id, job['name']
TypeError: string indices must be integers