apache/beam · error

NotImplemented

Error message

NotImplemented

What it means

PipelineResult is a base class whose waitUntilFinish (and rawMetrics) is an abstract placeholder that unconditionally throws 'NotImplemented'. Subclasses (e.g. the portable runner's PipelineResult) are expected to override it; calling it on the base implementation means the runner never supplied a concrete result object.

Solutions

1. Run the pipeline with a supported runner (portable/direct) that returns a concrete PipelineResult subclass.
2. If implementing a custom runner, subclass PipelineResult and override waitUntilFinish/rawMetrics.
3. In tests, mock or subclass PipelineResult instead of using the base class directly.

Example fix

// before
new PipelineResult().waitUntilFinish();
// after
class MyResult extends PipelineResult {
  waitUntilFinish(duration?: number): Promise<JobState_Enum> { /* poll job state */ }
}

if (Object.getPrototypeOf(result) === PipelineResult.prototype) throw new Error('Runner returned base PipelineResult; waitUntilFinish unavailable');

function hasWaitUntilFinish(r: PipelineResult): r is PipelineResult & { waitUntilFinish(d?: number): Promise<JobState_Enum> } {
  return r.waitUntilFinish !== PipelineResult.prototype.waitUntilFinish;
}

try {
  await result.waitUntilFinish();
} catch (e) {
  if ((e as Error).message === 'NotImplemented') {
    // use a runner that provides a concrete PipelineResult, or poll raw state yourself
  } else throw e;
}

When it happens

Trigger: Calling pipeline.waitUntilFinish() (optionally with a duration) on a PipelineResult instance created from the base class rather than a runner-provided subclass.

Common situations: Using a custom or test runner that returns the base PipelineResult; constructing PipelineResult directly in tests; a runner integration bug where the concrete result type was never wired in.

Related errors

• NotImplementedError
• A cannot be expanded
• Assigning an index is not yet supported. Consider using…
• Attempted to invoke timer
• Blocking computation failed. State

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

Appendix: source

Thrown at sdks/typescript/src/apache_beam/runners/runner.ts:29

 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import { JobState_Enum } from "../proto/beam_job_api";
import * as runnerApi from "../proto/beam_runner_api";
import { MonitoringInfo } from "../proto/metrics";
import { Pipeline } from "../internal/pipeline";
import { Root, PValue } from "../pvalue";
import { PipelineOptions } from "../options/pipeline_options";
import * as metrics from "../worker/metrics";

export class PipelineResult {
  waitUntilFinish(duration?: number): Promise<JobState_Enum> {
    throw new Error("NotImplemented");
  }

  async rawMetrics(): Promise<MonitoringInfo[]> {
    throw new Error("NotImplemented");
  }

  // TODO: Support filtering, slicing.
  async counters(): Promise<{ [key: string]: number }> {
    return Object.fromEntries(
      metrics.aggregateMetrics(
        await this.rawMetrics(),
        "beam:metric:user:sum_int64:v1",
      ),
    );
  }

  async distributions(): Promise<{ [key: string]: number }> {
    return Object.fromEntries(

View on GitHub (pinned to 12126d8942)