title: How to launch a Dataflow job with Apache Airflow and not block other tasks?
body: <h2>Problem</h2>
  
  <p>Airflow tasks of the type <code>DataflowTemplateOperator</code> take a <strong>long time to complete</strong>. This means <strong>other tasks can be blocked by it</strong> (correct?). </p>
  
  <p>When we run more of these tasks, that means <strong>we would need a bigger Cloud Composer cluster</strong> (in our case) to execute tasks that are essentially <strong>blocking while they shouldn't be</strong> (they should be <em>async</em> operations).  </p>
  
  <h2>Options</h2>
  
  <ul>
  <li><strong>Option 1</strong>: just launch the job and airflow job is successful</li>
  <li><strong>Option 2</strong>: write a wrapper as explained <a href="https://stackoverflow.com/questions/47450783/how-to-wait-for-an-asynchronous-event-in-a-task-of-a-dag-in-a-workflow-implement">here</a> and use a reschedule mode as explained <a href="https://airflow.apache.org/docs/stable/_api/airflow/sensors/base_sensor_operator/index.html" rel="nofollow noreferrer">here</a></li>
  </ul>
  
  <p>Option 1 does not seem feasible as the <code>DataflowTemplateOperator</code> only has an option to specify the wait time between completion checks called <code>poll_sleep</code> (<a href="https://airflow.apache.org/docs/stable/_api/airflow/contrib/operators/dataflow_operator/index.html#airflow.contrib.operators.dataflow_operator.DataflowTemplateOperator" rel="nofollow noreferrer">source</a>). </p>
  
  <p>For the <code>DataflowCreateJavaJobOperator</code> there is an option <code>check_if_running</code> to wait for completion of a previous job with the same name (<a href="https://github.com/apache/airflow/blob/8325dd5507c95eca15a030b2aa5f3e6391734eb9/airflow/gcp/operators/dataflow.py#L213" rel="nofollow noreferrer">see this code</a>)</p>
  
  <p>It seems that after launching a job, the <code>wait_for_finish</code> is executed (see <a href="https://github.com/apache/airflow/blob/8325dd5507c95eca15a030b2aa5f3e6391734eb9/airflow/gcp/hooks/dataflow.py#L395" rel="nofollow noreferrer">this line</a>), which boils down to an "incomplete" job (see <a href="https://github.com/apache/airflow/blob/8325dd5507c95eca15a030b2aa5f3e6391734eb9/airflow/gcp/hooks/dataflow.py#L237" rel="nofollow noreferrer">this line</a>).  </p>
  
  <p>For Option 2, I need Option 1.</p>
  
  <h2>Questions</h2>
  
  <ol>
  <li>Am I correct to assume that <strong>Dataflow tasks will block others</strong> in Cloud Composer/Airflow?</li>
  <li>Is there a way to schedule a job <strong>without a "wait to finish"</strong> using the built-in operators? (I might have overlooked something)</li>
  <li>Is there an easy way to <strong>write this myself</strong>? I'm thinking of just executing a bash launch script, followed by a task that looks if the job finished correctly, but in a reschedule mode.</li>
  <li>Is there <strong>another way to avoid blocking other tasks</strong> while running dataflow jobs? Basically this is an async operation and should not take resources.</li>
  </ol>