You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "stegololz (via GitHub)" <gi...@apache.org> on 2026/09/17 14:30:43 UTC
[PR] Keep pooled SFTP connection on application-level SFTP errors [airflow]
stegololz opened a new pull request, #73288:
URL: https://github.com/apache/airflow/pull/73288
`SFTPClientPool.get_sftp_client` yields the pooled client inside a bare
`except Exception` and marks the connection faulty for any exception the caller's
`async with` body raises, including SFTP status replies such as `SFTPNoSuchFile`
or `SFTPPermissionDenied`.
A status reply means the server answered over a live channel, so the connection is
still healthy. Dropping it costs a reconnect and re-auth each time, and since the
caller usually keeps iterating, one transient miss becomes self-sustaining.
Observed on a production task iterating ~36,000 files over one pool:
| phase | files | SSH connections |
|---|---|---|
| before first miss | 28,380 | 23 |
| after first miss | 7,598 | 9,244, with 7,598 re-auths |
The files were not missing; they read back fine afterwards. Each failure also
logged a full traceback, producing 135-190 MB of task logs per attempt.
Only `SFTPNoConnection`, `SFTPConnectionLost` and `SFTPBadMessage` now count as
transport faults. Every other `SFTPError` returns the connection to the pool
intact. Non-`SFTPError` exceptions and `CancelledError` keep the previous
behaviour and still drop it. The exception always propagates unchanged.
The rule sits in a module-level `_is_connection_faulty` rather than inline in the
`except` block because it is a pure function of the exception type: one
parametrised test covers the whole `SFTPError` taxonomy with no pool, event loop
or mocked connection to stand up. It also keeps `get_sftp_client` reading as
acquire, classify, log, release, and puts the rule beside the
`_TRANSPORT_SFTP_ERRORS` tuple it consults.
The pool was introduced in #64465.
Tested with `pytest providers/sftp/tests/unit/sftp/pools/test_sftp.py`: 32 passed,
covering the classifier table, a miss returning the connection to the pool, five
consecutive misses reusing the same client without reconnecting, and the transport
and non-`SFTPError` cases still dropping it.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Opus 5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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] Keep pooled SFTP connection on application-level SFTP errors [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 merged PR #73288:
URL: https://github.com/apache/airflow/pull/73288
--
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] Keep pooled SFTP connection on application-level SFTP errors [airflow]
Posted by "stegololz (via GitHub)" <gi...@apache.org>.
stegololz commented on code in PR #73288:
URL: https://github.com/apache/airflow/pull/73288#discussion_r4060234308
##########
providers/sftp/src/airflow/providers/sftp/pools/sftp.py:
##########
@@ -21,14 +21,31 @@
from contextlib import asynccontextmanager, suppress
from dataclasses import dataclass, field
from threading import Lock
-from typing import TYPE_CHECKING
from weakref import WeakKeyDictionary
+import asyncssh
+
from airflow.providers.sftp.hooks.sftp import SFTPHookAsync
from airflow.utils.log.logging_mixin import LoggingMixin
-if TYPE_CHECKING:
- import asyncssh
+#: SFTP status codes that report a broken or desynchronised transport.
+_TRANSPORT_SFTP_ERRORS: tuple[type[asyncssh.SFTPError], ...] = (
+ asyncssh.SFTPNoConnection,
+ asyncssh.SFTPConnectionLost,
+ asyncssh.SFTPBadMessage,
Review Comment:
done
--
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] Keep pooled SFTP connection on application-level SFTP errors [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #73288:
URL: https://github.com/apache/airflow/pull/73288#discussion_r4052851732
##########
providers/sftp/tests/unit/sftp/pools/test_sftp.py:
##########
@@ -88,6 +89,71 @@ async def test_get_sftp_client_marks_connection_faulty_on_exception(self, sftp_h
assert any(call.kwargs.get("faulty") is True for call in release_spy.call_args_list)
+ @pytest.mark.parametrize(
+ ("exc", "expected"),
+ [
+ pytest.param(asyncssh.SFTPNoSuchFile("missing"), False, id="no-such-file"),
+ pytest.param(asyncssh.SFTPPermissionDenied("denied"), False, id="permission-denied"),
+ pytest.param(asyncssh.SFTPFailure("failure"), False, id="failure"),
+ pytest.param(asyncssh.SFTPOpUnsupported("unsupported"), False, id="op-unsupported"),
+ pytest.param(asyncssh.SFTPConnectionLost("lost"), True, id="connection-lost"),
+ pytest.param(asyncssh.SFTPNoConnection("no connection"), True, id="no-connection"),
+ pytest.param(asyncssh.SFTPBadMessage("bad message"), True, id="bad-message"),
+ pytest.param(ValueError("boom"), True, id="not-an-sftp-error"),
Review Comment:
Same here, this is just a nit like previous comment, I would opt to import those exceptions from asyncssh.
##########
providers/sftp/src/airflow/providers/sftp/pools/sftp.py:
##########
@@ -21,14 +21,31 @@
from contextlib import asynccontextmanager, suppress
from dataclasses import dataclass, field
from threading import Lock
-from typing import TYPE_CHECKING
from weakref import WeakKeyDictionary
+import asyncssh
+
from airflow.providers.sftp.hooks.sftp import SFTPHookAsync
from airflow.utils.log.logging_mixin import LoggingMixin
-if TYPE_CHECKING:
- import asyncssh
+#: SFTP status codes that report a broken or desynchronised transport.
+_TRANSPORT_SFTP_ERRORS: tuple[type[asyncssh.SFTPError], ...] = (
+ asyncssh.SFTPNoConnection,
+ asyncssh.SFTPConnectionLost,
+ asyncssh.SFTPBadMessage,
Review Comment:
I would opt to import those exceptions from asyncssh:
`from asyncssh import SFTPBadMessage, SFTPConnectionLost, SFTPNoConnection`
--
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] Keep pooled SFTP connection on application-level SFTP errors [airflow]
Posted by "stegololz (via GitHub)" <gi...@apache.org>.
stegololz commented on PR #73288:
URL: https://github.com/apache/airflow/pull/73288#issuecomment-5757276305
Thanks both, all applied.
Imports moved to `from asyncssh import ...` in both files, `SFTPError` included,
`import asyncssh` back under `TYPE_CHECKING`.
Docstring now reads "a status-level failure that leaves the channel usable".
Added `assert release_spy.call_args_list`.
Dropped the identity assertion: `sftp_hook_mocked` hands back the same
`sftp_client_mock` every time, so it held either way.
Log volume was `IterableOperator` logging `error_detail` per failed item plus
`_create_connection` at INFO per connection. Description fixed.
32 passed, ruff clean.
---
Drafted-by: Claude Code (Opus 5); reviewed by @stegololz before posting
--
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] Keep pooled SFTP connection on application-level SFTP errors [airflow]
Posted by "stegololz (via GitHub)" <gi...@apache.org>.
stegololz commented on code in PR #73288:
URL: https://github.com/apache/airflow/pull/73288#discussion_r4060249996
##########
providers/sftp/tests/unit/sftp/pools/test_sftp.py:
##########
@@ -88,6 +89,71 @@ async def test_get_sftp_client_marks_connection_faulty_on_exception(self, sftp_h
assert any(call.kwargs.get("faulty") is True for call in release_spy.call_args_list)
+ @pytest.mark.parametrize(
+ ("exc", "expected"),
+ [
+ pytest.param(asyncssh.SFTPNoSuchFile("missing"), False, id="no-such-file"),
+ pytest.param(asyncssh.SFTPPermissionDenied("denied"), False, id="permission-denied"),
+ pytest.param(asyncssh.SFTPFailure("failure"), False, id="failure"),
+ pytest.param(asyncssh.SFTPOpUnsupported("unsupported"), False, id="op-unsupported"),
+ pytest.param(asyncssh.SFTPConnectionLost("lost"), True, id="connection-lost"),
+ pytest.param(asyncssh.SFTPNoConnection("no connection"), True, id="no-connection"),
+ pytest.param(asyncssh.SFTPBadMessage("bad message"), True, id="bad-message"),
+ pytest.param(ValueError("boom"), True, id="not-an-sftp-error"),
Review Comment:
done
--
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