
**Apache Airflow version**:
1. 1.10.9 Composer Airflow Image

**Environment**:

- **Cloud provider or hardware configuration**: Cloud Composer

**What happened**:
Error logs indicate appears to not recognize the job as Batch.


[2020-12-22 16:28:53,445] {taskinstance.py:1135} ERROR - 'type'
Traceback (most recent call last)
  File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 972, in _run_raw_tas
    result = task_copy.execute(context=context
  File "/usr/local/lib/airflow/airflow/providers/google/cloud/operators/dataflow.py", line 647, in execut
    on_new_job_id_callback=set_current_job_id
  File "/usr/local/lib/airflow/airflow/providers/google/common/hooks/base_google.py", line 383, in inner_wrappe
    return func(self, *args, **kwargs
  File "/usr/local/lib/airflow/airflow/providers/google/cloud/hooks/dataflow.py", line 804, in start_flex_templat
    jobs_controller.wait_for_done(
  File "/usr/local/lib/airflow/airflow/providers/google/cloud/hooks/dataflow.py", line 348, in wait_for_don
    while self._jobs and not all(self._check_dataflow_job_state(job) for job in self._jobs)
  File "/usr/local/lib/airflow/airflow/providers/google/cloud/hooks/dataflow.py", line 348, in <genexpr
    while self._jobs and not all(self._check_dataflow_job_state(job) for job in self._jobs)
  File "/usr/local/lib/airflow/airflow/providers/google/cloud/hooks/dataflow.py", line 321, in _check_dataflow_job_stat
    wait_for_running = job['type'] == DataflowJobType.JOB_TYPE_STREAMIN
KeyError: 'type


I have specified:

```
with models.DAG(
    dag_id="pdc-test",
    start_date=days_ago(1),
    schedule_interval=None,
) as dag_flex_template:
    start_flex_template = DataflowStartFlexTemplateOperator(
        task_id="pdc-test",
        body={
            "launchParameter": {
                "containerSpecGcsPath": GCS_FLEX_TEMPLATE_TEMPLATE_PATH,
                "jobName": DATAFLOW_FLEX_TEMPLATE_JOB_NAME,
                "parameters": {
                    "stage": STAGE,
                    "target": TARGET,
                    "path": PATH,
                    "filename": FILENAME,
                    "column": "geometry"
                },
                "environment": {
                    "network": NETWORK,
                    "subnetwork": SUBNETWORK,
                    "machineType": "n1-standard-1",
                    "numWorkers": "1",
                    "maxWorkers": "1",
                    "tempLocation": "gs://test-pipelines-work/batch",
                    "workerZone": "northamerica-northeast1",
                    "enableStreamingEngine": "false",
                    "serviceAccountEmail": "<number>-compute@developer.gserviceaccount.com",
                    "ipConfiguration": "WORKER_IP_PRIVATE"
                },
            }
        },
        location=LOCATION,
        project_id=GCP_PROJECT_ID
    )```


**What you expected to happen**:

Expecting the dag to run.

<!-- What do you think went wrong? -->

Appears the Operator is not handling the input as a batch type Flex Template. DataflowJobType should be BATCH and not STREAMING.

**How to reproduce it**:
1. Create a Batch Flex Template as of https://cloud.google.com/dataflow/docs/guides/templates/using-flex-templates
2. Point code above to your registered template and invoke.


airflow/providers/google/cloud/hooks/dataflow.py
#
#
#
"""This module contains a Google Dataflow Hook."""
import functools
import json
import re
import shlex
import subprocess
import time
import uuid
import warnings
from copy import deepcopy
from typing import Any, Callable, Dict, Generator, List, Optional, Sequence, Set, TypeVar, Union, cast
from googleapiclient.discovery import build
from airflow.exceptions import AirflowException
from airflow.providers.apache.beam.hooks.beam import BeamHook, BeamRunnerType, beam_options_to_args
from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.timeout import timeout
DEFAULT_DATAFLOW_LOCATION = "us-central1"
JOB_ID_PATTERN = re.compile(
    r"Submitted job: (?P<job_id_java>.*)|Created job with id: \[(?P<job_id_python>.*)\]"
)
T = TypeVar("T", bound=Callable)  
def process_line_and_extract_dataflow_job_id_callback(