Fix BigQueryInsertJobOperator cancel_on_kill

[lidalei]

This PR fixes an issue with BigQueryInsertJobOperator. If the task reaches timeout set by task execution_timeout, on_kill will be called but the self.job_id is None. This is because the function _submit_job is a blocking call but self.job_id is only set after it. This PR is hugely inspired by #22955.

[potiuk]

That looks cool

[potiuk]

Some errors though

[lidalei]

Some errors though

Fixed the failed test case. The Conflict exception will be raised when we call _begin.

Traceback (most recent call last):
  File "/Users/dalei/.pyenv/versions/3.8.10/lib/python3.8/code.py", line 90, in runcode
    exec(code, self.locals)
  File "<input>", line 1, in <module>
  File "/Users/dalei/go/src/github.com/lidalei/airflow/airflow/providers/google/common/hooks/base_google.py", line 463, in inner_wrapper
    return func(self, *args, **kwargs)
  File "/Users/dalei/go/src/github.com/lidalei/airflow/airflow/providers/google/cloud/hooks/bigquery.py", line 1542, in insert_job
    job._begin()
  File "/Users/dalei/go/src/github.com/lidalei/airflow/venv/lib/python3.8/site-packages/google/cloud/bigquery/job/query.py", line 1298, in _begin
    super(QueryJob, self)._begin(client=client, retry=retry, timeout=timeout)
  File "/Users/dalei/go/src/github.com/lidalei/airflow/venv/lib/python3.8/site-packages/google/cloud/bigquery/job/base.py", line 510, in _begin
    api_response = client._call_api(
  File "/Users/dalei/go/src/github.com/lidalei/airflow/venv/lib/python3.8/site-packages/google/cloud/bigquery/client.py", line 759, in _call_api
    return call()
  File "/Users/dalei/go/src/github.com/lidalei/airflow/venv/lib/python3.8/site-packages/google/api_core/retry.py", line 283, in retry_wrapped_func
    return retry_target(
  File "/Users/dalei/go/src/github.com/lidalei/airflow/venv/lib/python3.8/site-packages/google/api_core/retry.py", line 190, in retry_target
    return target()
  File "/Users/dalei/go/src/github.com/lidalei/airflow/venv/lib/python3.8/site-packages/google/cloud/_http/__init__.py", line 494, in api_request
    raise exceptions.from_http_response(response)
google.api_core.exceptions.Conflict: 409 POST https://bigquery.googleapis.com/bigquery/v2/projects/xxx/jobs?prettyPrint=false: Already Exists: Job xxx:EU.abc_test
Location: EU
Job ID: abc_test

[boring-cyborg]

Awesome work, congrats on your first merged pull request!