BigQuery: add 'retry' argument to '_AsyncJob.result'.

[tseaver]

Pass it through to the _begin call. Note that we need to modify
the api_core...PollingFuture class before we can safely pass the
retry through to its result.

[tswast]

LGTM, though I'll want to hear what Chris or Thea thinks since this is a divergence from the standard "futures" interface. Maybe we should store a private Retry object on the Job class instead? We already override done() for some of the job types.

[tseaver]

@tswast The PollingFuture base class already stores a retry object. If we assigned to it, we would need to ensure that its predicate included the test for _OperationNotComplete (see the DEFAULT_RETRY it uses).

[tseaver]

I had actually planned to update the PollingFuture.result method to take the optional retry argument as well, but wanted to think about how to deal with the _OperationNotComplete bit first.

[tseaver]

@tswast Please see #6305 for the changes to retry the polling requests. I haven't piped through the retry argument from _AsyncJob.result(), because of issues related to _OperationNotComplete.

[tseaver]

If we close this PR without merging (i.e., we decide not to add a retry argument to _AsyncJob.result), then we should change _AsyncJob.result to pass through self._retry to the _begin call. That change won't really be useful until #6305 lands, however.

[crwilcox]

@tseaver is this needed at all if we do #6305?

[tseaver]

@crwilcox

is this needed at all if we do #6305?

Depends The _AsyncJob.result method itself can invoke _begin, which takes an optional retry argument. It makes sense to me that we plumb that through so that callers of result can override the default.

If we decide not to expose it, then I think we still need to pass the self._retry attribute through to begin, because the application might've configured it to override e.g., the timeout from the defaults, via something like:

from google.cloud.bigquery import job

...
copy_job = job.CopyJob(
    job_ref, sources, destination, client=client, job_config=job_config)
copy_job._retry = copy_job._retry.with_deadline(600)
result = copy_job.result()

[tseaver]

Rebased to work around borked CI runs for irrelevant APIs.