apache/flink · error · CompletionException

Cannot deserialize and unwrap accumulators properly.

Error message

Cannot deserialize and unwrap accumulators properly.

What it means

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.

Source

Thrown at flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java:154

    }

    @Override
    public CompletableFuture<Collection<JobStatusMessage>> listJobs() {
        return miniCluster.listJobs();
    }

    @Override
    public CompletableFuture<Map<String, Object>> getAccumulators(JobID jobID, ClassLoader loader) {
        return miniCluster
                .getExecutionGraph(jobID)
                .thenApply(AccessExecutionGraph::getAccumulatorsSerialized)
                .thenApply(
                        accumulators -> {
                            try {
                                return AccumulatorHelper.deserializeAndUnwrapAccumulators(
                                        accumulators, loader);
                            } catch (Exception e) {
                                throw new CompletionException(
                                        "Cannot deserialize and unwrap accumulators properly.", e);
                            }
                        });
    }

    @Override
    public CompletableFuture<JobStatus> getJobStatus(JobID jobId) {
        return miniCluster.getJobStatus(jobId);
    }

    @Override
    public void close() {}

    @Override
    public MiniClusterClient.MiniClusterId getClusterId() {
        return MiniClusterId.INSTANCE;
    }

View on GitHub (pinned to 2f3c205e92)

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.

Example fix

// before: client ClassLoader is missing the user accumulator type
Future<Map<String,Object>> acc = client.getAccumulators(jobId, getClass().getClassLoader());

// after: use the same ClassLoader that loaded the job JAR
Future<Map<String,Object>> acc = client.getAccumulators(jobId, userCodeClassLoader);

Defensive patterns

Strategy: try-catch

Validate before calling

// Before calling getAccumulators, confirm the loader can see the accumulator types
Class<?> accValueClass = Class.forName("com.example.MyAccumulatorValue", false, loader);
if (!Serializable.class.isAssignableFrom(accValueClass)) {
    throw new IllegalStateException("Accumulator value type is not Serializable: " + accValueClass);
}

Try / catch

try {
    Map<String,Object> acc = client.getAccumulators(jobId, userCodeClassLoader).get(timeout, TimeUnit.SECONDS);
} catch (ExecutionException ee) {
    Throwable cause = (ee.getCause() instanceof CompletionException)
        ? ee.getCause().getCause() : ee.getCause();
    log.error("Failed to read accumulators for job {}", jobId, cause);
}

Prevention

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.

When it happens

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

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

Related errors

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