apache/beam · error

found uses of features unimplemented in prism in job

Error message

found %v uses of features unimplemented in prism in job %v:
%v

What it means

During Prepare, Prism collects a list of pipeline features it does not implement (via the check() helper, e.g. triggered side inputs). If any were found, it aggregates them into a joinError and rejects the job with this message listing each unimplemented feature. The job is marked Failed and never runs.

Solutions

1. Read the listed %v entries to identify each unimplemented feature and remove or restructure it in the pipeline.
2. Replace triggered side inputs on unbounded global windows with plain side inputs or window-agnostic lookups.
3. Run the pipeline on a runner that supports the feature (DirectRunner, Flink) instead of prism.
4. Check github.com/apache/beam issues (e.g. #31438) for the feature's prism support status before porting.

Example fix

// before
w := window.NewTriggeredFixedWindows(window.NewTrigger(...), ...)
beam.ParDo(s, sideInputDoFn, w) // prism: unsupported

// after
w := window.NewFixedWindows(...) // no exotic trigger
beam.ParDo(s, sideInputDoFn, w)

// Pre-check windowing strategy before submitting to prism
if ws.GetTrigger() != nil && usesSideInputs(pipeline) {
    return errors.New("pipeline uses triggered side inputs unsupported by prism")
}

err := runner.Prepare(req)
if err != nil && strings.Contains(err.Error(), "unimplemented in prism") {
    for _, line := range strings.Split(err.Error(), "\n")[1:] {
        log.Printf("unsupported feature: %s", line) // migrate each feature
    }
}

When it happens

Trigger: Submitting a pipeline whose WindowingStrategy triggers, side inputs (e.g. unbounded global window triggered side inputs), or other transforms hit a check() failure inside Prepare — any single unimplemented feature produces this aggregate error.

Common situations: Porting a Flink/Spark/Direct-runner pipeline to prism that relies on windowing features prism lacks (e.g. side inputs ready mid-window, exotic triggers); using TestStream windows+late data semantics prism doesn't support.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• capacity of cache cannot be negative, got
• computeFacts: two producers for one PCollection
• could not unmarshal iterable coder from
• could not unmarshal nullable coder from
• could not unmarshal sharded_key coder from

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

Appendix: source

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

		} else if hasStatefulTriggers(ws.GetTrigger()) {
			// Technically for any merging windows, but per the above, we only support session windows presently.
			check("WindowingStrategy: Using stateful triggers with merging windows isn't currently supported in prism. See https://github.com/apache/beam/issues/31438 for information.", prototext.Format(ws))
		}
		check("WindowingStrategy.OnTimeBehavior", ws.GetOnTimeBehavior(), pipepb.OnTimeBehavior_FIRE_IF_NONEMPTY, pipepb.OnTimeBehavior_FIRE_ALWAYS)

		// Allow earliest and latest in pane to unblock running python tasks.
		// Tests actually using the set behavior will fail.
		check("WindowingStrategy.OutputTime", ws.GetOutputTime(), pipepb.OutputTime_END_OF_WINDOW,
			pipepb.OutputTime_EARLIEST_IN_PANE, pipepb.OutputTime_LATEST_IN_PANE)

		if hasUnsupportedTriggers(ws.GetTrigger()) {
			check("WindowingStrategy.Trigger", ws.GetTrigger().String())
		}
	}
	if len(errs) > 0 {
		jErr := &joinError{errs: errs}
		slog.Error("unable to run job", slog.String("cause", "unimplemented features"), slog.String("jobname", req.GetJobName()), slog.String("errors", jErr.Error()))
		err := fmt.Errorf("found %v uses of features unimplemented in prism in job %v:\n%v", len(errs), req.GetJobName(), jErr)
		job.Failed(err)
		return nil, err
	}
	return &jobpb.PrepareJobResponse{
		PreparationId:       job.key,
		StagingSessionToken: job.key,
		ArtifactStagingEndpoint: &pipepb.ApiServiceDescriptor{
			Url: s.Endpoint(),
		},
	}, nil
}

func hasUnsupportedTriggers(tpb *pipepb.Trigger) bool {
	if tpb == nil {
		return false
	}

	unsupported := false

View on GitHub (pinned to 12126d8942)