{"record":{"id":"304f75abc1db48fe","repo":"apache/flink","slug":"could-not-serialize-comparator-into-the-configurat","errorCode":null,"errorMessage":"Could not serialize comparator into the configuration.","messagePattern":"Could not serialize comparator into the configuration\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RuntimeComparatorFactory.java","lineNumber":48,"sourceCode":"\n    private static final long serialVersionUID = 1L;\n\n    private static final String CONFIG_KEY = \"SER_DATA\";\n\n    private TypeComparator<T> comparator;\n\n    public RuntimeComparatorFactory() {}\n\n    public RuntimeComparatorFactory(TypeComparator<T> comparator) {\n        this.comparator = comparator;\n    }\n\n    @Override\n    public void writeParametersToConfig(Configuration config) {\n        try {\n            InstantiationUtil.writeObjectToConfig(comparator, config, CONFIG_KEY);\n        } catch (Exception e) {\n            throw new RuntimeException(\"Could not serialize comparator into the configuration.\", e);\n        }\n    }\n\n    @SuppressWarnings(\"unchecked\")\n    @Override\n    public void readParametersFromConfig(Configuration config, ClassLoader cl)\n            throws ClassNotFoundException {\n        try {\n            comparator =\n                    (TypeComparator<T>)\n                            InstantiationUtil.readObjectFromConfig(config, CONFIG_KEY, cl);\n        } catch (ClassNotFoundException e) {\n            throw e;\n        } catch (Exception e) {\n            throw new RuntimeException(\"Could not serialize serializer into the configuration.\", e);\n        }\n    }\n","sourceCodeStart":30,"sourceCodeEnd":66,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RuntimeComparatorFactory.java#L30-L66","documentation":"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).","triggerScenarios":"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).","commonSituations":"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.","solutions":["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.","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.","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.","Prefer using Flink's built-in comparator types (e.g. the ones produced by TypeInformation.createComparator with ExecutionConfig) instead of hand-rolled comparators."],"exampleFix":"// before\npublic class MyComparator extends TypeComparator<Tuple2<Long,String>> {\n    private Connection conn; // not serializable -> writeParametersToConfig fails\n}\n\n// after\npublic class MyComparator extends TypeComparator<Tuple2<Long,String>> implements Serializable {\n    private transient Connection conn; // re-open lazily on first compare\n}","handlingStrategy":"validation","validationCode":"public static void assertComparatorSerializable(TypeComparator<?> cmp) {\n    if (!(cmp instanceof java.io.Serializable)) {\n        throw new IllegalStateException(\"Comparator \" + cmp.getClass().getName()\n            + \" is not java.io.Serializable and will fail writeParametersToConfig\");\n    }\n    org.apache.flink.util.InstantiationUtil.serializeObject(cmp); // force a dry-run\n}","typeGuard":null,"tryCatchPattern":"try {\n    factory.writeParametersToConfig(config);\n} catch (RuntimeException e) {\n    // inspect e.getCause() for NotSerializableException naming the offending class\n    throw new IllegalStateException(\"Comparator not serializable: \" + e.getCause(), e);\n}","preventionTips":["Make every custom TypeComparator implement Serializable with an explicit serialVersionUID.","Prefer static nested or top-level comparator classes over anonymous/inner classes.","Dry-run serialization (InstantiationUtil.serializeObject) in a unit test for each custom comparator."],"tags":["serialization","comparator","java-serialization","job-submission"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}