{"record":{"id":"655f86719ee93d1d","repo":"apache/beam","slug":"timed-out-waiting-for-cache-file-for-pcollection-to-be","errorCode":null,"errorMessage":"Timed out waiting for cache file for PCollection `{}` to be available with path {}.","messagePattern":"Timed out waiting for cache file for PCollection `(.+?)` to be available with path (.+?)\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py","lineNumber":169,"sourceCode":"      is_cache_complete = lambda _: True\n\n    self._cache_dir = cache_dir\n    self._coder = coder\n    self._labels = labels\n    self._path = os.path.join(self._cache_dir, *self._labels)\n    self._is_cache_complete = is_cache_complete\n    self._pipeline_id = CacheKey.from_str(labels[-1]).pipeline_id\n\n  def _wait_until_file_exists(self, timeout_secs=30):\n    \"\"\"Blocks until the file exists for a maximum of timeout_secs.\n    \"\"\"\n    # Wait for up to `timeout_secs` for the file to be available.\n    start = time.time()\n    while not os.path.exists(self._path):\n      time.sleep(1)\n      if time.time() - start > timeout_secs:\n        pcollection_var = CacheKey.from_str(self._labels[-1]).var\n        raise RuntimeError(\n            'Timed out waiting for cache file for PCollection `{}` to be '\n            'available with path {}.'.format(pcollection_var, self._path))\n    return open(self._path, mode='rb')\n\n  def _emit_from_file(self, fh, tail):\n    \"\"\"Emits the TestStreamFile(Header|Record)s from file.\n\n    This returns a generator to be able to read all lines from the given file.\n    If `tail` is True, then it will wait until the cache is complete to exit.\n    Otherwise, it will read the file only once.\n    \"\"\"\n    # Always read at least once to read the whole file.\n    while True:\n      pos = fh.tell()\n      line = fh.readline()\n\n      # Check if we are at EOF or if we have an incomplete line.\n      if not line or (line and line[-1] != b'\\n'[0]):","sourceCodeStart":151,"sourceCodeEnd":187,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py#L151-L187","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nreader.read(labels=labels, timeout_secs=30)\n// after\nreader.read(labels=labels, timeout_secs=600)  # or re-run the recording job first","handlingStrategy":"retry","validationCode":"import os\nkey = CacheKey.from_str(labels[-1])\nif not os.path.exists(os.path.join(cache._cache_dir, *labels)):\n    # ensure recording job has produced output before reading\n    rerun_background_caching_job()","typeGuard":null,"tryCatchPattern":"try:\n    reader.read(labels=labels, timeout_secs=60)\nexcept RuntimeError as e:\n    if 'Timed out waiting for cache file' in str(e):\n        re_record_pipeline()  # or increase timeout / fix labels\n    else:\n        raise","preventionTips":["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."],"tags":["apache-beam","interactive-beam","cache","timeout","streaming"],"backgroundTag":"request-timeout","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}