apache/flink · error · CompletionException

Cannot deserialize and unwrap accumulators properly.

Error message

Cannot deserialize and unwrap accumulators properly.

What it means

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.

Source

Thrown at flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java:718

        final JobAccumulatorsHeaders accumulatorsHeaders = JobAccumulatorsHeaders.getInstance();
        final JobAccumulatorsMessageParameters accMsgParams =
                accumulatorsHeaders.getUnresolvedMessageParameters();
        accMsgParams.jobPathParameter.resolve(jobID);
        accMsgParams.includeSerializedAccumulatorsParameter.resolve(
                Collections.singletonList(true));

        CompletableFuture<JobAccumulatorsInfo> responseFuture =
                sendRequest(accumulatorsHeaders, accMsgParams);

        return responseFuture
                .thenApply(JobAccumulatorsInfo::getSerializedUserAccumulators)
                .thenApply(
                        accumulators -> {
                            try {
                                return AccumulatorHelper.deserializeAndUnwrapAccumulators(
                                        accumulators, loader);
                            } catch (Exception e) {
                                throw new CompletionException(
                                        "Cannot deserialize and unwrap accumulators properly.", e);
                            }
                        });
    }

    private CompletableFuture<SavepointInfo> pollSavepointAsync(
            final JobID jobId, final TriggerId triggerID) {
        return pollResourceAsync(
                () -> {
                    final SavepointStatusHeaders savepointStatusHeaders =
                            SavepointStatusHeaders.getInstance();
                    final SavepointStatusMessageParameters savepointStatusMessageParameters =
                            savepointStatusHeaders.getUnresolvedMessageParameters();
                    savepointStatusMessageParameters.jobIdPathParameter.resolve(jobId);
                    savepointStatusMessageParameters.triggerIdPathParameter.resolve(triggerID);
                    return sendRequest(savepointStatusHeaders, savepointStatusMessageParameters);
                });
    }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Pass a ClassLoader that includes the user JAR containing the accumulator classes to getAccumulators.
2. Ensure the client and cluster run the same Flink version and user JAR versions.
3. If using custom accumulators, verify their serialization is forward/backward compatible.
4. Check for ClassNotFoundException in the exception cause to identify the missing class.

Example fix

// before — default classloader missing user accumulator classes
client.getAccumulators(jobId, getClass().getClassLoader());
// after — use a classloader that includes the user JAR
URLClassLoader userLoader = new URLClassLoader(userJarUrls, getClass().getClassLoader());
client.getAccumulators(jobId, userLoader);

// Validate classloader has accumulator classes before calling getAccumulators
ClassLoader loader = getUserJarClassLoader();
try {
    Class.forName("com.example.MyAccumulator", false, loader);
} catch (ClassNotFoundException e) {
    throw new IllegalStateException("Accumulator class not on the provided classloader", e);
}

try {
    Map<String, Object> accs = client.getAccumulators(jobId, loader).get();
} catch (ExecutionException e) {
    Throwable cause = ExceptionUtils.stripExecutionException(e);
    if (cause.getMessage().contains("Cannot deserialize and unwrap accumulators")) {
        // likely ClassNotFoundException — check classloader
    }
}

When it happens

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

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

Related errors

• Cannot deserialize and unwrap accumulators properly.
• Failed to deserialize coordination response
• Failed to serialize ExecutionPlan.
• Unable to instantiate Hadoop InputSplit
• Unable to instantiate the hadoop input format

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