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

Error message

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

What it means

Beam Go stateful DoFns must operate on keyed data (KV pairs) so state can be scoped per key. This error is thrown during DoFn validation when ProcessElement declares a StateProvider parameter but the DoFn's input is a single element rather than a key/value pair. Without a key, Beam has no way to partition the state.

Source

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

			if method.Ret[0].T != watermarkStateT {
				err := errors.Errorf("mismatched output type in method %v, return %v. got: %v, want: %v",
					watermarkEstimatorStateName, 0, method.Ret[0].T, watermarkStateT)
				return errors.SetTopLevelMsgf(err, "mismatched output type in method %v, "+
					"return value at index %v got: %v, want: %v (from method %v). "+
					"Ensure that all watermark estimators in an SDF are the same type.",
					watermarkEstimatorStateName, 0, method.Ret[0].T, watermarkStateT, watermarkEstimatorStateName)
			}
		}
	}
	return nil
}

func validateState(fn *DoFn, numIn mainInputs) error {
	ps := fn.PipelineState()

	if _, hasSp := fn.methods[processElementName].StateProvider(); hasSp {
		if numIn == MainSingle {
			err := errors.Errorf("ProcessElement uses a StateProvider, but is not keyed")
			return errors.SetTopLevelMsgf(err, "ProcessElement uses a StateProvider, but is not keyed. "+
				"All stateful DoFns must take a key/value pair as an input.")
		}
		if len(ps) == 0 {
			err := errors.Errorf("ProcessElement uses a StateProvider, but noState structs are attached to the DoFn")
			return errors.SetTopLevelMsgf(err, "ProcessElement uses a StateProvider, but no State structs are "+
				"attached to the DoFn. Ensure that you are including the State structs you're using to read/write"+
				"global state as public uppercase member variables.")
		}
		stateKeys := make(map[string]state.PipelineState)
		for _, s := range ps {
			k := s.StateKey()
			if orig, ok := stateKeys[k]; ok {
				err := errors.Errorf("Duplicate state key %v", k)
				return errors.SetTopLevelMsgf(err, "Duplicate state key %v used by %v and %v. Ensure that state keys are"+
					"unique per DoFn", k, orig, s)
			}
			t := s.StateType()

View on GitHub (pinned to 12126d8942)

Solutions

Key the input: change ProcessElement to accept a KV[K, V] and apply beam.ParDo over a KVs PCollection (e.g. via beam.KV or beam.AddFixedKey).
Remove the StateProvider parameter if state isn't actually needed.
Verify the upstream PCollection is keyed (KV[K,V]) before the stateful DoFn.

Example fix

// before
func (fn *CountFn) ProcessElement(ctx context.Context, sp state.Provider, word string) error { ... }
// after
func (fn *CountFn) ProcessElement(ctx context.Context, sp state.Provider, w beam.KV[string, int]) error { ... }
// and pipe a keyed PCollection: beam.ParDo(s, &CountFn{}, beam.AddFixedKey(s, input))

Defensive patterns

Strategy: validation

Validate before calling

// before running: ensure input is keyed
if _, ok := input.(beam.KV[string, string]); !ok {
    input = beam.AddFixedKey(s, input)
}

Type guard

func isKV(v interface{}) bool { _, ok := v.(beam.KV[interface{}, interface{}]); return ok }

Try / catch

err := beam.Run(ctx, p)
if err != nil && strings.Contains(err.Error(), "uses a StateProvider, but is not keyed") {
    log.Fatalf("stateful DoFn %T needs a KV input: %v", dofn, err)
}

Prevention

Always key PCollections feeding stateful DoFns (beam.AddFixedKey or explicit KV)
Keep stateful DoFn ProcessElement signatures in a team convention doc
Validate DoFns in unit tests before pipeline runs

When it happens

Trigger: Declaring a state.Provider (or StateProvider) parameter in ProcessElement while the DoFn's input element is not a KV (two-element struct with K/V fields), e.g. ProcessElement(ctx, sp state.Provider, value string) where input is a plain string.

Common situations: Adding state to an existing stateless DoFn without changing the input to KV; copying stateful example code into a DoFn that consumes unkeyed PCollections.

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

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