Override BQ load job location when necessary

[ahmedabu98]

Found a case where if we retry a successful load job (e.g. due to bundle failure), we get a 409 ALREADY_EXISTS error along with a job reference that does not contain a location.

In finish bundle, we perform wait operations on job references, where we get and poll until the job is finished. If we attempt to get a job without a location, the API will default it to US multi region. However, this is problematic when we're writing to a different region because BQ won't be able to find the job and we encounter a 404 NOT_FOUND error.

In such a case where the job reference has null location, we should override it with the table's location

[github-actions]

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @shunping for label python.
R: @chamikaramj for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[shunping]

Looks like there are some presubmit errors caused by the code change.

apitools.base.protorpclite.messages.ValidationError: Expected type <class 'str'> for field location, found <Mock name='mock.tables.Get().location' id='133722614311760'> (type <class 'mock.mock.Mock'>) 

I think we need to mock location when creating the mocked object of bq_client.

[shunping]

waiting on author

[ahmedabu98]

@shunping can you PTAL?

[shunping]

LGTM. Thanks!

[Abacn]

There is a racing condition here such that the retry job could happen when the conflict job is running, and then get_table_location will fail on 404, thus failing the Dataflow job.