Skip to content
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

Fixed dask executor and tests #22027

Merged
merged 10 commits into from
Mar 8, 2022
Merged

Fixed dask executor and tests #22027

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
6d86480
Fixed dask executor and tests, distributed package does not ship with…
subkanthi Mar 6, 2022
f9ba4aa
Merge branch 'main' into fix_dask_tests
subkanthi Mar 6, 2022
bf13e7b
Merge branch 'main' into fix_dask_tests
subkanthi Mar 7, 2022
b33fbd3
Removed certs and commented out test_tls test
subkanthi Mar 7, 2022
6d03436
Merge branch 'main' into fix_dask_tests
subkanthi Mar 7, 2022
3785d20
Commented out config_vars for dask executor tests
subkanthi Mar 7, 2022
810fbf7
Merge branch 'main' into fix_dask_tests
subkanthi Mar 7, 2022
8093f79
Merge branch 'master' of github.com:subkanthi/airflow into fix_dask_t…
subkanthi Mar 8, 2022
4013070
Uncommented test_tls in test_dask_executor
subkanthi Mar 8, 2022
e1865e3
Merge branch 'fix_dask_tests' of github.com:subkanthi/airflow into fi…
subkanthi Mar 8, 2022
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
2 changes: 1 addition & 1 deletion airflow/executors/dask_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ def airflow_run():
raise AirflowException(f"Attempted to submit task to an unavailable queue: '{queue}'")
resources = {queue: 1}

future = self.client.submit(airflow_run, pure=False, resources=resources)
future = self.client.submit(subprocess.check_call, command, pure=False, resources=resources)
Copy link
Contributor Author

Choose a reason for hiding this comment

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

Related to #11451

self.futures[future] = key # type: ignore

def _process_future(self, future: Future) -> None:
Expand Down
8 changes: 4 additions & 4 deletions tests/executors/test_dask_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@

# For now we are temporarily removing Dask support until we get Dask Team help us in making the
# tests pass again
skip_dask_tests = True
skip_dask_tests = False


@pytest.mark.skipif(skip_dask_tests, reason="The tests are skipped because it needs testing from Dask team")
Expand Down Expand Up @@ -121,9 +121,9 @@ def setUp(self):

@conf_vars(
{
('dask', 'tls_ca'): get_cert('tls-ca-cert.pem'),
('dask', 'tls_cert'): get_cert('tls-key-cert.pem'),
('dask', 'tls_key'): get_cert('tls-key.pem'),
('dask', 'tls_ca'): 'certs/tls-ca-cert.pem',
('dask', 'tls_cert'): 'certs/tls-key-cert.pem',
('dask', 'tls_key'): 'certs/tls-key.pem',
}
)
def test_tls(self):
Expand Down