apache/flink · error · RuntimeException

{e.getMessage()}

Error message

{e.getMessage()}

What it means

In setOutputType (and the private trySerialize), if serializeElements throws IOException the code wraps it in a RuntimeException carrying only e.getMessage() and the cause. This happens when the framework sets the output type after construction and re-serialization under the resolved serializer fails for the given elements.

Source

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

    // 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,
                "The output type should've been specified before shipping the graph to the cluster");
        checkIterable(elements, outTypeInfo.getTypeClass());
        TypeSerializer<OUT> newSerializer =
                outTypeInfo.createSerializer(executionConfig.getSerializerConfig());
        if (Objects.equals(serializer, newSerializer)) {
            return;
        }
        serializer = newSerializer;
        try {
            serializeElements(elements);
        } catch (IOException e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    private void trySerialize(Iterable<OUT> elements) {
        try {
            serializeElements(elements);
        } catch (IOException e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    // ------------------------------------------------------------------------
    //  Utilities
    // ------------------------------------------------------------------------

    /**
     * Verifies that all elements in the iterable are non-null, and are of the given class, or a
     * subclass thereof.

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Construct FromElementsGeneratorFunction with the same TypeInformation the framework will infer (use explicit returns()/TypeInformation to avoid divergence).
2. Ensure all elements are serializable under any serializer the resolved TypeInformation would produce.
3. Inspect the wrapped IOException cause for the failing element/field.
4. Avoid passing an ExecutionConfig at construction that differs from the runtime config, which can yield a different serializer.

Example fix

// before: constructor uses one type, framework infers another -> re-serialize fails
new FromElementsGeneratorFunction<>(Types.GENERIC(Object.class), myPojos);
// after: declare the concrete type so setOutputType keeps the same serializer
new FromElementsGeneratorFunction<>(Types.POJO(MyPojo.class), myPojos);

// Use the same TypeInformation the framework will infer to avoid a serializer swap:
TypeInformation<OUT> ti = TypeInformation.of(MyPojo.class);
FromElementsGeneratorFunction<OUT> fn = new FromElementsGeneratorFunction<>(ti, elements);
// Ensure setOutputType will resolve to the same serializer:
OUT sample = elements.iterator().next();
TypeSerializer<OUT> s1 = ti.createSerializer(config.getSerializerConfig());
s1.serialize(sample, new DataOutputViewStreamWrapper(new ByteArrayOutputStream())); // must not throw

try {
    ((OutputTypeConfigurable<OUT>) fn).setOutputType(inferredType, config);
} catch (RuntimeException e) {
    Throwable c = e.getCause() != null ? e.getCause() : e;
    throw new RuntimeException("Re-serialization under inferred type failed: " + inferredType, c);
}

When it happens

Trigger: OutputTypeConfigurable#setOutputType is invoked by the framework with the inferred TypeInformation, the resolved serializer differs from the one used at construction, and re-serializing the elements under the new serializer fails.

Common situations: The element type inferred by Flink differs from what was passed to the constructor, so re-serialization under the new serializer rejects an element; custom types whose serializer behavior depends on config only available at setOutputType time.

Related errors

• Serializing the source elements failed: {e.getMessage()}
• Reached the end of the collection. This could be caused by i
• Failed to deserialize an element from the source. If you are
• Invalid version %d
• Exception in writeRecord

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