{"record":{"id":"c1e21866820d9cdd","repo":"apache/flink","slug":"cannot-deserialize-and-unwrap-accumulators-properl","errorCode":null,"errorMessage":"Cannot deserialize and unwrap accumulators properly.","messagePattern":"Cannot deserialize and unwrap accumulators properly\\.","errorType":"exception","errorClass":"CompletionException","httpStatus":null,"severity":"warning","filePath":"flink-clients/src/main/java/org/apache/flink/client/deployment/application/EmbeddedJobClient.java","lineNumber":128,"sourceCode":"            @Nullable final String savepointDirectory, SavepointFormatType formatType) {\n        return dispatcherGateway.triggerSavepointAndGetLocation(\n                jobId, savepointDirectory, formatType, TriggerSavepointMode.SAVEPOINT, timeout);\n    }\n\n    @Override\n    public CompletableFuture<Map<String, Object>> getAccumulators() {\n        checkNotNull(classLoader);\n\n        return dispatcherGateway\n                .requestJob(jobId, timeout)\n                .thenApply(ArchivedExecutionGraph::getAccumulatorsSerialized)\n                .thenApply(\n                        accumulators -> {\n                            try {\n                                return AccumulatorHelper.deserializeAndUnwrapAccumulators(\n                                        accumulators, classLoader);\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<JobExecutionResult> getJobExecutionResult() {\n        checkNotNull(classLoader);\n\n        final Duration retryPeriod = Duration.ofMillis(100L);\n        return JobStatusPollingUtils.getJobResult(\n                        dispatcherGateway, jobId, retryExecutor, timeout, retryPeriod)\n                .thenApply(\n                        (jobResult) -> {\n                            try {\n                                return jobResult.toJobExecutionResult(classLoader);\n                            } catch (Throwable t) {\n                                throw new CompletionException(","sourceCodeStart":110,"sourceCodeEnd":146,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-clients/src/main/java/org/apache/flink/client/deployment/application/EmbeddedJobClient.java#L110-L146","documentation":"Thrown by EmbeddedJobClient.getAccumulators when AccumulatorHelper.deserializeAndUnwrapAccumulators fails while converting serialized accumulator results back into Java objects. The accumulators are fetched from the ArchivedExecutionGraph via the dispatcher gateway, then deserialized using the provided classLoader. If a custom accumulator class is not on the classloader, or the serialized format is incompatible (version mismatch), the deserialization throws and is wrapped in a CompletionException. This occurs in embedded/application mode where the JobClient talks directly to the dispatcher.","triggerScenarios":"Using custom accumulator classes that aren't on the EmbeddedJobClient's classloader; upgrading Flink versions where the accumulator serialization format changed; a ClassNotFound or ClassCastException during deserialization.","commonSituations":"Application mode deployment where the user-code classloader differs from the system classloader; custom Accumulator implementations that aren't bundled in the job JAR; version mismatch between the submitting client and the running cluster.","solutions":["Ensure custom accumulator classes are on the classloader passed to EmbeddedJobClient","Verify the Flink version of the submitting client matches the cluster version exactly","Bundle all custom accumulator implementations in the job JAR","If accumulators are not critical, catch the CompletionException and proceed without them"],"exampleFix":"// before\nMap<String, Object> accumulators = jobClient.getAccumulators().get();\n\n// after\nMap<String, Object> accumulators;\ntry {\n    accumulators = jobClient.getAccumulators().get();\n} catch (ExecutionException e) {\n    LOG.warn(\"Failed to deserialize accumulators, continuing without them\", e);\n    accumulators = Collections.emptyMap();\n}","handlingStrategy":"fallback","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    Map<String, Object> accumulators = jobClient.getAccumulators().get();\n} catch (ExecutionException e) {\n    LOG.warn(\"Failed to deserialize accumulators, continuing without them: {}\", e.getMessage());\n    accumulators = Collections.emptyMap();\n}","preventionTips":["Ensure custom accumulator classes are on the classloader passed to EmbeddedJobClient","Match the Flink client and cluster versions exactly to avoid serialization format mismatches","Bundle custom accumulator implementations in the job JAR","Treat accumulators as non-critical metadata and handle deserialization failures gracefully"],"tags":["accumulators","embedded","application-mode","serialization","classloader"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}