{"record":{"id":"b1a5398ba70bbb08","repo":"apache/flink","slug":"cannot-deserialize-and-unwrap-accumulators-properl-b1a539","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/MiniClusterClient.java","lineNumber":154,"sourceCode":"    }\n\n    @Override\n    public CompletableFuture<Collection<JobStatusMessage>> listJobs() {\n        return miniCluster.listJobs();\n    }\n\n    @Override\n    public CompletableFuture<Map<String, Object>> getAccumulators(JobID jobID, ClassLoader loader) {\n        return miniCluster\n                .getExecutionGraph(jobID)\n                .thenApply(AccessExecutionGraph::getAccumulatorsSerialized)\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    @Override\n    public CompletableFuture<JobStatus> getJobStatus(JobID jobId) {\n        return miniCluster.getJobStatus(jobId);\n    }\n\n    @Override\n    public void close() {}\n\n    @Override\n    public MiniClusterClient.MiniClusterId getClusterId() {\n        return MiniClusterId.INSTANCE;\n    }\n","sourceCodeStart":136,"sourceCodeEnd":172,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java#L136-L172","documentation":"Thrown by MiniClusterClient.getAccumulators when AccumulatorHelper.deserializeAndUnwrapAccumulators fails while converting the serialized accumulator map returned by the execution graph into a plain Map<String,Object>. The Future is completed exceptionally with a CompletionException wrapping the original cause. This is the client-side (test/embedded) path for reading job accumulators from a local MiniCluster.","triggerScenarios":"Calling MiniClusterClient.getAccumulators(jobID, loader) when the accumulated values cannot be deserialized with the supplied ClassLoader. Common roots: the ClassLoader does not contain the accumulator value's class, the accumulator type is not Serializable, or a serializer mismatch exists between the accumulator produced by the job and the class available on the client.","commonSituations":"Running integration tests that read accumulators from a MiniCluster where the user job JAR (or its accumulator type classes) are not on the test ClassLoader. Mismatched Flink versions between client and cluster. Custom accumulator types whose classes are scoped to the user jar and not visible to the test classpath.","solutions":["Ensure the class of each accumulator value is on the ClassLoader passed to getAccumulators (typically the same ClassLoader used to load the user job JAR).","Verify the accumulator value type implements java.io.Serializable and that all nested fields are also serializable.","Check the underlying cause via CompletionException.getCause() / FlinkExceptions.findSerializedThrowable to pinpoint which key/value failed, then fix the serializer or classpath for that type.","If using a custom accumulator, register a matching TypeSerializer/TypeSerializerSnapshot and confirm the serialized format is version-compatible."],"exampleFix":"// before: client ClassLoader is missing the user accumulator type\nFuture<Map<String,Object>> acc = client.getAccumulators(jobId, getClass().getClassLoader());\n\n// after: use the same ClassLoader that loaded the job JAR\nFuture<Map<String,Object>> acc = client.getAccumulators(jobId, userCodeClassLoader);","handlingStrategy":"try-catch","validationCode":"// Before calling getAccumulators, confirm the loader can see the accumulator types\nClass<?> accValueClass = Class.forName(\"com.example.MyAccumulatorValue\", false, loader);\nif (!Serializable.class.isAssignableFrom(accValueClass)) {\n    throw new IllegalStateException(\"Accumulator value type is not Serializable: \" + accValueClass);\n}","typeGuard":null,"tryCatchPattern":"try {\n    Map<String,Object> acc = client.getAccumulators(jobId, userCodeClassLoader).get(timeout, TimeUnit.SECONDS);\n} catch (ExecutionException ee) {\n    Throwable cause = (ee.getCause() instanceof CompletionException)\n        ? ee.getCause().getCause() : ee.getCause();\n    log.error(\"Failed to read accumulators for job {}\", jobId, cause);\n}","preventionTips":["Always pass the user-code ClassLoader (the one that loaded the job jar) to getAccumulators, never the system loader.","Make all custom accumulator value types implement Serializable with explicit serialVersionUID.","Keep client and cluster Flink versions aligned to avoid serializer format drift."],"tags":["accumulator","deserialization","classpath","minicluster"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}