apache/flink · error · RuntimeException

An aggregator is already registered under the given name.

Error message

An aggregator is already registered under the given name.

What it means

Thrown by AggregatorRegistry.registerAggregator when an aggregator is already registered under the given name. The registry maps each name to exactly one aggregator instance to keep per-superstep aggregation unambiguous; a second registration under the same name would overwrite the first. The API rejects duplicates with a RuntimeException.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/aggregators/AggregatorRegistry.java:46

/** A registry for iteration {@link Aggregator}s. */
@Internal
public class AggregatorRegistry {

    private final Map<String, Aggregator<?>> registry = new HashMap<String, Aggregator<?>>();

    private ConvergenceCriterion<? extends Value> convergenceCriterion;

    private String convergenceCriterionAggregatorName;

    // --------------------------------------------------------------------------------------------

    public void registerAggregator(String name, Aggregator<?> aggregator) {
        if (name == null || aggregator == null) {
            throw new IllegalArgumentException("Name and aggregator must not be null");
        }
        if (this.registry.containsKey(name)) {
            throw new RuntimeException("An aggregator is already registered under the given name.");
        }
        this.registry.put(name, aggregator);
    }

    public Collection<AggregatorWithName<?>> getAllRegisteredAggregators() {
        ArrayList<AggregatorWithName<?>> list =
                new ArrayList<AggregatorWithName<?>>(this.registry.size());

        for (Map.Entry<String, Aggregator<?>> entry : this.registry.entrySet()) {
            @SuppressWarnings("unchecked")
            Aggregator<Value> valAgg = (Aggregator<Value>) entry.getValue();
            list.add(new AggregatorWithName<>(entry.getKey(), valAgg));
        }
        return list;
    }

    public <T extends Value> void registerAggregationConvergenceCriterion(
            String name, Aggregator<T> aggregator, ConvergenceCriterion<T> convergenceCheck) {

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Check the registry (or wrap registration in a helper) before re-registering; use distinct names per aggregator.
2. Centralize aggregator registration so each name is registered exactly once.
3. If re-registration is intentional for the same instance, guard: only register when not already present.

Example fix

// before
registry.registerAggregator("sum", agg1);
registry.registerAggregator("sum", agg2);
// after
// collect names once
Set<String> registered = new HashSet<>();
void registerOnce(String name, Aggregator<?> agg) {
    if (registered.add(name)) registry.registerAggregator(name, agg);
}

// track registered names
if (!registeredNames.contains(name)) {
    registry.registerAggregator(name, aggregator);
    registeredNames.add(name);
}

When it happens

Trigger: Calling registerAggregator(name, agg) twice with the same name but different aggregator instances; two operators/UDFs in an iterative job both registering an aggregator under 'sum'.

Common situations: Shared iteration setup code called multiple times; library code and user code both registering a well-known aggregator name; refactor that merged two iteration builders.

Related errors

• Name and aggregator must not be null
• cache file {name}already exists!
• Name, aggregator, or convergence criterion must not be null
• A TypeInfoFactory for type '{}' is already registered.
• Could not fetch next KeyValue pair.

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