apache/beam · info

pipeline canceled

Error message

pipeline canceled

What it means

ErrCancel is prism runner's sentinel error representing deliberate pipeline cancellation by the user (e.g. via the Cancel RPC). It is not a failure: execute.go checks errors.Is(err, ErrCancel) and context.Cause(ctx) to route the job to the 'Canceled' state instead of 'Failed'. It is declared as a package-level sentinel so callers can match it with errors.Is.

Source

Thrown at sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go:36

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"sync/atomic"

	jobpb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/jobmanagement_v1"
	pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/urns"
	"google.golang.org/protobuf/encoding/prototext"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"
)

var (
	// ErrCancel represents a pipeline cancellation by the user.
	ErrCancel = errors.New("pipeline canceled")
)

func (s *Server) nextId() string {
	v := atomic.AddUint32(&s.index, 1)
	return fmt.Sprintf("job-%03d", v)
}

type unimplementedError struct {
	feature string
	value   any
}

func (err unimplementedError) Error() string {
	return fmt.Sprintf("unsupported feature %q set with value %v", err.feature, err.value)
}

func (err unimplementedError) LogValue() slog.Value {
	return slog.GroupValue(

View on GitHub (pinned to 12126d8942)

Solutions

1. Treat this as expected control flow, not an error — check errors.Is(err, jobservices.ErrCancel) before marking a job failed
2. Use context.Cause(ctx) to distinguish user cancellation from other context errors
3. If you need custom cancel semantics, wrap ErrCancel with fmt.Errorf('%w: reason', jobservices.ErrCancel)

Example fix

// before
if err := executePipeline(ctx, wks, j); err != nil {
	j.Failed(err)
}
// after
if err := executePipeline(ctx, wks, j); err != nil {
	if errors.Is(err, jobservices.ErrCancel) {
		j.Canceled()
		return
	}
	j.Failed(err)
}

// check cancellation cause before treating err as failure
cause := context.Cause(ctx)
isUserCancel := cause == jobservices.ErrCancel || errors.Is(err, jobservices.ErrCancel)

func isCanceledErr(err error) bool {
	return errors.Is(err, jobservices.ErrCancel) || errors.Is(context.Cause(context.Background()), jobservices.ErrCancel)
}

if err := executePipeline(ctx, wks, j); err != nil {
	if errors.Is(err, jobservices.ErrCancel) {
		j.Canceled()
		return nil
	}
	j.Failed(err)
}

When it happens

Trigger: Calling jobservices.Server.Cancel (or the cancellation RPC endpoint) while a pipeline is running through the prism runner; the executePipeline context is then cancelled with cause ErrCancel.

Common situations: Users aborting long-running pipelines via the job API/UI; test harnesses cancelling jobs; confusion when a canceled job is reported as failed because the sentinel is not matched with errors.Is.

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

• makeWindowedValueCoder: couldn't process coder for pcollecti
• lpUnknownCoders: coder %q not present in base map
• lpUnknownCoders: couldn't handle component %d %q of %q %v: %
• forceLpCoders: coder %q not present in base map
• retrieveCoders: coder %q not present in base map

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