apache/beam · error · Error

Not currently processing a bundle.

Error message

Not currently processing a bundle.

What it means

Worker.getBundleId() returns the id of the bundle currently being processed. When no bundle is active (currentBundleId is null/undefined) it throws, because callers like getStateProvider need an active bundle context to be valid.

Source

Thrown at sdks/typescript/src/apache_beam/worker/worker.ts:442

      if (typeof this.getStateChannel === "function") {
        this.stateProvider = new CachingStateProvider(
          new GrpcStateProvider(
            this.getStateChannel(
              this.descriptor.stateApiServiceDescriptor!.url,
            ),
            this.getBundleId(),
          ),
        );
      } else {
        this.stateProvider = this.getStateChannel;
      }
    }
    return this.stateProvider;
  }

  getBundleId() {
    if (this.currentBundleId === null || this.currentBundleId === undefined) {
      throw new Error("Not currently processing a bundle.");
    }
    return this.currentBundleId!;
  }

  // Put this on a worker thread...
  async process(instructionId: string) {
    console.debug("Processing ", this.descriptor.id, "for", instructionId);
    this.metricsContainer.reset();
    this.currentBundleId = instructionId;
    this.loggingStageInfo.instructionId = instructionId;
    loggingLocalStorage.enterWith(this.loggingStageInfo);
    // We must await these in reverse topological order.
    for (const o of this.topologicallyOrderedOperators.slice().reverse()) {
      this.loggingStageInfo.transformId = o.transformId;
      await o.startBundle();
    }
    this.loggingStageInfo.transformId = undefined;
    // Now finish bundles all the bundles.

View on GitHub (pinned to 12126d8942)

Solutions

1. Call getBundleId()/getStateProvider only inside bundle processing (i.e. from within DoFn setup/process invocations during process(instructionId)).
2. Track the currentBundleId yourself or guard the call: if (worker.currentBundleId != null) ... .
3. Pass the instruction id explicitly instead of relying on ambient bundle state.
4. In tests, start a fake bundle via process() or set up the id before asserting.

Example fix

// before
// const provider = worker.getStateProvider(); // outside process()
// after
// async process: within process(instructionId) { const id = worker.getBundleId(); ... }

// check before calling
if (worker.currentBundleId == null) {
  throw new Error('No active bundle; cannot get state provider');
}

const hasActiveBundle = (w: {currentBundleId: string | null | undefined}): w is {currentBundleId: string} =>
  w.currentBundleId != null;

let bundleId: string;
try {
  bundleId = worker.getBundleId();
} catch (e) {
  if (e.message === 'Not currently processing a bundle.') {
    throw new Error('Call getBundleId only during process()');
  }
  throw e;
}

When it happens

Trigger: Calling worker.getBundleId() or obtaining a state provider between bundles, before process(instructionId) starts, or after a bundle completes and the id is reset to null.

Common situations: Custom runner/driver code probing the worker outside a process() call; callbacks invoked after the bundle finished; unit tests instantiating Worker and calling getBundleId directly.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• State stream is closed.
• Unknown context parameter: ${param.parDoParamName}
• Unsupported window mapping fn: ${sideInput.windowMappingFn.u
• Unimplemented access pattern: ${accessPattern}
• Illegal access to pipeline after visitor traversal was compl

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