apache/beam · error · RuntimeError

Timed out waiting for cache file for PCollection

Error message

Timed out waiting for cache file for PCollection `{}` to be available with path {}.

What it means

The streaming cache reader polls for the cache file backing the PCollection every second; if the file never appears within timeout_secs it gives up and raises RuntimeError naming the PCollection variable and expected path. This guards reads against a writer/producer that died or never recorded.

Solutions

Ensure the background caching/recording job is running and completed enough to write the cache file for that PCollection.
Verify the cache_dir and cache labels match those used at write time (CacheKey labels/PCollection var).
Increase timeout_secs passed to read() if the producer is merely slow.
Clear and re-record the cache if the file was deleted or the recording is stale.

Example fix

// before
reader.read(labels=labels, timeout_secs=30)
// after
reader.read(labels=labels, timeout_secs=600)  # or re-run the recording job first

Defensive patterns

Strategy: retry

Validate before calling

import os
key = CacheKey.from_str(labels[-1])
if not os.path.exists(os.path.join(cache._cache_dir, *labels)):
    # ensure recording job has produced output before reading
    rerun_background_caching_job()

Try / catch

try:
    reader.read(labels=labels, timeout_secs=60)
except RuntimeError as e:
    if 'Timed out waiting for cache file' in str(e):
        re_record_pipeline()  # or increase timeout / fix labels
    else:
        raise

Prevention

Keep the recording/background caching job alive until all reads finish.
Match cache_dir and labels exactly between writer and reader.
Allow generous timeouts for slow streaming sources; avoid tmpdir cleaners deleting ib- caches.

When it happens

Trigger: Reading from StreamingCache whose writer hasn't created the cache file yet (recording stopped/crashed), wrong cache_dir or PCollection labels, tailing a cache that was cleared, or a pipeline that never ran to produce that PCollection.

Common situations: Replaying an interactive recording after the background caching job was killed; pointing the cache at a stale/empty directory; a slow producer exceeding the default timeout; cache files removed by tmp cleanup.

Understand the failure class

Background: Request timed out: what client-side request timeouts mean across libraries (Request timed out, TIMED_OUT, APITimeoutError) — this error's family across 39 libraries.

Timeouts: ETIMEDOUT, deadlines, and hung requests — what actually expires when a request times out.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/655f86719ee93d1d. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py:169

      is_cache_complete = lambda _: True

    self._cache_dir = cache_dir
    self._coder = coder
    self._labels = labels
    self._path = os.path.join(self._cache_dir, *self._labels)
    self._is_cache_complete = is_cache_complete
    self._pipeline_id = CacheKey.from_str(labels[-1]).pipeline_id

  def _wait_until_file_exists(self, timeout_secs=30):
    """Blocks until the file exists for a maximum of timeout_secs.
    """
    # Wait for up to `timeout_secs` for the file to be available.
    start = time.time()
    while not os.path.exists(self._path):
      time.sleep(1)
      if time.time() - start > timeout_secs:
        pcollection_var = CacheKey.from_str(self._labels[-1]).var
        raise RuntimeError(
            'Timed out waiting for cache file for PCollection `{}` to be '
            'available with path {}.'.format(pcollection_var, self._path))
    return open(self._path, mode='rb')

  def _emit_from_file(self, fh, tail):
    """Emits the TestStreamFile(Header|Record)s from file.

    This returns a generator to be able to read all lines from the given file.
    If `tail` is True, then it will wait until the cache is complete to exit.
    Otherwise, it will read the file only once.
    """
    # Always read at least once to read the whole file.
    while True:
      pos = fh.tell()
      line = fh.readline()

      # Check if we are at EOF or if we have an incomplete line.
      if not line or (line and line[-1] != b'\n'[0]):

View on GitHub (pinned to 12126d8942)