{"record":{"id":"407c11e74ce7a5cf","repo":"apache/flink","slug":"failed-to-serialize-executionplan","errorCode":null,"errorMessage":"Failed to serialize ExecutionPlan.","messagePattern":"Failed to serialize ExecutionPlan\\.","errorType":"exception","errorClass":"CompletionException","httpStatus":null,"severity":"error","filePath":"flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java","lineNumber":384,"sourceCode":"\n    @Override\n    public CompletableFuture<JobID> submitJob(@Nonnull ExecutionPlan executionPlan) {\n        CompletableFuture<java.nio.file.Path> executionPlanFileFuture =\n                CompletableFuture.supplyAsync(\n                        () -> {\n                            try {\n                                final java.nio.file.Path executionPlanFile =\n                                        Files.createTempFile(\n                                                \"flink-executionPlan-\" + executionPlan.getJobID(),\n                                                \".bin\");\n                                try (ObjectOutputStream objectOut =\n                                        new ObjectOutputStream(\n                                                Files.newOutputStream(executionPlanFile))) {\n                                    objectOut.writeObject(executionPlan);\n                                }\n                                return executionPlanFile;\n                            } catch (IOException e) {\n                                throw new CompletionException(\n                                        new FlinkException(\n                                                \"Failed to serialize ExecutionPlan.\", e));\n                            }\n                        },\n                        executorService);\n\n        CompletableFuture<Tuple2<JobSubmitRequestBody, Collection<FileUpload>>> requestFuture =\n                executionPlanFileFuture.thenApply(\n                        executionPlanFile -> {\n                            List<String> jarFileNames = new ArrayList<>(8);\n                            List<JobSubmitRequestBody.DistributedCacheFile> artifactFileNames =\n                                    new ArrayList<>(8);\n                            Collection<FileUpload> filesToUpload = new ArrayList<>(8);\n\n                            filesToUpload.add(\n                                    new FileUpload(\n                                            executionPlanFile, RestConstants.CONTENT_TYPE_BINARY));\n","sourceCodeStart":366,"sourceCodeEnd":402,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java#L366-L402","documentation":"Thrown when RestClusterClient.submitJob fails to serialize the ExecutionPlan to a temporary file via Java ObjectOutputStream. The ExecutionPlan (containing the job graph, user jars, and artifacts) must be written to a .bin file that is then uploaded to the JobManager REST endpoint. This wraps the underlying IOException (disk failure, non-serializable object, or permission error).","triggerScenarios":"Calling submitJob(ExecutionPlan) where the ExecutionPlan graph contains a non-Serializable object; the system temp directory (java.io.tmpdir) is full or not writable; the ObjectOutputStream.writeObject(executionPlan) call fails because an element in the plan's user jars or graph is not serializable.","commonSituations":"User job JAR contains a non-serializable UDF field (e.g., a raw database Connection or Thread object stored in an operator); running in a container with a read-only /tmp; disk pressure on the client machine; classpath mismatch where an object referenced by the ExecutionPlan is from an incompatible classloader.","solutions":["Check the wrapped IOException cause in the stack trace — if it is NotSerializableException, find and mark the offending field as transient or make the enclosing class implement Serializable.","Verify the client's temp directory is writable and has free space: check java.io.tmpdir and disk usage.","Ensure all user jar paths and artifact paths in the ExecutionPlan are accessible from the client's filesystem.","If the error is transient (disk full), free disk space and retry job submission."],"exampleFix":"// before — non-serializable field in a UDF\npublic class MyMapper extends RichMapFunction<String,String> {\n    private Connection dbConn; // not serializable!\n}\n// after — open in open() method, mark transient\npublic class MyMapper extends RichMapFunction<String,String> {\n    private transient Connection dbConn;\n    @Override\n    public void open(Configuration cfg) { dbConn = DriverManager.getConnection(...); }\n}","handlingStrategy":"try-catch","validationCode":"// Validate ExecutionPlan is serializable before submission\ntry (ByteArrayOutputStream bos = new ByteArrayOutputStream();\n     ObjectOutputStream oos = new ObjectOutputStream(bos)) {\n    oos.writeObject(executionPlan);\n} catch (IOException e) {\n    throw new IllegalArgumentException(\"ExecutionPlan is not serializable: \" + e.getMessage(), e);\n}","typeGuard":null,"tryCatchPattern":"try {\n    client.submitJob(executionPlan).get(30, TimeUnit.SECONDS);\n} catch (ExecutionException e) {\n    Throwable cause = ExceptionUtils.stripExecutionException(e);\n    if (cause instanceof FlinkException && cause.getMessage().contains(\"Failed to serialize\")) {\n        // handle serialization failure — inspect NotSerializableException in cause.getCause()\n    }\n}","preventionTips":["Ensure all operator state fields and UDF fields are Serializable or transient.","Test serialization of your job graph in a unit test before deploying.","Monitor client disk space on the /tmp partition."],"tags":["serialization","job-submission","rest-client","io"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}