{"record":{"id":"8f3a1fde4c9181bf","repo":"apache/flink","slug":"cannot-deserialize-and-unwrap-accumulators-properl-8f3a1f","errorCode":null,"errorMessage":"Cannot deserialize and unwrap accumulators properly.","messagePattern":"Cannot deserialize and unwrap accumulators properly\\.","errorType":"exception","errorClass":"CompletionException","httpStatus":null,"severity":"error","filePath":"flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java","lineNumber":718,"sourceCode":"        final JobAccumulatorsHeaders accumulatorsHeaders = JobAccumulatorsHeaders.getInstance();\n        final JobAccumulatorsMessageParameters accMsgParams =\n                accumulatorsHeaders.getUnresolvedMessageParameters();\n        accMsgParams.jobPathParameter.resolve(jobID);\n        accMsgParams.includeSerializedAccumulatorsParameter.resolve(\n                Collections.singletonList(true));\n\n        CompletableFuture<JobAccumulatorsInfo> responseFuture =\n                sendRequest(accumulatorsHeaders, accMsgParams);\n\n        return responseFuture\n                .thenApply(JobAccumulatorsInfo::getSerializedUserAccumulators)\n                .thenApply(\n                        accumulators -> {\n                            try {\n                                return AccumulatorHelper.deserializeAndUnwrapAccumulators(\n                                        accumulators, loader);\n                            } catch (Exception e) {\n                                throw new CompletionException(\n                                        \"Cannot deserialize and unwrap accumulators properly.\", e);\n                            }\n                        });\n    }\n\n    private CompletableFuture<SavepointInfo> pollSavepointAsync(\n            final JobID jobId, final TriggerId triggerID) {\n        return pollResourceAsync(\n                () -> {\n                    final SavepointStatusHeaders savepointStatusHeaders =\n                            SavepointStatusHeaders.getInstance();\n                    final SavepointStatusMessageParameters savepointStatusMessageParameters =\n                            savepointStatusHeaders.getUnresolvedMessageParameters();\n                    savepointStatusMessageParameters.jobIdPathParameter.resolve(jobId);\n                    savepointStatusMessageParameters.triggerIdPathParameter.resolve(triggerID);\n                    return sendRequest(savepointStatusHeaders, savepointStatusMessageParameters);\n                });\n    }","sourceCodeStart":700,"sourceCodeEnd":736,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java#L700-L736","documentation":"Thrown when getAccumulators cannot deserialize and unwrap serialized user accumulators from the JobAccumulatorsInfo REST response. AccumulatorHelper.deserializeAndUnwrapAccumulators uses the provided ClassLoader to reconstruct accumulator objects from their serialized form. Any deserialization failure (class not found, incompatible class, corrupt data) is caught and wrapped.","triggerScenarios":"Calling getAccumulators(jobID, loader) where the ClassLoader does not have access to the accumulator classes; the accumulator implementation changed between versions; the serialized accumulator bytes are corrupted or from an incompatible serialization format.","commonSituations":"The user accumulator class lives in a user JAR that is not on the client classpath; the client uses a different Flink version than the cluster; custom accumulator implementation was changed without a serialization snapshot migration.","solutions":["Pass a ClassLoader that includes the user JAR containing the accumulator classes to getAccumulators.","Ensure the client and cluster run the same Flink version and user JAR versions.","If using custom accumulators, verify their serialization is forward/backward compatible.","Check for ClassNotFoundException in the exception cause to identify the missing class."],"exampleFix":"// before — default classloader missing user accumulator classes\nclient.getAccumulators(jobId, getClass().getClassLoader());\n// after — use a classloader that includes the user JAR\nURLClassLoader userLoader = new URLClassLoader(userJarUrls, getClass().getClassLoader());\nclient.getAccumulators(jobId, userLoader);","handlingStrategy":"validation","validationCode":"// Validate classloader has accumulator classes before calling getAccumulators\nClassLoader loader = getUserJarClassLoader();\ntry {\n    Class.forName(\"com.example.MyAccumulator\", false, loader);\n} catch (ClassNotFoundException e) {\n    throw new IllegalStateException(\"Accumulator class not on the provided classloader\", e);\n}","typeGuard":null,"tryCatchPattern":"try {\n    Map<String, Object> accs = client.getAccumulators(jobId, loader).get();\n} catch (ExecutionException e) {\n    Throwable cause = ExceptionUtils.stripExecutionException(e);\n    if (cause.getMessage().contains(\"Cannot deserialize and unwrap accumulators\")) {\n        // likely ClassNotFoundException — check classloader\n    }\n}","preventionTips":["Pass a ClassLoader that includes user JARs to getAccumulators.","Keep accumulator implementations Serializable and stable across versions.","Match client and cluster Flink versions."],"tags":["accumulators","rest-client","serialization","classloader"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}