Repository navigation
Conversation
run_beam_command read the pipeline subprocess's stdout/stderr with select.select(), which cannot watch a file descriptor numbered >= FD_SETSIZE (1024) and raises 'filedescriptor out of range in select()'. When the task process already holds many open descriptors, the subprocess pipe fds land above 1024 and the pipeline run aborts. ulimit does not help because the limit is compile-time, not RLIMIT_NOFILE. Replace select.select() with selectors.DefaultSelector (epoll/poll), which has no FD_SETSIZE ceiling. The synchronous Beam path (used by the Beam operators and the Google Dataflow pipeline-launch path) is affected; the async path is not. Update the existing logging test to patch selectors and add a regression test that runs a real subprocess under a high fd count. Same class of fix as the SSH provider changes in apache#74211 and apache#74299.
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What happens
When you run a Beam pipeline operator — BeamRunPythonPipelineOperator,
BeamRunJavaPipelineOperator, BeamRunGoPipelineOperator, or a Google Dataflow
operator that launches through the shared Beam hook — the task launches the
pipeline as a subprocess and streams its stdout/stderr back to the task log.
That streaming is done in run_beam_command, which waits for output using
Python's stdlib select.select():
select.select() is backed by a fixed-size fd_set bitmap of FD_SETSIZE (1024)
bits. It physically cannot represent a file descriptor whose number is >= 1024,
and raises "ValueError: filedescriptor out of range in select()" when asked to.
This ceiling is compile-time; it is unrelated to RLIMIT_NOFILE, so raising
ulimit -n does not help.
So whether this fires depends only on the fd NUMBER assigned to the subprocess
pipes. File descriptors are handed out lowest-free-first, so when the task
process already holds many descriptors open, the Beam subprocess's stdout/stderr
pipes are assigned numbers >= 1024 — and the pipeline run aborts with the error
above, even though nothing is actually wrong with the pipeline.
This is the exact same root cause and class of fix already accepted for the SSH
provider in #74211 (hook), and it traces back to the same
FD_SETSIZE issue I originally reported in #74205. The synchronous Beam path is
affected; the async path (run_beam_command_async, asyncio-based) is not.
Fix
Replace the select.select() loop with selectors.DefaultSelector (epoll/poll),
which has no FD_SETSIZE ceiling. Behaviour is otherwise unchanged.
Reproduction / validation
process held 1,100+ open fds: the old select() code raises
"filedescriptor out of range in select()"; the fix captures the subprocess
output correctly.
flips pass/fail is the fd number.
that runs a real subprocess under a high fd count (fails on old select(),
passes with selectors). ruff check + format clean.
related: #74205