{"record":{"id":"b97cec1c8b9172a0","repo":"apache/flink","slug":"reached-the-end-of-the-collection-this-could-be-c","errorCode":null,"errorMessage":"Reached the end of the collection. This could be caused by issues with the serializer or by calling the map() function more times than there are elements in the collection. Make sure that you set the number of records to be produced by the DataGeneratorSource equal to the number of elements in the collection.","messagePattern":"Reached the end of the collection\\. This could be caused by issues with the serializer or by calling the map\\(\\) function more times than there are elements in the collection\\. Make sure that you set the number of records to be produced by the DataGeneratorSource equal to the number of elements in the collection\\.","errorType":"exception","errorClass":"NoSuchElementException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-datagen/src/main/java/org/apache/flink/connector/datagen/functions/FromElementsGeneratorFunction.java","lineNumber":134,"sourceCode":"    }\n\n    @Override\n    public OUT map(Long nextIndex) throws Exception {\n        // Move iterator to the required position in case of failure recovery\n        while (numElementsEmitted < nextIndex) {\n            numElementsEmitted++;\n            tryDeserialize(serializer, input);\n        }\n        numElementsEmitted++;\n        return tryDeserialize(serializer, input);\n    }\n\n    private OUT tryDeserialize(TypeSerializer<OUT> serializer, DataInputView input)\n            throws IOException {\n        try {\n            return serializer.deserialize(input);\n        } catch (EOFException eof) {\n            throw new NoSuchElementException(\n                    \"Reached the end of the collection. This could be caused by issues with the serializer or by calling the map() function more times than there are elements in the collection. Make sure that you set the number of records to be produced by the DataGeneratorSource equal to the number of elements in the collection.\");\n        } catch (Exception e) {\n            throw new IOException(\n                    \"Failed to deserialize an element from the source. \"\n                            + \"If you are using user-defined serialization (Value and Writable types), check the \"\n                            + \"serialization functions.\\nSerializer is \"\n                            + serializer,\n                    e);\n        }\n    }\n\n    // For backward compatibility: Supports legacy usage of\n    // StreamExecutionEnvironment#fromElements() which lacked type information and relied on the\n    // returns() method. See FLINK-21386 for details.\n    @Override\n    public void setOutputType(TypeInformation<OUT> outTypeInfo, ExecutionConfig executionConfig) {\n        Preconditions.checkState(\n                elements != null,","sourceCodeStart":116,"sourceCodeEnd":152,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-datagen/src/main/java/org/apache/flink/connector/datagen/functions/FromElementsGeneratorFunction.java#L116-L152","documentation":"FromElementsGeneratorFunction#tryDeserialize catches EOFException and throws NoSuchElementException with guidance: the input stream was exhausted before the expected number of elements were deserialized. This means either the serializer is buggy (writes/reads mismatched length) or the DataGeneratorSource was told to produce more records than there are elements.","triggerScenarios":"DataGeneratorSource.count is set higher than the number of serialized elements, so map() is called past the end of the byte buffer. Also triggered by a serializer that misreports record length.","commonSituations":"Mismatch between the number of elements passed to FromElementsGeneratorFunction and the count configured on DataGeneratorSource; a buggy custom TypeSerializer whose serialize/deserialize are not length-consistent.","solutions":["Set DataGeneratorSource.count equal to the exact number of elements provided to FromElementsGeneratorFunction.","If using a collection, pass collection.size() as the count rather than a hardcoded number.","Verify the TypeSerializer is round-trip consistent (write then read yields the same number of records).","Avoid reusing a FromElementsGeneratorFunction instance whose internal input position has advanced."],"exampleFix":"// before\nList<String> elems = List.of(\"a\", \"b\", \"c\");\nvar fn = new FromElementsGeneratorFunction<>(Types.STRING, elems);\nvar source = DataGeneratorSource.builder(fn, 10L, RateLimiterStrategy.noop(), Types.STRING); // 10 != 3\n// after\nvar source = DataGeneratorSource.builder(fn, (long) elems.size(), RateLimiterStrategy.noop(), Types.STRING);","handlingStrategy":"validation","validationCode":"// Ensure the source count equals the number of elements:\nList<OUT> elems = new ArrayList<>();\nelements.forEach(elems::add);\nlong count = elems.size();\nDataGeneratorSource.<OUT>builder(fn, count, RateLimiterStrategy.noop(), typeInfo);","typeGuard":null,"tryCatchPattern":"try {\n    return fn.map(index);\n} catch (NoSuchElementException e) {\n    // count exceeded element count; stop producing\n    throw e;\n}","preventionTips":["Set DataGeneratorSource.count to the exact element count.","Derive count from collection.size() instead of hardcoding.","Verify the TypeSerializer is round-trip length-consistent."],"tags":["serialization","from-elements","data-generator","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}