apache/flink · error · IllegalArgumentException

Cannot use custom partitioner for a non-grouped GroupReduce

Error message

Cannot use custom partitioner for a non-grouped GroupReduce (AllGroupReduce)

What it means

Thrown by GroupReduceOperatorBase.setCustomPartitioner(Partitioner<?>) when a custom partitioner is set on a GroupReduce operation that has no grouping keys (an 'all-group reduce' / AllGroupReduce). A custom partitioner routes records by key to specific partitions; without a group key, there is no key to partition by, so the partitioner is meaningless.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java:161

            this.combinable = combinable;
        }
    }

    /**
     * Checks whether the operation is combinable.
     *
     * @return True, if the UDF is combinable, false if not.
     * @see #setCombinable(boolean)
     */
    public boolean isCombinable() {
        return this.combinable;
    }

    public void setCustomPartitioner(Partitioner<?> customPartitioner) {
        if (customPartitioner != null) {
            int[] keys = getKeyColumns(0);
            if (keys == null || keys.length == 0) {
                throw new IllegalArgumentException(
                        "Cannot use custom partitioner for a non-grouped GroupReduce (AllGroupReduce)");
            }
            if (keys.length > 1) {
                throw new IllegalArgumentException(
                        "Cannot use the key partitioner for composite keys (more than one key field)");
            }
        }
        this.customPartitioner = customPartitioner;
    }

    public Partitioner<?> getCustomPartitioner() {
        return customPartitioner;
    }

    private TypeComparator<IN> getTypeComparator(
            TypeInformation<IN> typeInfo,
            int[] sortColumns,
            boolean[] sortOrderings,

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Call .groupBy(keyFields) or .groupBy(keySelector) on the DataSet before the reduce operation, so the group reduce is keyed.
2. If you truly want an all-group-reduce, do not set a custom partitioner.
3. Ensure keys are configured on the GroupReduceOperatorBase (getKeyColumns(0) returns a non-empty array) before calling setCustomPartitioner.

Example fix

// before
DataSet<MyType> result = dataSet.reduceGroup(myReducer); // no groupBy -> AllGroupReduce
((GroupReduceOperatorBase) op).setCustomPartitioner(myPartitioner); // throws

// after
DataSet<MyType> result = dataSet.groupBy("keyField").reduceGroup(myReducer); // keyed reduce
((GroupReduceOperatorBase) op).setCustomPartitioner(myPartitioner); // ok

void safeSetCustomPartitioner(GroupReduceOperatorBase<?, ?, ?> op, Partitioner<?> partitioner) {
    if (partitioner == null) { op.setCustomPartitioner(null); return; }
    int[] keys = op.getKeyColumns(0);
    if (keys == null || keys.length == 0) {
        throw new IllegalArgumentException("Cannot use custom partitioner on a non-keyed GroupReduce");
    }
    op.setCustomPartitioner(partitioner);
}

boolean isKeyedGroupReduce(GroupReduceOperatorBase<?, ?, ?> op) {
    int[] keys = op.getKeyColumns(0);
    return keys != null && keys.length > 0;
}

try {
    reduceOp.setCustomPartitioner(partitioner);
} catch (IllegalArgumentException e) {
    if (e.getMessage().contains("non-grouped")) {
        // add a groupBy before the reduce
        log.error("Must call groupBy() before setting custom partitioner");
    }
}

When it happens

Trigger: Calling reduceGroup without specifying .groupBy(...) (making it an all-reduce), then calling setCustomPartitioner(partitioner). Attempting to set a partitioner on a non-keyed group reduce.

Common situations: Forgetting to call groupBy() before reduceGroup(). Building a reduce operation programmatically and setting a partitioner before configuring keys. Misunderstanding that all-group-reduce sends all data to one reducer regardless of partitioning.

Related errors

• Cannot use the key partitioner for composite keys (more than
• Cannot set a UDF as combinable if it does not implement the
• The filter factor cannot be smaller than zero.
• Target field {} was added twice to input {}
• The number of specified keys is different.

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