apache/flink · error · NoSuchElementException

Reached the end of the collection. This could be caused by i

Error message

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.

What it means

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.

Source

Thrown at flink-connectors/flink-connector-datagen/src/main/java/org/apache/flink/connector/datagen/functions/FromElementsGeneratorFunction.java:134

    }

    @Override
    public OUT map(Long nextIndex) throws Exception {
        // Move iterator to the required position in case of failure recovery
        while (numElementsEmitted < nextIndex) {
            numElementsEmitted++;
            tryDeserialize(serializer, input);
        }
        numElementsEmitted++;
        return tryDeserialize(serializer, input);
    }

    private OUT tryDeserialize(TypeSerializer<OUT> serializer, DataInputView input)
            throws IOException {
        try {
            return serializer.deserialize(input);
        } catch (EOFException eof) {
            throw new NoSuchElementException(
                    "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.");
        } catch (Exception e) {
            throw new IOException(
                    "Failed to deserialize an element from the source. "
                            + "If you are using user-defined serialization (Value and Writable types), check the "
                            + "serialization functions.\nSerializer is "
                            + serializer,
                    e);
        }
    }

    // For backward compatibility: Supports legacy usage of
    // StreamExecutionEnvironment#fromElements() which lacked type information and relied on the
    // returns() method. See FLINK-21386 for details.
    @Override
    public void setOutputType(TypeInformation<OUT> outTypeInfo, ExecutionConfig executionConfig) {
        Preconditions.checkState(
                elements != null,

View on GitHub (pinned to 2f3c205e92)

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.

Example fix

// before
List<String> elems = List.of("a", "b", "c");
var fn = new FromElementsGeneratorFunction<>(Types.STRING, elems);
var source = DataGeneratorSource.builder(fn, 10L, RateLimiterStrategy.noop(), Types.STRING); // 10 != 3
// after
var source = DataGeneratorSource.builder(fn, (long) elems.size(), RateLimiterStrategy.noop(), Types.STRING);

Defensive patterns

Strategy: validation

Validate before calling

// Ensure the source count equals the number of elements:
List<OUT> elems = new ArrayList<>();
elements.forEach(elems::add);
long count = elems.size();
DataGeneratorSource.<OUT>builder(fn, count, RateLimiterStrategy.noop(), typeInfo);

Try / catch

try {
    return fn.map(index);
} catch (NoSuchElementException e) {
    // count exceeded element count; stop producing
    throw e;
}

Prevention

Set DataGeneratorSource.count to the exact element count.
Derive count from collection.size() instead of hardcoding.
Verify the TypeSerializer is round-trip length-consistent.

When it happens

Trigger: 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.

Common situations: 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.

Related errors

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