apache/flink · error · UnsupportedOperationException

The accumulator '${name}' already exists and cannot be added

Error message

The accumulator '${name}' already exists and cannot be added.

What it means

Thrown by AbstractRuntimeUDFContext.addAccumulator() when an accumulator with the given name already exists in the accumulators map. Accumulator names must be unique within a single task's RuntimeContext; attempting to register a second accumulator under an already-used name is rejected.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java:137

    public LongCounter getLongCounter(String name) {
        return (LongCounter) getAccumulator(name, LongCounter.class);
    }

    @Override
    public Histogram getHistogram(String name) {
        return (Histogram) getAccumulator(name, Histogram.class);
    }

    @Override
    public DoubleCounter getDoubleCounter(String name) {
        return (DoubleCounter) getAccumulator(name, DoubleCounter.class);
    }

    @Override
    public <V, A extends Serializable> void addAccumulator(
            String name, Accumulator<V, A> accumulator) {
        if (accumulators.containsKey(name)) {
            throw new UnsupportedOperationException(
                    "The accumulator '" + name + "' already exists and cannot be added.");
        }
        accumulators.put(name, accumulator);
    }

    @SuppressWarnings("unchecked")
    @Override
    public <V, A extends Serializable> Accumulator<V, A> getAccumulator(String name) {
        return (Accumulator<V, A>) accumulators.get(name);
    }

    @Override
    public ClassLoader getUserCodeClassLoader() {
        return this.userCodeClassLoader.asClassLoader();
    }

    @Override
    public void registerUserCodeClassLoaderReleaseHookIfAbsent(

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Use unique accumulator names, e.g. prefix with a task-specific or function-instance-specific identifier.
2. Before adding, check getRuntimeContext().getAccumulator(name) != null and reuse or skip instead of re-adding.
3. Register accumulators in open() (called once) rather than in per-record methods like map()/flatMap() where they might be called repeatedly.

Example fix

// before — registers on every record, throws on second
public String map(String value) {
    getRuntimeContext().addAccumulator("count", new LongCounter());
    ...
}

// after — register once in open()
private LongCounter counter;

@Override
public void open(OpenContext ctx) {
    counter = new LongCounter();
    getRuntimeContext().addAccumulator("count", counter);
}

public String map(String value) {
    counter.add(1);
    ...
}

// Check before adding
RuntimeContext ctx = getRuntimeContext();
if (ctx.getAccumulator("myCounter") == null) {
    ctx.addAccumulator("myCounter", new LongCounter());
}
// Best practice: do this once in open()

When it happens

Trigger: A RichFunction calls getRuntimeContext().addAccumulator(name, accumulator) more than once with the same name string, or two functions in the same operator chain share the same accumulator name and both register it.

Common situations: A function is reused across multiple operators in a chain and registers accumulators with hardcoded names. Multiple instances of the same RichFunction class in one task slot. Accumulator name collisions caused by copying function code without changing the name string.

Related errors

• Cannot create accumulator ${accumulatorClass.getName()}
• The runtime context has not been initialized.
• This state is only accessible by functions executed on a Key
• The broadcast variable with name '${name}' is not a List. A
• The broadcast variable with name '${name}' has not been set.

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