{"record":{"id":"75439581133dd547","repo":"apache/flink","slug":"custom-partitioners-can-only-be-used-with-keys-tha","errorCode":null,"errorMessage":"Custom partitioners can only be used with keys that have one key field.","messagePattern":"Custom partitioners can only be used with keys that have one key field\\.","errorType":"exception","errorClass":"InvalidProgramException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/operators/Keys.java","lineNumber":169,"sourceCode":"        public TypeInformation<?>[] getKeyFieldTypes() {\n            TypeInformation<?>[] fieldTypes = new TypeInformation[keyFields.size()];\n            for (int i = 0; i < keyFields.size(); i++) {\n                fieldTypes[i] = keyFields.get(i).getType();\n            }\n            return fieldTypes;\n        }\n\n        @Override\n        public TypeInformation<?>[] getOriginalKeyFieldTypes() {\n            return originalKeyTypes;\n        }\n\n        @Override\n        public <E> void validateCustomPartitioner(\n                Partitioner<E> partitioner, TypeInformation<E> typeInfo) {\n\n            if (keyFields.size() != 1) {\n                throw new InvalidProgramException(\n                        \"Custom partitioners can only be used with keys that have one key field.\");\n            }\n\n            if (typeInfo == null) {\n                // try to extract key type from partitioner\n                try {\n                    typeInfo = TypeExtractor.getPartitionerTypes(partitioner);\n                } catch (Throwable t) {\n                    // best effort check, so we ignore exceptions\n                }\n            }\n\n            // only check if type is known and not a generic type\n            if (typeInfo != null && !(typeInfo instanceof GenericTypeInfo)) {\n                // check equality of key and partitioner type\n                if (!keyType.equals(typeInfo)) {\n                    throw new InvalidProgramException(\n                            \"The partitioner is incompatible with the key type. \"","sourceCodeStart":151,"sourceCodeEnd":187,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/operators/Keys.java#L151-L187","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Reduce the key to a single field before applying partitionCustom (e.g. a KeySelector returning just the partitioning attribute).","If you need multi-field distribution, use hash or range partitioning instead of a custom Partitioner.","Move partitionCustom upstream where the key is still single-field."],"exampleFix":"// before (sel returns Tuple2)\nds.partitionCustom(p, sel);\n// after (sel returns only the partitioning field)\nds.partitionCustom(p, (KeySelector<IN,String>) e -> e.tenantId);","handlingStrategy":"validation","validationCode":"if (keys.getNumberOfKeyFields() != 1) {\n    throw new IllegalArgumentException(\n        \"Custom partitioner requires a single-field key; got \" + keys.getNumberOfKeyFields());\n}","typeGuard":"static boolean isSingleFieldKey(Keys<?> k) { return k.getNumberOfKeyFields() == 1; }","tryCatchPattern":null,"preventionTips":["Apply partitionCustom only to single-field keys.","Use a KeySelector that returns the partitioning attribute alone.","For multi-field distribution, use hash or range partitioning."],"tags":["flink-core","keys","custom-partitioner","partitioning","validation"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}