Airflow uses standard the Python logging framework to write logs, and for the duration of a task, the root logger is configured to write to the task’s log. So to track the Dataflow pipeline's progress in Airflow, the logging level in your Dataflow pipeline needs to be set to INFO, I had set to ERROR originally. Once I updated the logging level, the operator was able to submit the job and obtain the dataflow_job_id in XCOM, marking itself as success shortly after, and the sensor followed up and tracked the job status to completion.

logging.getLogger().setLevel(logging.INFO)

Read more here: Writing to Airflow task logs from your code