apache/beam · error

ProcessElement uses a TimerProvider, but is not keyed. All s

Error message

ProcessElement uses a TimerProvider, but is not keyed. All stateful DoFns must take a key/value pair as an input.

What it means

Like state, timers in Beam Go are keyed: each timer belongs to a key so expiry events can be routed per key. This error is thrown when ProcessElement declares a TimerProvider parameter but the DoFn's input is a single (unkeyed) element, making it impossible to scope timers.

Source

Thrown at sdks/go/pkg/beam/core/graph/fn.go:1424

	_, peNum, peExists := fn.methods[processElementName].Emits()

	if otExists == peExists {
		if otNum != peNum {
			return fmt.Errorf("OnTimer and ProcessElement functions for DoFn should have exactly same emitters, no. of emitters used in OnTimer: %v, no. of emitters used in ProcessElement: %v", otNum, peNum)
		}
	} else {
		return fmt.Errorf("OnTimer and ProcessElement functions for DoFn should have exactly same emitters, emitters used in OnTimer: %v, emitters used in ProcessElement: %v", otExists, peExists)
	}

	return nil
}

func validateTimer(fn *DoFn, numIn mainInputs) error {
	pt, fieldNames := fn.PipelineTimers()

	if _, ok := fn.methods[processElementName].TimerProvider(); ok {
		if numIn == MainSingle {
			err := errors.Errorf("ProcessElement uses a TimerProvider, but is not keyed")
			return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but is not keyed. "+
				"All stateful DoFns must take a key/value pair as an input.")
		}
		if len(pt) == 0 {
			err := errors.New("ProcessElement uses a TimerProvider, but no Timer fields are defined in the DoFn")
			return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but no timer fields are defined in the DoFn"+
				", Ensure that your DoFn exports the Timer fields used to set and clear timers.")
		}
		timerKeys := make(map[string]string)
		for i, t := range pt {
			for timerFamilyID := range t.Timers() {
				if timer, ok := timerKeys[timerFamilyID]; ok {
					err := errors.Errorf("Duplicate timer key %v", timerFamilyID)
					return errors.SetTopLevelMsgf(err, "Duplicate timer family ID %v used by struct fields %v and %v. Ensure that timer family IDs are unique per DoFn", timerFamilyID, timer, fieldNames[i])
				}
				timerKeys[timerFamilyID] = fieldNames[i]
			}
		}

View on GitHub (pinned to 12126d8942)

Solutions

1. Key the input: change ProcessElement to accept a KV[K, V] and key the upstream PCollection (e.g. beam.AddFixedKey).
2. Remove the TimerProvider parameter if timers aren't needed.
3. Confirm the PCollection feeding this DoFn is of KV type.

Example fix

// before
func (fn *BatchFn) ProcessElement(ctx context.Context, tp timer.Provider, line string) error { ... }
// after
func (fn *BatchFn) ProcessElement(ctx context.Context, tp timer.Provider, k beam.KV[string, string]) error { ... }
// pipe: beam.AddFixedKey(s, input)

// ensure the input to timer DoFns is keyed
if !isKeyedPCollection(input) {
    input = beam.AddFixedKey(s, input)
}

func isKeyedElem(v interface{}) bool {
    switch v.(type) {
    case beam.KV[interface{}, interface{}]:
        return true
    }
    return false
}

if err := beam.Run(ctx, p); err != nil {
    if strings.Contains(err.Error(), "uses a TimerProvider, but is not keyed") {
        log.Fatalf("timer DoFn %T requires a KV input: %v", dofn, err)
    }
    return err
}

When it happens

Trigger: Declaring a timer.Provider parameter in ProcessElement while the input element is not a KV pair, e.g. ProcessElement(ctx, tp timer.Provider, line string).

Common situations: Adding timers to an existing unkeyed DoFn; copying timer example code into a pipeline whose PCollections aren't keyed; forgetting the beam.AddFixedKey step.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• OnTimer function is defined for the DoFn but no TimerProvide
• OnTimer and ProcessElement functions for DoFn should have ex
• OnTimer and ProcessElement functions for DoFn should have ex
• ProcessElement uses a StateProvider, but is not keyed. All s
• OnTimer function not defined for DoFn: %v. Ensure that OnTim

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