apache/flink · warning · CompletionException

Cannot deserialize and unwrap accumulators properly.

Error message

Cannot deserialize and unwrap accumulators properly.

What it means

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.

Source

Thrown at flink-clients/src/main/java/org/apache/flink/client/deployment/application/EmbeddedJobClient.java:128

            @Nullable final String savepointDirectory, SavepointFormatType formatType) {
        return dispatcherGateway.triggerSavepointAndGetLocation(
                jobId, savepointDirectory, formatType, TriggerSavepointMode.SAVEPOINT, timeout);
    }

    @Override
    public CompletableFuture<Map<String, Object>> getAccumulators() {
        checkNotNull(classLoader);

        return dispatcherGateway
                .requestJob(jobId, timeout)
                .thenApply(ArchivedExecutionGraph::getAccumulatorsSerialized)
                .thenApply(
                        accumulators -> {
                            try {
                                return AccumulatorHelper.deserializeAndUnwrapAccumulators(
                                        accumulators, classLoader);
                            } catch (Exception e) {
                                throw new CompletionException(
                                        "Cannot deserialize and unwrap accumulators properly.", e);
                            }
                        });
    }

    @Override
    public CompletableFuture<JobExecutionResult> getJobExecutionResult() {
        checkNotNull(classLoader);

        final Duration retryPeriod = Duration.ofMillis(100L);
        return JobStatusPollingUtils.getJobResult(
                        dispatcherGateway, jobId, retryExecutor, timeout, retryPeriod)
                .thenApply(
                        (jobResult) -> {
                            try {
                                return jobResult.toJobExecutionResult(classLoader);
                            } catch (Throwable t) {
                                throw new CompletionException(

View on GitHub (pinned to 2f3c205e92)

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

Example fix

// before
Map<String, Object> accumulators = jobClient.getAccumulators().get();

// after
Map<String, Object> accumulators;
try {
    accumulators = jobClient.getAccumulators().get();
} catch (ExecutionException e) {
    LOG.warn("Failed to deserialize accumulators, continuing without them", e);
    accumulators = Collections.emptyMap();
}

Defensive patterns

Strategy: fallback

Try / catch

try {
    Map<String, Object> accumulators = jobClient.getAccumulators().get();
} catch (ExecutionException e) {
    LOG.warn("Failed to deserialize accumulators, continuing without them: {}", e.getMessage());
    accumulators = Collections.emptyMap();
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/c1e21866820d9cdd. Report an issue: GitHub.