{"record":{"id":"000d525cd1e49b5a","repo":"apache/flink","slug":"could-not-clone-serializer-instance-of-class-clas","errorCode":null,"errorMessage":"Could not clone serializer instance of class {className}","messagePattern":"Could not clone serializer instance of class (.+?)","errorType":"exception","errorClass":"CloneFailedException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java","lineNumber":687,"sourceCode":"        // kryoRegistrations may be null if this Kryo serializer is deserialized from an old version\n        if (kryoRegistrations == null) {\n            this.kryoRegistrations =\n                    buildKryoRegistrations(\n                            type,\n                            registeredTypes,\n                            registeredTypesWithSerializerClasses,\n                            registeredTypesWithSerializers,\n                            TernaryBoolean.UNDEFINED);\n        }\n    }\n\n    private SerializableSerializer<? extends Serializer<?>> deepCopySerializer(\n            SerializableSerializer<? extends Serializer<?>> original) {\n        try {\n            return InstantiationUtil.clone(\n                    original, Thread.currentThread().getContextClassLoader());\n        } catch (IOException | ClassNotFoundException ex) {\n            throw new CloneFailedException(\n                    \"Could not clone serializer instance of class \" + original.getClass(), ex);\n        }\n    }\n\n    // --------------------------------------------------------------------------------------------\n    // For testing\n    // --------------------------------------------------------------------------------------------\n\n    private void enterExclusiveThread() {\n        // we use simple get, check, set here, rather than CAS\n        // we don't need lock-style correctness, this is only a sanity-check and we thus\n        // favor speed at the cost of some false negatives in this check\n        Thread previous = currentThread;\n        Thread thisThread = Thread.currentThread();\n\n        if (previous == null) {\n            currentThread = thisThread;\n        } else if (previous != thisThread) {","sourceCodeStart":669,"sourceCodeEnd":705,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java#L669-L705","documentation":"KryoSerializer must be duplicated per thread; duplicating is done by deep-copying any SerializableSerializer instances held in Kryo registrations via InstantiationUtil.clone (Java serialization clone with the context classloader). If that Java-serialization clone fails with IOException or ClassNotFoundException, Flink throws CloneFailedException 'Could not clone serializer instance of class <className>'. The root failure is almost always that the custom Kryo serializer class is not serializable or not loadable.","triggerScenarios":"Registering a Kryo serializer instance (registeredTypesWithSerializers / registerTypeWithKryoSerializer) whose class does not implement java.io.Serializable; the serializer instance holding non-serializable fields; the serializer class not being on the user code classloader when the clone happens.","commonSituations":"Users pass a custom 'new MyKryoSerializer()' carrying a connection or configuration object; serializer classes defined in a jar not shipped with the job; serializer written as a lambda or inner class capturing an enclosing non-serializable object.","solutions":["Make the custom Kryo serializer class implement java.io.Serializable and mark any non-serializable fields transient (re-create them in readObject).","Prefer registering the serializer CLASS (registerTypeWithKryoSerializer(Class type, Class<? extends Serializer> serializerClass)) instead of an instance — class-based registration avoids the deep copy entirely.","Ensure the jar containing the serializer class is included in the job submission (flink run --jar / uber-jar).","Remove captured references: make the serializer a static top-level class rather than an inner class or lambda."],"exampleFix":"// before\nenv.registerTypeWithKryoSerializer(MyType.class, new MyTypeSerializer(conn)); // not serializable\n\n// after (class-based registration, no instance to clone)\nenv.registerTypeWithKryoSerializer(MyType.class, MyTypeSerializer.class);\n\n// or make the instance serializable:\npublic class MyTypeSerializer extends Serializer<MyType> implements java.io.Serializable {\n    private static final long serialVersionUID = 1L;\n    private transient Connection conn;\n}","handlingStrategy":"validation","validationCode":"static <T extends com.esotericsoftware.kryo.Serializer<?>> T checkCloneable(T ser) {\n    try {\n        org.apache.flink.util.InstantiationUtil.clone(\n            new org.apache.flink.api.common.typeutils.base.SerializableSerializer<>(ser),\n            Thread.currentThread().getContextClassLoader());\n        return ser;\n    } catch (Exception e) {\n        throw new IllegalArgumentException(\"Serializer \" + ser.getClass() + \" must be Java-serializable\", e);\n    }\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Prefer class-based Kryo serializer registration over instance-based.","Make custom serializers static top-level classes implementing Serializable with serialVersionUID.","Keep serializers stateless; hold no connections or heavy objects."],"tags":["kryo","serialization","custom-serializer","flink-core"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}