apache/beam · warning · UnsupportedOperationException

UnsupportedOperationException

Error message

UnsupportedOperationException

What it means

BeamFnDataInboundObserver.flush() is unimplemented and always throws UnsupportedOperationException. This observer buffers Elements in a queue and forwards them in a background thread, so explicit flushing is not supported on this receiver.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserver.java:115

    for (TimerEndpoint<?> endpoint : timerEndpoints) {
      transformIdToTimerFamilyIdToTimerEndpoint
          .computeIfAbsent(endpoint.getTransformId(), unused -> new HashMap<>())
          .put(endpoint.getTimerFamilyId(), new EndpointStatus<>(endpoint));
    }
    this.queue = new CancellableQueue<>(100);
    this.totalNumEndpoints = dataEndpoints.size() + timerEndpoints.size();
    this.numEndpointsThatAreIncomplete = totalNumEndpoints;
    this.consumingReceivedData = new AtomicBoolean(false);
  }

  @Override
  public void accept(BeamFnApi.Elements elements) throws Exception {
    queue.put(elements);
  }

  @Override
  public void flush() throws Exception {
    throw new UnsupportedOperationException();
  }

  @Override
  public void close() throws Exception {
    queue.cancel(CloseException.INSTANCE);
  }

  public boolean isConsumingReceivedData() {
    return consumingReceivedData.get();
  }

  // Copies the elements of list to an array and removes references to elements that
  // have been iterated past.
  private static class DiscardingIterator<T> implements Iterator<T> {
    private int index = 0;
    private final @Nullable Object[] array;

    DiscardingIterator(List<T> list) {

View on GitHub (pinned to 12126d8942)

Solutions

1. Do not call flush() on BeamFnDataInboundObserver; rely on its internal queue/thread to deliver elements.
2. Wrap the call in a check for the concrete observer type or catch UnsupportedOperationException for inbound receivers.
3. If flush semantics are required, use the outbound MultiplexingRecordView/data receiver that supports flushing.

Example fix

// before
receiver.flush();
// after
if (receiver instanceof BeamFnDataInboundObserver) {
  // flush is not supported for inbound observers
} else {
  receiver.flush();
}

boolean supportsFlush = !(receiver instanceof BeamFnDataInboundObserver);

try { receiver.flush(); } catch (UnsupportedOperationException e) { /* inbound observer: flush unsupported */ }

When it happens

Trigger: Calling flush() directly on a BeamFnDataInboundObserver obtained via the BeamFnData*Client factories, e.g. through generic code that flushes all receivers after writing data.

Common situations: Generic data-distribution code that calls receiver.flush() unconditionally; custom instrumentation or tests invoking flush on inbound observers.

Related errors

• Runner does not support draining.
• Instruction id was poisoned
• Unregistering consumer which was not registered.
• PoisonedException
• NoSuchElementException

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