You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "pankajastro (via GitHub)" <gi...@apache.org> on 2024/11/22 07:34:26 UTC
[PR] BigQueryInsertJobOperator: log transient error and reload job propert… [airflow]
pankajastro opened a new pull request, #44279:
URL: https://github.com/apache/airflow/pull/44279
…y before marking task as failed or success
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and reload job property before marking task as failed or success [airflow]
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro commented on PR #44279:
URL: https://github.com/apache/airflow/pull/44279#issuecomment-2493675054
> There's another active PR: #40863
Yeah, but it has some open discussions, and it's also a pretty old PR. so I thought maybe a new PR would easy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and reload job property before marking task as failed or success [airflow]
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro commented on code in PR #44279:
URL: https://github.com/apache/airflow/pull/44279#discussion_r1862715519
##########
providers/src/airflow/providers/google/cloud/operators/bigquery.py:
##########
@@ -2592,8 +2592,15 @@ def _submit_job(
nowait=True,
)
- @staticmethod
- def _handle_job_error(job: BigQueryJob | UnknownJob) -> None:
+ def _handle_job_error(self, job: BigQueryJob | UnknownJob) -> None:
+ self.log.info("Job %s is completed. Checking the job status", self.job_id)
+ # I've noticed that sometimes BigQuery jobs transiently report the wrong status, causing the Airflow job to be incorrectly marked as successful.
+ # To avoid this, we refresh the job properties before checking the final state and handling any errors.
+ while job.state != "DONE":
Review Comment:
Hi @eladkal,
Thank you so much for your review and feedback; I really appreciate it. I understand your concerns regarding the handling of third-party library issues in Airflow, and I agree that addressing it in this manner might not be the ideal approach.
The issue we encountered is quite unusual. We experienced it in production, where the job ultimately timed out after 6 hours, but the `job.result()` method returned before the job had actually finished.
I suspect this could be related to network issues or temporary service disruptions that might have caused the j`ob.result()` to return prematurely.
Additionally, I feel that checking `job.error_result` may not always be sufficient to detect all errors. As noted in the BigQuery documentation, error_result may not be present in every error response from BigQuery.
https://cloud.google.com/bigquery/docs/error-messages
As for the opening issue in the upstream library, I'm hesitant to pursue it at this stage. I don't have a consistent way to reproduce it, nor enough evidence to definitively say that the problem lies with the BigQuery client or BigQuery service itself.
To address the error handling, would it make sense to modify the `_handle_job_error` method to log any errors in `job.errors` and fail the job if `job.state != "DONE"`? Below is the proposed update:
```python
def _handle_job_error(self, job: BigQueryJob | UnknownJob) -> None:
self.log.info("Job %s is completed. Checking the job status", self.job_id)
# Log any transient errors encountered during the job execution
for error in job.errors or []:
self.log.error("BigQuery Job Error: %s", error)
if job.error_result:
raise AirflowException(f"BigQuery job {job.job_id} failed: {job.error_result}")
# Check the final state.
if job.state != "DONE":
raise AirflowException(f"Job failed with state: {job.state}")
```
I would really appreciate your thoughts on whether this approach makes sense, or if you have any further suggestions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and reload job property before marking task as failed or success [airflow]
Posted by "rawwar (via GitHub)" <gi...@apache.org>.
rawwar commented on PR #44279:
URL: https://github.com/apache/airflow/pull/44279#issuecomment-2493124169
There's another active PR: https://github.com/apache/airflow/pull/40863
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and reload job property before marking task as failed or success [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #44279:
URL: https://github.com/apache/airflow/pull/44279#discussion_r1853912328
##########
providers/src/airflow/providers/google/cloud/operators/bigquery.py:
##########
@@ -2592,8 +2592,15 @@ def _submit_job(
nowait=True,
)
- @staticmethod
- def _handle_job_error(job: BigQueryJob | UnknownJob) -> None:
+ def _handle_job_error(self, job: BigQueryJob | UnknownJob) -> None:
+ self.log.info("Job %s is completed. Checking the job status", self.job_id)
+ # I've noticed that sometimes BigQuery jobs transiently report the wrong status, causing the Airflow job to be incorrectly marked as successful.
+ # To avoid this, we refresh the job properties before checking the final state and handling any errors.
+ while job.state != "DONE":
Review Comment:
I don't follow why we should do this.
Is there a bug report to BigQuery that we can track?
I am very against such permanent workarounds. Airflow should not be the fixer for upstream library bugs. Its OK to do this a temporary workaround when upstream library is engaged and working on a fix but otherwise not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and check job state before marking task as success [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on code in PR #44279:
URL: https://github.com/apache/airflow/pull/44279#discussion_r1863373235
##########
providers/src/airflow/providers/google/cloud/operators/bigquery.py:
##########
@@ -2592,8 +2592,15 @@ def _submit_job(
nowait=True,
)
- @staticmethod
- def _handle_job_error(job: BigQueryJob | UnknownJob) -> None:
+ def _handle_job_error(self, job: BigQueryJob | UnknownJob) -> None:
+ self.log.info("Job %s is completed. Checking the job status", self.job_id)
+ # I've noticed that sometimes BigQuery jobs transiently report the wrong status, causing the Airflow job to be incorrectly marked as successful.
+ # To avoid this, we refresh the job properties before checking the final state and handling any errors.
+ while job.state != "DONE":
Review Comment:
Hey @pankajastro and thanks for putting more details!
Indeed, according to BigQuery docs - the lack of `error_result` could be considered as an expected behavior:
```
The table does not include all possible HTTP errors or other networking errors. Therefore, don't assume that an error object is present in every error response from BigQuery.
```
Therefore, handling it with an additional `if job.state != "DONE"` could be sufficient for now to ensure that the operator does not report errorneous results.
Applying the `while` loop on the other hand, as currently implmented in @kandharvishnu 's [PR](https://github.com/apache/airflow/blame/8fdf29d39dd315f3d7144b4bba66f348221292e6/providers/src/airflow/providers/google/cloud/operators/bigquery.py#L2688), exceeds the atomicity the operation (we already called `job.result()` that should provides a final answer - why do we need to recall it?) - which makes us in this case, Airflow maintainers, responsible for handling BQ transient errors. I assume that we don't want to get there.
To sum it up, I'm ok with merging this solution as-is, with Elad's consent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and check job state before marking task as success [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on code in PR #44279:
URL: https://github.com/apache/airflow/pull/44279#discussion_r1863373235
##########
providers/src/airflow/providers/google/cloud/operators/bigquery.py:
##########
@@ -2592,8 +2592,15 @@ def _submit_job(
nowait=True,
)
- @staticmethod
- def _handle_job_error(job: BigQueryJob | UnknownJob) -> None:
+ def _handle_job_error(self, job: BigQueryJob | UnknownJob) -> None:
+ self.log.info("Job %s is completed. Checking the job status", self.job_id)
+ # I've noticed that sometimes BigQuery jobs transiently report the wrong status, causing the Airflow job to be incorrectly marked as successful.
+ # To avoid this, we refresh the job properties before checking the final state and handling any errors.
+ while job.state != "DONE":
Review Comment:
Hey @pankajastro and thanks for putting more details!
Indeed, according to BigQuery docs - the lack of `error_result` could be considered as an expected behavior:
```
The table does not include all possible HTTP errors or other networking errors. Therefore, don't assume that an error object is present in every error response from BigQuery.
```
Therefore, handling it with an additional `if job.state != "DONE"` could be sufficient for now to ensure that the operator does not report errorneous results.
Applying the `while` loop on the other hand, as currently implmented in @kandharvishnu 's [PR](https://github.com/apache/airflow/blame/8fdf29d39dd315f3d7144b4bba66f348221292e6/providers/src/airflow/providers/google/cloud/operators/bigquery.py#L2688), exceeds the atomicity the operation (we already called `job.result()` that should provides a final answer - why do we need to recall it?). That makes us in this case, Airflow maintainers, responsible for handling BQ transient errors - I assume that we don't want to get there.
To sum it up, I'm ok with merging this solution as-is, with Elad's consent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and reload job property before marking task as failed or success [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #44279:
URL: https://github.com/apache/airflow/pull/44279#discussion_r1853912328
##########
providers/src/airflow/providers/google/cloud/operators/bigquery.py:
##########
@@ -2592,8 +2592,15 @@ def _submit_job(
nowait=True,
)
- @staticmethod
- def _handle_job_error(job: BigQueryJob | UnknownJob) -> None:
+ def _handle_job_error(self, job: BigQueryJob | UnknownJob) -> None:
+ self.log.info("Job %s is completed. Checking the job status", self.job_id)
+ # I've noticed that sometimes BigQuery jobs transiently report the wrong status, causing the Airflow job to be incorrectly marked as successful.
+ # To avoid this, we refresh the job properties before checking the final state and handling any errors.
+ while job.state != "DONE":
Review Comment:
I don't follow why we should do this.
Is there a bug report to BigQuery that we can track?
I am very against such permanent workarounds. Airflow should not be the fixer for upstream library bugs. Its OK to do this a temporary workaround when upstream library is engaged and working on a fix but other not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and check job state before marking task as success [airflow]
Posted by "dejii (via GitHub)" <gi...@apache.org>.
dejii commented on code in PR #44279:
URL: https://github.com/apache/airflow/pull/44279#discussion_r2700707375
##########
providers/src/airflow/providers/google/cloud/operators/bigquery.py:
##########
@@ -2592,8 +2592,15 @@ def _submit_job(
nowait=True,
)
- @staticmethod
- def _handle_job_error(job: BigQueryJob | UnknownJob) -> None:
+ def _handle_job_error(self, job: BigQueryJob | UnknownJob) -> None:
+ self.log.info("Job %s is completed. Checking the job status", self.job_id)
+ # I've noticed that sometimes BigQuery jobs transiently report the wrong status, causing the Airflow job to be incorrectly marked as successful.
+ # To avoid this, we refresh the job properties before checking the final state and handling any errors.
+ while job.state != "DONE":
Review Comment:
@shahar1 @pankajastro – I'm suspecting that the `job.state != "DONE"` didn't need to be introduced here. I took a look at the client library while fixing a silent bug introduced by this PR (see fix: https://github.com/apache/airflow/pull/60679), and found that `job.result()` can only return successfully when `state=DONE` or it returns an Exception which will cause the task to fail.
References:
1. https://github.com/googleapis/python-bigquery/blob/73228432a3c821db05d898ea4a4788adf15b033d/google/cloud/bigquery/job/base.py#L990-L1016
2. https://github.com/googleapis/python-bigquery/blob/73228432a3c821db05d898ea4a4788adf15b033d/google/cloud/bigquery/job/query.py#L1770-L1787
Also, per [docs](https://docs.cloud.google.com/bigquery/docs/reference/rest/v2/Job#JobStatus):
> Output only. Running state of the job. Valid states include 'PENDING', 'RUNNING', and 'DONE'.
So it's very unlikely that `"Job failed with state: PENDING|RUNNING"` is a valid exception.
Let me know your thoughts here, I can remove it in my [PR](https://github.com/apache/airflow/pull/60679) since it's somewhat related.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and reload job property before marking task as failed or success [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #44279:
URL: https://github.com/apache/airflow/pull/44279#discussion_r1853912328
##########
providers/src/airflow/providers/google/cloud/operators/bigquery.py:
##########
@@ -2592,8 +2592,15 @@ def _submit_job(
nowait=True,
)
- @staticmethod
- def _handle_job_error(job: BigQueryJob | UnknownJob) -> None:
+ def _handle_job_error(self, job: BigQueryJob | UnknownJob) -> None:
+ self.log.info("Job %s is completed. Checking the job status", self.job_id)
+ # I've noticed that sometimes BigQuery jobs transiently report the wrong status, causing the Airflow job to be incorrectly marked as successful.
+ # To avoid this, we refresh the job properties before checking the final state and handling any errors.
+ while job.state != "DONE":
Review Comment:
I don't follow why we should do this.
Is there even a bug report to BigQuery?
I am very against such permanent workarounds. Airflow should not be the fixer for upstream library bugs. Its OK to do this a temporary workaround when upstream library is engaged and working on a fix but other not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and check job state before marking task as success [airflow]
Posted by "dejii (via GitHub)" <gi...@apache.org>.
dejii commented on code in PR #44279:
URL: https://github.com/apache/airflow/pull/44279#discussion_r2700707375
##########
providers/src/airflow/providers/google/cloud/operators/bigquery.py:
##########
@@ -2592,8 +2592,15 @@ def _submit_job(
nowait=True,
)
- @staticmethod
- def _handle_job_error(job: BigQueryJob | UnknownJob) -> None:
+ def _handle_job_error(self, job: BigQueryJob | UnknownJob) -> None:
+ self.log.info("Job %s is completed. Checking the job status", self.job_id)
+ # I've noticed that sometimes BigQuery jobs transiently report the wrong status, causing the Airflow job to be incorrectly marked as successful.
+ # To avoid this, we refresh the job properties before checking the final state and handling any errors.
+ while job.state != "DONE":
Review Comment:
@shahar1 @pankajastro – I'm suspecting that the `job.state != "DONE"` didn't need to be introduced here. I took a look at the client library while fixing a silent bug introduced by this PR (see fix: https://github.com/apache/airflow/pull/60679), and found that `job.result()` can only return successfully when `state=DONE` or it raises an Exception which will cause the task to fail.
References:
1. https://github.com/googleapis/python-bigquery/blob/73228432a3c821db05d898ea4a4788adf15b033d/google/cloud/bigquery/job/base.py#L990-L1016
2. https://github.com/googleapis/python-bigquery/blob/73228432a3c821db05d898ea4a4788adf15b033d/google/cloud/bigquery/job/query.py#L1770-L1787
Also, per [docs](https://docs.cloud.google.com/bigquery/docs/reference/rest/v2/Job#JobStatus):
> Output only. Running state of the job. Valid states include 'PENDING', 'RUNNING', and 'DONE'.
So it's very unlikely that `"Job failed with state: PENDING|RUNNING"` is a valid exception.
Let me know your thoughts here, I can remove it in my [PR](https://github.com/apache/airflow/pull/60679) since it's somewhat related.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and reload job property before marking task as failed or success [airflow]
Posted by "kandharvishnu (via GitHub)" <gi...@apache.org>.
kandharvishnu commented on PR #44279:
URL: https://github.com/apache/airflow/pull/44279#issuecomment-2493499113
Closes this issue - https://github.com/apache/airflow/issues/40839
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and reload job property before marking task as failed or success [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #44279:
URL: https://github.com/apache/airflow/pull/44279#discussion_r1853416369
##########
providers/src/airflow/providers/google/cloud/operators/bigquery.py:
##########
@@ -2592,8 +2592,15 @@ def _submit_job(
nowait=True,
)
- @staticmethod
- def _handle_job_error(job: BigQueryJob | UnknownJob) -> None:
+ def _handle_job_error(self, job: BigQueryJob | UnknownJob) -> None:
+ self.log.debug("Job %s is completed. checking the job status", self.job_id)
Review Comment:
```suggestion
self.log.debug("Job %s is completed. Checking the job status", self.job_id)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] BigQueryInsertJobOperator: log transient error and check job state before marking task as success [airflow]
Posted by "pankajastro (via GitHub)" <gi...@apache.org>.
pankajastro merged PR #44279:
URL: https://github.com/apache/airflow/pull/44279
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org