Operation ongoing in bundle

Error message

Operation ongoing in bundle {} for PTransform{{id={}, name={}, state={}}} for at least {} without outputting or completing:
  at {}

What it means

ExecutionStateSampler warning emitted when a bundle's lull can be attributed to a specific PTransform: it names the transform id, unique name, and execution state, and appends the stuck thread's stack trace. Like the generic lull warning it is diagnostic, helping pinpoint which transform in the pipeline stopped making progress.

Solutions

Read the transform name/id and stack trace to locate the stuck user code in that PTransform.
Add timeouts or async handling to blocking calls inside that transform.
Increase the lull threshold if that transform legitimately runs long without outputting.
Optimize the transform (batching, caching) so it outputs elements more frequently.

Example fix

// before
records.forEach(r -> slowRpc.call(r)); // no progress indication, huge batch
// after
for (Record r : records) {
  slowRpc.callWithTimeout(r, Duration.ofSeconds(30));
}
c.setFlushAfterEachElement(true); // outputs progress so sampler sees liveness

Defensive patterns

Strategy: validation

Validate before calling

// Ensure the transform attributed in the warning has bounded, timeout-protected work
assert rpcTimeoutMs > 0 && batchSize > 0;

Prevention

Add per-call timeouts in the transform named in the warning
Output elements periodically to signal liveness to the sampler
Profile the named PTransform for long tail latencies
Increase lull threshold only when the work is legitimately long

When it happens

Trigger: takeSample() observes a thread stuck beyond the lull threshold while currentExecutionState is non-null, meaning the sampler knows which registered PTransform (via the state registry) the thread is executing inside.

Common situations: A specific DoFn/transform doing blocking calls or heavy computation without progress; identifying the offending transform in Dataflow/Beam portable pipelines; misconfigured lull thresholds flagging slow-but-healthy transforms.

Related errors

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

Appendix: source

Thrown at sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ExecutionStateSampler.java:466

                  > MAX_LULL_TIME_MS + lastLullReport // At least once every MAX_LULL_TIME_MS.
          ) {
            lastLullReport = lullTimeMs;
            Thread thread = trackedThread.get();
            if (thread == null) {
              LOG.warn(
                  "Operation ongoing in bundle {} for at least {} without outputting "
                      + "or completing (stack trace unable to be generated).",
                  processBundleId.get(),
                  DURATION_FORMATTER.print(Duration.millis(lullTimeMs).toPeriod()));
            } else if (currentExecutionState == null) {
              LOG.warn(
                  "Operation ongoing in bundle {} for at least {} without outputting "
                      + "or completing:\n  at {}",
                  processBundleId.get(),
                  DURATION_FORMATTER.print(Duration.millis(lullTimeMs).toPeriod()),
                  Joiner.on("\n  at ").join(thread.getStackTrace()));
            } else {
              LOG.warn(
                  "Operation ongoing in bundle {} for PTransform{{id={}, name={}, state={}}} "
                      + "for at least {} without outputting or completing:\n  at {}",
                  processBundleId.get(),
                  currentExecutionState.ptransformId,
                  currentExecutionState.ptransformUniqueName,
                  currentExecutionState.stateName,
                  DURATION_FORMATTER.print(Duration.millis(lullTimeMs).toPeriod()),
                  Joiner.on("\n  at ").join(thread.getStackTrace()));
            }
          }
        }
      }
      return Optional.empty();
    }

    /** Returns status information related to this tracker or null if not tracking a bundle. */
    public @Nullable ExecutionStateTrackerStatus getStatus() {
      Thread thread = trackedThread.get();

View on GitHub (pinned to 12126d8942)