apache/flink · error · InvalidProgramException

Custom partitioners can only be used with keys that have one

Error message

Custom partitioners can only be used with keys that have one key field.

What it means

Thrown by SelectorFunctionKeys.validateCustomPartitioner() when a custom Partitioner is attached but the key resolves to more than one flat field. A Partitioner<E> maps a single key value to a partition index; multi-field keys have no single value to pass, so custom partitioning is only legal for single-field keys.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/operators/Keys.java:169

        public TypeInformation<?>[] getKeyFieldTypes() {
            TypeInformation<?>[] fieldTypes = new TypeInformation[keyFields.size()];
            for (int i = 0; i < keyFields.size(); i++) {
                fieldTypes[i] = keyFields.get(i).getType();
            }
            return fieldTypes;
        }

        @Override
        public TypeInformation<?>[] getOriginalKeyFieldTypes() {
            return originalKeyTypes;
        }

        @Override
        public <E> void validateCustomPartitioner(
                Partitioner<E> partitioner, TypeInformation<E> typeInfo) {

            if (keyFields.size() != 1) {
                throw new InvalidProgramException(
                        "Custom partitioners can only be used with keys that have one key field.");
            }

            if (typeInfo == null) {
                // try to extract key type from partitioner
                try {
                    typeInfo = TypeExtractor.getPartitionerTypes(partitioner);
                } catch (Throwable t) {
                    // best effort check, so we ignore exceptions
                }
            }

            // only check if type is known and not a generic type
            if (typeInfo != null && !(typeInfo instanceof GenericTypeInfo)) {
                // check equality of key and partitioner type
                if (!keyType.equals(typeInfo)) {
                    throw new InvalidProgramException(
                            "The partitioner is incompatible with the key type. "

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Reduce the key to a single field before applying partitionCustom (e.g. a KeySelector returning just the partitioning attribute).
2. If you need multi-field distribution, use hash or range partitioning instead of a custom Partitioner.
3. Move partitionCustom upstream where the key is still single-field.

Example fix

// before (sel returns Tuple2)
ds.partitionCustom(p, sel);
// after (sel returns only the partitioning field)
ds.partitionCustom(p, (KeySelector<IN,String>) e -> e.tenantId);

if (keys.getNumberOfKeyFields() != 1) {
    throw new IllegalArgumentException(
        "Custom partitioner requires a single-field key; got " + keys.getNumberOfKeyFields());
}

static boolean isSingleFieldKey(Keys<?> k) { return k.getNumberOfKeyFields() == 1; }

When it happens

Trigger: Calling .partitionCustom(partitioner) on a keyBy/Grouping whose KeySelector returns a Tuple (multiple flat fields), or attaching a partitioner to a key defined by multiple field positions.

Common situations: Trying to use a custom partitioner on a composite key; assuming partitionCustom works like range/hash partitioning on multi-field keys; refactoring a single-field key into a tuple without updating the partitioner usage.

Related errors

• The partitioner is incompatible with the key type. Partition
• Return type {keyType} of KeySelector {keyExtractor.getClass(
• Tuple size must be greater than 0. Size: {type.getArity()}
• The grouping fields must not be empty.
• This type ({ffd.getType()}) cannot be used as key.

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