
**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
"""Helper class with Dataflow job types."""
    JOB_TYPE_UNKNOWN = "JOB_TYPE_UNKNOWN"
    JOB_TYPE_BATCH = "JOB_TYPE_BATCH"
    JOB_TYPE_STREAMING = "JOB_TYPE_STREAMING"
class _DataflowJobsController(LoggingMixin):
    """
    Interface for communication with Google API.
    It's not use Apache Beam, but only Google Dataflow API.
    :param dataflow: Discovery resource
    :param project_number: The Google Cloud Project ID.
    :param location: Job location.
    :param poll_sleep: The status refresh rate for pending operations.
    :param name: The Job ID prefix used when the multiple_jobs option is passed is set to True.
    :param job_id: ID of a single job.
    :param num_retries: Maximum number of retries in case of connection problems.
    :param multiple_jobs: If set to true this task will be searched by name prefix (``name`` parameter),
        not by specific job ID, then actions will be performed on all matching jobs.
    :param drain_pipeline: Optional, set to True if want to stop streaming job by draining it
        instead of canceling.
    :param cancel_timeout: wait time in seconds for successful job canceling
    :param wait_until_finished: If True, wait for the end of pipeline execution before exiting. If False,
        it only submits job and check once is job not in terminal state.
        The default behavior depends on the type of pipeline:
        * for the streaming pipeline, wait for jobs to start,
        * for the batch pipeline, wait for the jobs to complete.
    """
    def __init__(