apache/beam · error · IOException

Interrupted while discovering Flight endpoints

Error message

Interrupted while discovering Flight endpoints

What it means

FlightBoundedSource's reader (start) was interrupted while discovering Flight endpoints via listFlights/getInfo; the interrupt flag is restored and an IOException('Interrupted while discovering Flight endpoints') is thrown. Endpoint discovery failed because the blocking call did not complete.

Solutions

1. Retry the pipeline run; interruption during startup is usually transient.
2. Check runner logs for shutdown/cancel signals preceding the interruption.
3. Ensure the Flight server's endpoint discovery responds quickly to avoid long blocking windows.
4. Keep worker lifetimes stable (avoid aggressive preemption during startup).

// verify endpoint discovery works before starting workers
flightClient.listFlights(FlightInfo).forEach(...);

try {
  reader.start();
} catch (IOException e) {
  if (e.getMessage().contains("Interrupted while discovering Flight endpoints")) {
    // surface as retryable / restart source
  }
}

When it happens

Trigger: Thread interruption during the endpoint discovery RPC inside start() — typically runner shutdown or cancellation while the reader initializes.

Common situations: Pipeline cancelled shortly after start, worker preemption, autoscaler killing workers mid-startup.

Related errors

• Interrupted while fetching Flight schema
• A schema is required to write non-schema'd data.
• A sink must inherit iobase.Sink, iobase.NativeSink, or be a…
• An explicit schema is required to write non-schema'd…
• AUTO is applicable only to reading files

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

Appendix: source

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

      String defaultHost = checkNotNull(spec.host(), "host");
      SerializableEndpoint endpoint = source.endpoint;
      if (endpoint == null) {
        try (FlightClient discoveryClient =
            createClient(allocator, defaultHost, spec.port(), spec.useTls())) {
          FlightInfo info =
              discoveryClient.getInfo(
                  FlightDescriptor.command(
                      checkNotNull(spec.command(), "command").getBytes(StandardCharsets.UTF_8)),
                  spec.callOptions());
          List<FlightEndpoint> endpoints = info.getEndpoints();
          if (endpoints.isEmpty()) {
            return false;
          }
          endpoint =
              SerializableEndpoint.fromFlightEndpoint(endpoints.get(0), defaultHost, spec.port());
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          throw new IOException("Interrupted while discovering Flight endpoints", e);
        }
        currentSource = new FlightBoundedSource(spec, source.beamSchema, endpoint);
      }

      client =
          createClient(
              allocator,
              endpoint.getHost(defaultHost),
              endpoint.getPort(spec.port()),
              spec.useTls());
      stream = client.getStream(endpoint.getTicket(), spec.callOptions());
      currentBatchIterator = Collections.emptyIterator();
      return advance();
    }

    @Override
    public boolean advance() throws IOException {
      while (true) {

View on GitHub (pinned to 12126d8942)