Skip to content

Remove S3KeysUnchangedSensor from S3 system test #26606

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
eladkal merged 1 commit into from
Sep 27, 2022
Merged

Remove S3KeysUnchangedSensor from S3 system test #26606

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions tests/system/providers/amazon/aws/example_s3.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

from airflow.models.baseoperator import chain
from airflow.models.dag import DAG
from airflow.operators.python import BranchPythonOperator
from airflow.providers.amazon.aws.operators.s3 import (
S3CopyObjectOperator,
S3CreateBucketOperator,
Expand Down Expand Up @@ -202,6 +203,12 @@ def check_fn(files: list) -> bool:
)
# [END howto_operator_s3_file_transform]

# This task skips the `sensor_keys_unchanged` task because the S3KeysUnchangedSensor

@eladkal eladkal Sep 22, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We have similar sensor in google provider for GCS
Was it handled diffrently there?

@syedahsn syedahsn Sep 26, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The DebugExecutor runs all sensors in reschedule mode, which is causing problems for us with this sensor, and that executor is used for system tests. I'm not sure how they are working around that, but maybe @bhirsz can weigh in?

@bhirsz bhirsz Sep 27, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We overwrite the poke mode for the sensors (but not for all, I need to investigate the results..):
Editing the comment - it appears the solution is not correct and it's failing with:

  File "/workspace/system_tests/airflow/airflow/sensors/base.py", line 276, in mode_setter
    raise ValueError("cannot set mode to 'poke'.")
ValueError: cannot set mode to 'poke'.

So the other option with the modified design is only option.

Other option is to design the tests in a way, that if sensor is in reschedule mode it will not fail on first try (ie. if you're waiting for some file to appear, to upload the file first then run the sensor).

@o-nikolas o-nikolas Sep 27, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Other option is to design the tests in a way, that if sensor is in reschedule mode it will not fail on first try (ie. if you're waiting for some file to appear, to upload the file first then run the sensor).

Interesting! This would be quite a fundamental change since it's fairly deep in core Airflow code where the sensor mode is hard coded to reschedule, so to hook into that would be tricky.

Can we at least agree to create a new issue to track this possible work so that we can unblock this PR? We're happy to skip this one sensor as this PR suggest, to unblock getting coverage from the rest of this test module.

@eladkal eladkal Sep 27, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

#26725

# runs in poke mode only, which is not supported by the DebugExecutor, causing system tests to fail.
branching = BranchPythonOperator(
task_id='branch_to_delete_objects', python_callable=lambda: 'delete_objects'
)

# [START howto_sensor_s3_keys_unchanged]
sensor_keys_unchanged = S3KeysUnchangedSensor(
task_id="sensor_keys_unchanged",
Expand All @@ -218,6 +225,7 @@ def check_fn(files: list) -> bool:
keys=key_2,
)
# [END howto_operator_s3_delete_objects]
delete_objects.trigger_rule = TriggerRule.ALL_DONE

# [START howto_operator_s3_delete_bucket]
delete_bucket = S3DeleteBucketOperator(
Expand Down Expand Up @@ -251,6 +259,7 @@ def check_fn(files: list) -> bool:
[sensor_one_key, sensor_two_keys, sensor_key_with_function],
copy_object,
file_transform,
branching,
sensor_keys_unchanged,
# TEST TEARDOWN
delete_objects,
Expand Down