{"record":{"id":"467b4711319ccf28","repo":"apache/flink","slug":"number-of-input-splits-has-to-be-at-least-1-467b47","errorCode":null,"errorMessage":"Number of input splits has to be at least 1.","messagePattern":"Number of input splits has to be at least 1\\.","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/io/GenericInputFormat.java","lineNumber":53,"sourceCode":"    protected int partitionNumber;\n\n    // --------------------------------------------------------------------------------------------\n\n    @Override\n    public void configure(Configuration parameters) {\n        //\tnothing by default\n    }\n\n    @Override\n    public BaseStatistics getStatistics(BaseStatistics cachedStatistics) throws IOException {\n        // no statistics available, by default.\n        return cachedStatistics;\n    }\n\n    @Override\n    public GenericInputSplit[] createInputSplits(int numSplits) throws IOException {\n        if (numSplits < 1) {\n            throw new IllegalArgumentException(\"Number of input splits has to be at least 1.\");\n        }\n\n        numSplits = (this instanceof NonParallelInput) ? 1 : numSplits;\n        GenericInputSplit[] splits = new GenericInputSplit[numSplits];\n        for (int i = 0; i < splits.length; i++) {\n            splits[i] = new GenericInputSplit(i, numSplits);\n        }\n        return splits;\n    }\n\n    @Override\n    public DefaultInputSplitAssigner getInputSplitAssigner(GenericInputSplit[] splits) {\n        return new DefaultInputSplitAssigner(splits);\n    }\n\n    // --------------------------------------------------------------------------------------------\n\n    @Override","sourceCodeStart":35,"sourceCodeEnd":71,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/io/GenericInputFormat.java#L35-L71","documentation":"Thrown by GenericInputFormat.createInputSplits(numSplits) when numSplits < 1. The framework calls createInputSplits with the job's configured parallelism to produce GenericInputSplit partitions; a non-positive value is meaningless and rejected. Note the check runs before the NonParallelInput special-case, so even parallelism-1 sources must pass a positive count.","triggerScenarios":"The execution environment / source is configured with parallelism <= 0, or a custom source driver invokes createInputSplits(0) / createInputSplits(-1). It can also surface from an ExecutionConfig or env.setDefaultParallelism set to a non-positive value.","commonSituations":"env.setParallelism(0) by mistake; a source whose parallelism is derived from a config/variable that resolved to 0 or negative (e.g. an unset env var parsed as 0); programmatic submission with an explicit zero split count; tests that mock the input format and call createInputSplits(0).","solutions":["Set a positive parallelism on the environment or the source operator (>= 1).","If deriving parallelism from configuration, validate it is >= 1 before job submission and fall back to a sane default like the available parallelism or 1.","For non-parallel sources, ensure the underlying type implements NonParallelInput so the special-case reduces to 1 split — but still pass numSplits >= 1."],"exampleFix":"// before\nenv.setParallelism(0);\n// after\nenv.setParallelism(1); // or omit to use default parallelism","handlingStrategy":"validation","validationCode":"int parallelism = env.getParallelism();\nif (parallelism < 1) {\n    throw new IllegalStateException(\"Parallelism must be >= 1, got \" + parallelism);\n}\n// or when deriving from config\nint p = Integer.parseInt(config.get(\"source.parallelism\").orElse(\"1\"));\nif (p < 1) p = 1;","typeGuard":"static int positiveParallelism(int p) {\n    if (p < 1) throw new IllegalArgumentException(\"parallelism must be >= 1, got \" + p);\n    return p;\n}","tryCatchPattern":"try {\n    source.createInputSplits(parallelism);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"at least 1\")) {\n        env.setParallelism(1);\n        return source.createInputSplits(1);\n    }\n    throw e;\n}","preventionTips":["Always set a positive parallelism on the env or source (>= 1).","Validate any config-derived parallelism before submission; default to 1 if unset.","For non-parallel sources, implement NonParallelInput but still pass numSplits >= 1.","Unit-test custom sources with createInputSplits(1)."],"tags":["input-format","parallelism","configuration","runtime"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}