{"record":{"id":"cbdbf154440de8fd","repo":"apache/flink","slug":"could-not-serialize-serializer-into-the-configurat-cbdbf1","errorCode":null,"errorMessage":"Could not serialize serializer into the configuration.","messagePattern":"Could not serialize serializer into the configuration\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RuntimeSerializerFactory.java","lineNumber":59,"sourceCode":"    // Because we read the class from the TaskConfig and instantiate ourselves\n    public RuntimeSerializerFactory() {}\n\n    public RuntimeSerializerFactory(TypeSerializer<T> serializer, Class<T> clazz) {\n        if (serializer == null || clazz == null) {\n            throw new NullPointerException();\n        }\n\n        this.clazz = clazz;\n        this.serializer = serializer;\n    }\n\n    @Override\n    public void writeParametersToConfig(Configuration config) {\n        try {\n            InstantiationUtil.writeObjectToConfig(clazz, config, CONFIG_KEY_CLASS);\n            InstantiationUtil.writeObjectToConfig(serializer, config, CONFIG_KEY_SER);\n        } catch (Exception e) {\n            throw new RuntimeException(\"Could not serialize serializer into the configuration.\", e);\n        }\n    }\n\n    @Override\n    public void readParametersFromConfig(Configuration config, ClassLoader cl)\n            throws ClassNotFoundException {\n        if (config == null || cl == null) {\n            throw new NullPointerException();\n        }\n\n        try {\n            this.clazz = InstantiationUtil.readObjectFromConfig(config, CONFIG_KEY_CLASS, cl);\n            this.serializer = InstantiationUtil.readObjectFromConfig(config, CONFIG_KEY_SER, cl);\n        } catch (ClassNotFoundException e) {\n            throw e;\n        } catch (Exception e) {\n            throw new RuntimeException(\"Could not load deserializer from the configuration.\", e);\n        }","sourceCodeStart":41,"sourceCodeEnd":77,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RuntimeSerializerFactory.java#L41-L77","documentation":"RuntimeSerializerFactory.writeParametersToConfig Java-serializes both the type class and the TypeSerializer into the job Configuration. This error means that serialization failed, most commonly because the TypeSerializer instance (or the Class object's surrounding graph) is not java.io.Serializable. Flink's own serializers usually are; user-defined serializers frequently are not.","triggerScenarios":"Executing a job where a RuntimeSerializerFactory wraps a custom TypeSerializer that does not implement Serializable, or whose fields reference non-serializable objects. Also triggered if writeObject on the serializer throws (final fields, failing custom writeObject).","commonSituations":"Custom TypeSerializer registered via env.registerTypeWithSerializer or provided in a TypeInformation that keeps a non-serializable helper (schema registry client, pooled buffer, model object). Serializer implemented as a non-static inner class. Serializer capturing 'this' of the enclosing job class which itself holds an ExecutionEnvironment (a classic NotSerializableException).","solutions":["Make the custom TypeSerializer implement java.io.Serializable and give it a stable serialVersionUID.","Convert inner/anonymous serializer classes to static nested or top-level classes so they do not capture the enclosing (often non-serializable) instance.","Find the offending field from the nested NotSerializableException and mark it transient, rebuilding it lazily after deserialization.","Where possible, register a serializer factory pattern (like RuntimeSerializerFactory itself) or use Flink-provided TypeInformation/serializer instances, which are already serializable."],"exampleFix":"// before\npublic class MySer extends TypeSerializer<MyPojo> { // not Serializable -> 683\n    private final Codec codec = Codec.create();\n}\n\n// after\npublic class MySer extends TypeSerializer<MyPojo> implements Serializable {\n    private static final long serialVersionUID = 1L;\n    private transient Codec codec;\n    private Codec codec() { if (codec == null) codec = Codec.create(); return codec; }\n}","handlingStrategy":"validation","validationCode":"public static void assertSerializerSerializable(TypeSerializer<?> ser) {\n    if (!(ser instanceof java.io.Serializable)) {\n        throw new IllegalStateException(\"Serializer \" + ser.getClass().getName()\n            + \" is not Serializable; RuntimeSerializerFactory.writeParametersToConfig will fail\");\n    }\n    org.apache.flink.util.InstantiationUtil.serializeObject(ser);\n}","typeGuard":null,"tryCatchPattern":"try {\n    factory.writeParametersToConfig(config);\n} catch (RuntimeException e) {\n    Throwable c = e.getCause(); // NotSerializableException names the offending class\n    throw new IllegalStateException(\"Serializer graph not serializable: \" + c, e);\n}","preventionTips":["Every custom TypeSerializer must implement Serializable.","Avoid serializers as inner classes of the job class (they capture the non-serializable environment).","Unit-test serializer round-trip through Configuration for each custom serializer."],"tags":["serialization","type-serializer","java-serialization","job-submission"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}