apache/flink · error · RuntimeException

Could not serialize comparator into the configuration.

Error message

Could not serialize comparator into the configuration.

What it means

RuntimeComparatorFactory wraps a TypeComparator and serializes it into a job Configuration so it can travel from the client to the cluster. This error means InstantiationUtil.writeObjectToConfig failed while Java-serializing the comparator object. The root cause is almost always a comparator instance that is not java.io.Serializable or that holds a non-serializable field (e.g. a raw Connection, Thread, or anonymous class capture).

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RuntimeComparatorFactory.java:48

    private static final long serialVersionUID = 1L;

    private static final String CONFIG_KEY = "SER_DATA";

    private TypeComparator<T> comparator;

    public RuntimeComparatorFactory() {}

    public RuntimeComparatorFactory(TypeComparator<T> comparator) {
        this.comparator = comparator;
    }

    @Override
    public void writeParametersToConfig(Configuration config) {
        try {
            InstantiationUtil.writeObjectToConfig(comparator, config, CONFIG_KEY);
        } catch (Exception e) {
            throw new RuntimeException("Could not serialize comparator into the configuration.", e);
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public void readParametersFromConfig(Configuration config, ClassLoader cl)
            throws ClassNotFoundException {
        try {
            comparator =
                    (TypeComparator<T>)
                            InstantiationUtil.readObjectFromConfig(config, CONFIG_KEY, cl);
        } catch (ClassNotFoundException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeException("Could not serialize serializer into the configuration.", e);
        }
    }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Make the comparator class implement java.io.Serializable (TypeComparator itself does not extend Serializable), and mark any non-serializable field as transient, re-creating it in readObject or on first use.
2. If the comparator is an anonymous/inner class, convert it to a static top-level or static nested class so it does not capture the enclosing instance.
3. Inspect the 'Caused by' NotSerializableException in the stack trace to find the exact offending class, then fix that class or remove it from the comparator's fields.
4. Prefer using Flink's built-in comparator types (e.g. the ones produced by TypeInformation.createComparator with ExecutionConfig) instead of hand-rolled comparators.

Example fix

// before
public class MyComparator extends TypeComparator<Tuple2<Long,String>> {
    private Connection conn; // not serializable -> writeParametersToConfig fails
}

// after
public class MyComparator extends TypeComparator<Tuple2<Long,String>> implements Serializable {
    private transient Connection conn; // re-open lazily on first compare
}

public static void assertComparatorSerializable(TypeComparator<?> cmp) {
    if (!(cmp instanceof java.io.Serializable)) {
        throw new IllegalStateException("Comparator " + cmp.getClass().getName()
            + " is not java.io.Serializable and will fail writeParametersToConfig");
    }
    org.apache.flink.util.InstantiationUtil.serializeObject(cmp); // force a dry-run
}

try {
    factory.writeParametersToConfig(config);
} catch (RuntimeException e) {
    // inspect e.getCause() for NotSerializableException naming the offending class
    throw new IllegalStateException("Comparator not serializable: " + e.getCause(), e);
}

When it happens

Trigger: Calling writeParametersToConfig (directly, or indirectly by executing a job that uses a RuntimeComparatorFactory, e.g. via keyBy/types on the Java Tuple API or a custom TypeInformation whose comparator has non-serializable state) with a comparator whose class does not implement Serializable, or whose graph includes a non-serializable member. Also triggered when the comparator's own writeObject throws (e.g. a Kryo-backed comparator whose referenced type cannot be serialized).

Common situations: User defines a custom RecordComparator/TypeComparator as an anonymous inner class (implicitly holds the enclosing instance, which is not serializable). Comparator captures a lambda or object holding an open resource. After a Flink upgrade, a comparator dependency class stopped implementing Serializable. Using a comparator that wraps a non-serializable user bean.

Related errors

• Could not serialize serializer into the configuration.
• Failed to serialize ExecutionPlan.
• Cannot copy serializer
• Error during Java serialization.
• Could not duplicate SimpleVersionedSerializer.

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