Cloud Composer DAG File move for failed Dataflow job does not happen

The Composer DAG (simple_load_dag.py) that has a Dataflow job that ingests data from GCS bucket and moves to different subfolder in case of failure or success accordingly, is not working as expected.

My observation is that the move to failed folder does not happen when the Dataflow job fails. When the dataflow job runs successfully, the file gets moved to correct folder, as expected.