apache/flink · error · IllegalArgumentException

The merged accumulator must be AverageAccumulator.

Error message

The merged accumulator must be AverageAccumulator.

What it means

Thrown by AverageAccumulator.merge when the argument is not an AverageAccumulator. Flink calls merge() across subtask accumulators of the same name; if the other accumulator is a different type (e.g., a DoubleCounter or a custom accumulator), it cannot contribute the count+sum pair that AverageAccumulator needs to compute the mean, so it rejects the merge.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/accumulators/AverageAccumulator.java:78

            return 0.0;
        }
        return this.sum / this.count;
    }

    @Override
    public void resetLocal() {
        this.count = 0;
        this.sum = 0;
    }

    @Override
    public void merge(Accumulator<Double, Double> other) {
        if (other instanceof AverageAccumulator) {
            AverageAccumulator avg = (AverageAccumulator) other;
            this.count += avg.count;
            this.sum += avg.sum;
        } else {
            throw new IllegalArgumentException(
                    "The merged accumulator must be AverageAccumulator.");
        }
    }

    @Override
    public AverageAccumulator clone() {
        AverageAccumulator average = new AverageAccumulator();
        average.count = this.count;
        average.sum = this.sum;
        return average;
    }

    @Override
    public String toString() {
        return "AverageAccumulator " + this.getLocalValue() + " for " + this.count + " elements";
    }
}

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Ensure every accumulator that should merge with this AverageAccumulator is also an AverageAccumulator registered under the same name.
2. Avoid manual merge() calls; rely on the runtime to merge identically-named, identically-typed accumulators.
3. Rename colliding accumulators so each name maps to exactly one accumulator type.

Example fix

// before
getRuntimeContext().getAverageAccumulator("stats"); // op A
getRuntimeContext().getDoubleCounter("stats");      // op B -> merge fails
// after
getRuntimeContext().getAverageAccumulator("stats.avg");
getRuntimeContext().getDoubleCounter("stats.sum");

// ensure all merged accumulators under one name are AverageAccumulator
getRuntimeContext().getAverageAccumulator("stats.avg");

public static boolean isAverage(Accumulator<?,?> a) {
    return a instanceof AverageAccumulator;
}

When it happens

Trigger: An accumulator named the same as an AverageAccumulator but backed by a different accumulator class; calling avgAccumulator.merge(someOtherAccumulator) directly in user code.

Common situations: Name collision between an AverageAccumulator and another built-in/custom accumulator in the same job; migrating from a DoubleCounter to an AverageAccumulator without renaming.

Related errors

• The accumulator object '{name}' was created with two differe
• Could not stop with a savepoint job "{}".
• Could not stop with a detached savepoint job "{}".
• Failed to trigger a savepoint for the job {}.
• JAR file is not a file: {}

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