{"record":{"id":"3afd9fe81edfba35","repo":"apache/beam","slug":"plan-v-failed-v-callbacks","errorCode":null,"errorMessage":"Plan %v failed %v callbacks","messagePattern":"Plan (.+?) failed (.+?) callbacks","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/plan.go","lineNumber":198,"sourceCode":"\t\t}\n\t}\n\n\tnewFinalizer := bundleFinalizer{\n\t\tcallbacks:         []bundleFinalizationCallback{},\n\t\tlastValidCallback: time.Now(),\n\t}\n\n\tfor _, idx := range failedIndices {\n\t\tnewFinalizer.callbacks = append(newFinalizer.callbacks, p.bf.callbacks[idx])\n\t\tif newFinalizer.lastValidCallback.Before(p.bf.callbacks[idx].validUntil) {\n\t\t\tnewFinalizer.lastValidCallback = p.bf.callbacks[idx].validUntil\n\t\t}\n\t}\n\n\tp.bf = &newFinalizer\n\n\tif len(failedIndices) > 0 {\n\t\treturn errors.Errorf(\"Plan %v failed %v callbacks\", p.ID(), len(failedIndices))\n\t}\n\treturn nil\n}\n\n// GetExpirationTime returns the last expiration time of any of the callbacks registered by the bundleFinalizer.\n// Once we have passed this time, it is safe to move this plan to inactive without missing any valid callbacks.\nfunc (p *Plan) GetExpirationTime() time.Time {\n\treturn p.bf.lastValidCallback\n}\n\n// Down takes the plan and associated units down. Does not panic.\nfunc (p *Plan) Down(ctx context.Context) error {\n\t// Technically racy, but only one thread calls this method on the plan.\n\tif p.getStatus() == Down {\n\t\treturn nil // ok: already down\n\t}\n\tp.setStatus(Down)\n","sourceCodeStart":180,"sourceCodeEnd":216,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/plan.go#L180-L216","documentation":"Plan.Finalize() runs each registered bundleFinalizer callback whose validity window (validUntil) has not expired and collects the indices of callbacks that failed. If any failed, it replaces the finalizer and returns an aggregate error naming the plan and the count of failed callbacks.","triggerScenarios":"One or more callbacks registered via bundleFinalizer return errors when invoked during Finalize() — e.g. an SDF checkpoint/cleanup callback (like a watermark or state cleanup) fails.","commonSituations":"Custom bundle finalizers whose callbacks touch external resources (files, DBs) that are unavailable at finalize time; callback validity windows raced with the cleanup; SDF checkpoint bookkeeping failing after long bundles.","solutions":["Find which callback(s) failed by instrumenting/logging inside the registered callbacks","Fix the failing callback's resource access (connectivity, permissions) or make it tolerant of already-cleaned-up state","Check callback validUntil windows — expired callbacks are skipped, near-expiry ones may race; extend the window if appropriate","Retry the Finalize() call if the underlying operation is transient"],"exampleFix":"null","handlingStrategy":"try-catch","validationCode":"// Make callbacks cheap and side-effect-tolerant before registration\nbf.Callback(callbackID, validUntil, func() error { return cleanupIfNotAlreadyDone() })","typeGuard":null,"tryCatchPattern":"if err := plan.Finalize(); err != nil {\n    if strings.Contains(err.Error(), \"failed\") {\n        // some callbacks failed; inspect callback logs, then optionally retry\n        return plan.Finalize()\n    }\n    return err\n}","preventionTips":["Make bundle callbacks idempotent (safe to re-run or already-done)","Keep callbacks fast and independent of external resources when possible","Set generous validUntil windows to avoid expiry races","Log per-callback errors inside the callback itself for diagnosability"],"tags":["go","apache-beam","execution-plan","callbacks"],"backgroundTag":"internal-invariant-violation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}