apache/beam · critical · RuntimeException

Failed to read elements from the bounded reader.

Error message

Failed to read elements from the bounded reader.

What it means

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.

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.

Example fix

// before: credentials expire mid-checkpoint, advance() throws IOException
BoundedSource<String> source = TextIO.read().from("gs://bucket/data").getBoundedSource();
// after: use a source whose client can refresh credentials, and rely on retries
BoundedSource<String> source = TextIO.read()
    .from("gs://bucket/data")
    .withCoder(StringUtf8Coder.of())
    .getBoundedSource(); // runner retries transient IOExceptions during checkpointing

Defensive patterns

Strategy: try-catch

Validate before calling

// Before relying on checkpointing, sanity-check the bounded reader/source is reachable
try (BoundedSource.BoundedReader<T> reader = source.createReader(options)) {
  if (!reader.start()) {
    throw new IllegalStateException("Bounded reader cannot start; checkpointing would fail");
  }
}

Try / catch

try {
  pipeline.run(); // checkpointing happens inside
} catch (RuntimeException e) {
  if (e.getMessage() != null && e.getMessage().contains("Failed to read elements from the bounded reader")) {
    // inspect e.getCause() (IOException), validate storage/credentials, then retry the job
    Throwable cause = e.getCause();
    if (cause instanceof java.io.IOException) { /* recover: refresh creds, reopen source */ }
  }
  throw e;
}

Prevention

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.

When it happens

Trigger: 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).

Common situations: 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.

Understand the failure class

Background: "failed to read file", EACCES, ENOENT and "could not read <path>" errors: when a program can't read a file from disk — this error's family across 49 libraries.

Related errors

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

Appendix: source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/UnboundedReadFromBoundedSource.java:555

          // Splits the residualSource and tracks the new residualElements in current source.
          BoundedSource<T> residualSplit = null;
          Double fractionConsumed = reader.getFractionConsumed();
          if (fractionConsumed != null && 0 <= fractionConsumed && fractionConsumed <= 1) {
            double fractionRest = 1 - fractionConsumed;
            int splitAttempts = 8;
            for (int i = 0; i < 8 && residualSplit == null; ++i) {
              double fractionToSplit = fractionConsumed + fractionRest * i / splitAttempts;
              residualSplit = reader.splitAtFraction(fractionToSplit);
            }
          }
          List<TimestampedValue<T>> newResidualElements = Lists.newArrayList();
          try {
            while (advance()) {
              newResidualElements.add(
                  TimestampedValue.of(reader.getCurrent(), reader.getCurrentTimestamp()));
            }
          } catch (IOException e) {
            throw new RuntimeException("Failed to read elements from the bounded reader.", e);
          }
          return new Checkpoint<>(newResidualElements, residualSplit);
        }
      }
    }
  }
}

View on GitHub (pinned to 12126d8942)