{"record":{"id":"9bdeb9d6ee632209","repo":"apache/beam","slug":"failed-to-read-elements-from-the-bounded-reader","errorCode":null,"errorMessage":"Failed to read elements from the bounded reader.","messagePattern":"Failed to read elements from the bounded reader\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"critical","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/UnboundedReadFromBoundedSource.java","lineNumber":555,"sourceCode":"          // Splits the residualSource and tracks the new residualElements in current source.\n          BoundedSource<T> residualSplit = null;\n          Double fractionConsumed = reader.getFractionConsumed();\n          if (fractionConsumed != null && 0 <= fractionConsumed && fractionConsumed <= 1) {\n            double fractionRest = 1 - fractionConsumed;\n            int splitAttempts = 8;\n            for (int i = 0; i < 8 && residualSplit == null; ++i) {\n              double fractionToSplit = fractionConsumed + fractionRest * i / splitAttempts;\n              residualSplit = reader.splitAtFraction(fractionToSplit);\n            }\n          }\n          List<TimestampedValue<T>> newResidualElements = Lists.newArrayList();\n          try {\n            while (advance()) {\n              newResidualElements.add(\n                  TimestampedValue.of(reader.getCurrent(), reader.getCurrentTimestamp()));\n            }\n          } catch (IOException e) {\n            throw new RuntimeException(\"Failed to read elements from the bounded reader.\", e);\n          }\n          return new Checkpoint<>(newResidualElements, residualSplit);\n        }\n      }\n    }\n  }\n}\n","sourceCodeStart":537,"sourceCodeEnd":563,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/UnboundedReadFromBoundedSource.java#L537-L563","documentation":"During a checkpoint of an UnboundedSource wrapper around a BoundedSource (UnboundedReadFromBoundedSource), the checkpoint mark callback drains the remaining elements from the bounded reader by repeatedly calling advance(). If advance() (or the underlying bounded reader's reads) throws an IOException, it is rethrown as a RuntimeException with this message, wrapping the original IOException as the cause.","triggerScenarios":"The runner checkpoints the bounded-source-backed unbounded read; CheckpointMark.getCheckpointMark() calls advance() on the underlying BoundedSource.BoundedReader, which throws IOException due to I/O failure on the underlying storage (e.g. closed stream, network failure reading files, snapshot/reader inconsistency after the split was resumed).","commonSituations":"Reading from GCS/HDFS/S3-backed bounded sources whose underlying stream was closed or expired mid-checkpoint; runners that trigger frequent checkpointing (streaming pipelines) hitting transient storage outages; a reader implementation that mismanages state after splitAtWatermark/element consumption so subsequent advance() fails.","solutions":["Inspect the wrapped IOException (the cause) to identify and fix the underlying I/O failure (expired credentials, closed stream, network error).","Retry the pipeline; for transient storage errors, configure runner-level retry and backoff policies on the read.","Ensure credentials/storage client for the bounded source are valid and long-lived enough to survive streaming checkpoint lifetimes.","Check the BoundedReader implementation for state corruption after splits; update the Beam SDK / source connector to a version fixing reader restart bugs."],"exampleFix":"// before: credentials expire mid-checkpoint, advance() throws IOException\nBoundedSource<String> source = TextIO.read().from(\"gs://bucket/data\").getBoundedSource();\n// after: use a source whose client can refresh credentials, and rely on retries\nBoundedSource<String> source = TextIO.read()\n    .from(\"gs://bucket/data\")\n    .withCoder(StringUtf8Coder.of())\n    .getBoundedSource(); // runner retries transient IOExceptions during checkpointing","handlingStrategy":"try-catch","validationCode":"// Before relying on checkpointing, sanity-check the bounded reader/source is reachable\ntry (BoundedSource.BoundedReader<T> reader = source.createReader(options)) {\n  if (!reader.start()) {\n    throw new IllegalStateException(\"Bounded reader cannot start; checkpointing would fail\");\n  }\n}","typeGuard":null,"tryCatchPattern":"try {\n  pipeline.run(); // checkpointing happens inside\n} catch (RuntimeException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"Failed to read elements from the bounded reader\")) {\n    // inspect e.getCause() (IOException), validate storage/credentials, then retry the job\n    Throwable cause = e.getCause();\n    if (cause instanceof java.io.IOException) { /* recover: refresh creds, reopen source */ }\n  }\n  throw e;\n}","preventionTips":["Keep source credentials valid for the full streaming-pipeline lifetime and enable automatic credential refresh.","Prefer bounded-source connectors with resilient, reopenable readers.","Avoid checkpointing more aggressively than the underlying storage can handle; configure reasonable checkpoint intervals.","Pin one Beam SDK version and update the bounded source connector to the latest patch release."],"tags":["io","checkpointing","bounded-source","streaming","java"],"backgroundTag":"file-read-failed","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"}