{"record":{"id":"c48b0126fa57214a","repo":"apache/beam","slug":"pipeline-canceled","errorCode":null,"errorMessage":"pipeline canceled","messagePattern":"pipeline canceled","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"info","filePath":"sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go","lineNumber":36,"sourceCode":"import (\n\t\"context\"\n\t\"errors\"\n\t\"fmt\"\n\t\"log/slog\"\n\t\"sync\"\n\t\"sync/atomic\"\n\n\tjobpb \"github.com/apache/beam/sdks/v2/go/pkg/beam/model/jobmanagement_v1\"\n\tpipepb \"github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1\"\n\t\"github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/urns\"\n\t\"google.golang.org/protobuf/encoding/prototext\"\n\t\"google.golang.org/protobuf/proto\"\n\t\"google.golang.org/protobuf/types/known/timestamppb\"\n)\n\nvar (\n\t// ErrCancel represents a pipeline cancellation by the user.\n\tErrCancel = errors.New(\"pipeline canceled\")\n)\n\nfunc (s *Server) nextId() string {\n\tv := atomic.AddUint32(&s.index, 1)\n\treturn fmt.Sprintf(\"job-%03d\", v)\n}\n\ntype unimplementedError struct {\n\tfeature string\n\tvalue   any\n}\n\nfunc (err unimplementedError) Error() string {\n\treturn fmt.Sprintf(\"unsupported feature %q set with value %v\", err.feature, err.value)\n}\n\nfunc (err unimplementedError) LogValue() slog.Value {\n\treturn slog.GroupValue(","sourceCodeStart":18,"sourceCodeEnd":54,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go#L18-L54","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Treat this as expected control flow, not an error — check errors.Is(err, jobservices.ErrCancel) before marking a job failed","Use context.Cause(ctx) to distinguish user cancellation from other context errors","If you need custom cancel semantics, wrap ErrCancel with fmt.Errorf('%w: reason', jobservices.ErrCancel)"],"exampleFix":"// before\nif err := executePipeline(ctx, wks, j); err != nil {\n\tj.Failed(err)\n}\n// after\nif err := executePipeline(ctx, wks, j); err != nil {\n\tif errors.Is(err, jobservices.ErrCancel) {\n\t\tj.Canceled()\n\t\treturn\n\t}\n\tj.Failed(err)\n}","handlingStrategy":"try-catch","validationCode":"// check cancellation cause before treating err as failure\ncause := context.Cause(ctx)\nisUserCancel := cause == jobservices.ErrCancel || errors.Is(err, jobservices.ErrCancel)","typeGuard":"func isCanceledErr(err error) bool {\n\treturn errors.Is(err, jobservices.ErrCancel) || errors.Is(context.Cause(context.Background()), jobservices.ErrCancel)\n}","tryCatchPattern":"if err := executePipeline(ctx, wks, j); err != nil {\n\tif errors.Is(err, jobservices.ErrCancel) {\n\t\tj.Canceled()\n\t\treturn nil\n\t}\n\tj.Failed(err)\n}","preventionTips":["Always match ErrCancel with errors.Is, never string comparison","Use context.Cause to distinguish user cancellation from deadlines","Record cancellation requests in job metadata before executing"],"tags":["go","beam","prism-runner","cancellation"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}