apache/beam · error · RuntimeException

Interrupted while fetching Flight schema

Error message

Interrupted while fetching Flight schema

What it means

In ArrowFlightIO.Read.expand, fetching the FlightInfo (schema) from the Flight service was interrupted while blocking on the RPC; the code restores the interrupt flag and throws RuntimeException('Interrupted while fetching Flight schema'). The read cannot proceed without the schema.

Solutions

1. Rerun the pipeline — interruption during startup is usually transient.
2. Check for cancellation requests in the runner logs around the failure.
3. Verify the Flight server responds promptly; a hung getInfo can prompt timeouts/interruption.
4. Avoid interrupting/terminating the driver process while the graph is being built.

// probe Flight server responsiveness before launching
flightClient.getInfo(FlightDescriptor.path("probe"));

try {
  result = pipeline.run().waitUntilFinish();
} catch (RuntimeException e) {
  if (e.getCause() instanceof InterruptedException) {
    // restart the pipeline; startup interruption is transient
  }
}

When it happens

Trigger: The thread executing expand is interrupted while calling flightClient.getInfo(FlightDescriptor) — e.g., pipeline cancellation or shutdown during pipeline construction.

Common situations: User cancels the pipeline during startup, runner kills the submitting thread, overly aggressive timeouts during job setup.

Related errors

• Interrupted while discovering Flight endpoints
• Unsupported Beam type for ArrowFlightIO.write()…
• A schema is required to write non-schema'd data.
• All dicts in batch must have the same keys. extra keys
• An explicit schema is required to write non-schema'd…

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

Appendix: source

Thrown at sdks/java/io/arrow-flight/src/main/java/org/apache/beam/sdk/io/arrowflight/ArrowFlightIO.java:289

    @Override
    public PCollection<Row> expand(PBegin input) {
      checkArgument(host() != null, "withHost() is required");
      checkArgument(command() != null, "withCommand() is required");

      Schema beamSchema;
      try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE);
          FlightClient client =
              createClient(allocator, checkNotNull(host(), "host"), port(), useTls())) {
        FlightInfo info =
            client.getInfo(
                FlightDescriptor.command(
                    checkNotNull(command(), "command").getBytes(StandardCharsets.UTF_8)),
                callOptions());
        beamSchema = ArrowConversion.ArrowSchemaTranslator.toBeamSchema(info.getSchema());
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException("Interrupted while fetching Flight schema", e);
      }

      return input
          .apply(org.apache.beam.sdk.io.Read.from(new FlightBoundedSource(this, beamSchema)))
          .setRowSchema(beamSchema);
    }

    CallOption[] callOptions() {
      return ArrowFlightIO.callOptions(token());
    }

    @Override
    public void populateDisplayData(DisplayData.Builder builder) {
      super.populateDisplayData(builder);
      builder.addIfNotNull(DisplayData.item("host", host()));
      builder.add(DisplayData.item("port", port()));
      builder.add(DisplayData.item("useTls", useTls()));
      builder.addIfNotNull(DisplayData.item("command", command()));

View on GitHub (pinned to 12126d8942)