apache/flink · error · UnsupportedOperationException

The key is only defined for keyed operator

Error message

The key is only defined for keyed operator

What it means

Thrown by ProcessOperator.currentKey, the base-class stub that non-keyed process operators inherit: accessing the current key is only valid on keyed operators because only they have a key context. Any key access (e.g. from oldKey during watermark/state access) on a non-keyed operator hits this sentinel UnsupportedOperationException.

Solutions

1. Address the cause reported by the error message: The key is only defined for keyed operator
2. Verify the inputs, configuration values, and classpath/dependency setup related to this operation, then retry.

Example fix

Correct the condition described ("The key is only defined for keyed operator") and rerun the job or command.

When it happens

Trigger: Triggered at runtime when the operation fails because: The key is only defined for keyed operator.

Common situations: Commonly caused by misconfiguration, missing dependencies or files, unsupported types or operations, or invalid user input leading to: The key is only defined for keyed operator.

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/981901ce96aa8699. Report an issue: GitHub.

Appendix: source

Thrown at flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/ProcessOperator.java:169

                // time
                eventTimeWatermarkHandler.processWatermark(watermark.getWatermark(), 0);
            } else {
                output.emitWatermark(watermark);
            }
        }
    }

    protected TimestampCollector<OUT> getOutputCollector() {
        return new OutputCollector<>(output);
    }

    @Override
    public void endInput() throws Exception {
        userFunction.endInput(nonPartitionedContext);
    }

    protected Object currentKey() {
        throw new UnsupportedOperationException("The key is only defined for keyed operator");
    }

    protected BiConsumer<Runnable, Object> getProcessorWithKey() {
        if (isAsyncKeyOrderedProcessingEnabled()) {
            return (r, k) -> asyncProcessWithKey(k, r::run);
        } else {
            return (r, k) -> {
                Object oldKey = currentKey();
                setCurrentKey(k);
                try {
                    r.run();
                } finally {
                    setCurrentKey(oldKey);
                }
            };
        }
    }

View on GitHub (pinned to 2f3c205e92)