nathanmarz/storm · error · IllegalArgumentException

Output fields for chained aggregators must be distinct

Error message

Output fields for chained aggregators must be distinct: ${allOutFields}

What it means

ChainedAggregatorDeclarer.chainEnd() concatenates the output fields of all chained aggregators into one Fields. Duplicate field names across the chained aggregators would make the combined output tuple ambiguous, so an IllegalArgumentException is thrown when the concatenated list contains duplicates.

Solutions

Rename each chained aggregator's output fields so they are unique
Use project() after chainEnd to select/rename fields if downstream needs specific names
Restructure into separate chainedAgg calls if fields must repeat

Example fix

// before
.chainAgg(partitionAggregate(..., new Fields("count")))
.chainAgg(partitionAggregate(..., new Fields("count")))
.chainEnd();
// after
.chainAgg(partitionAggregate(..., new Fields("count")))
.chainAgg(partitionAggregate(..., new Fields("total")))
.chainEnd();

Defensive patterns

Strategy: validation

Validate before calling

List<String> outs = new ArrayList<>();
aggs.forEach(a -> outs.addAll(a.outputFields().toList()));
if (new HashSet<>(outs).size() != outs.size()) throw new IllegalArgumentException("Chained agg output fields must be unique");

Type guard

boolean allDistinct(List<String> fields) {
    return new HashSet<>(fields).size() == fields.size();
}

Try / catch

try {
    declarer.chainEnd();
} catch (IllegalArgumentException e) {
    if (e.getMessage().contains("must be distinct")) {
        throw new IllegalStateException("Rename duplicate chained aggregator output fields", e);
    } throw e;
}

Prevention

Prefix each chained aggregator's output fields uniquely (e.g. cnt_, sum_)
Review chained aggregator specs in code review for repeated output names
Add a build-the-topology unit test to fail fast

When it happens

Trigger: chainAgg(...) with multiple aggregators where two of them declare the same output field name (e.g. two Counts emitted as Fields("count")), then calling chainEnd().

Common situations: Chaining aggregate(new Count(), new Fields("count")) with another agg that also emits "count"; copy-pasted aggregator specs; combining sum and count both defaulting to output field "sum"/"count" names.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

AI-assisted analysis of nathanmarz/storm@cdb116e942 (2026-09-12). Data as JSON: /api/errors/f1adf12852f568e4. Report an issue: GitHub.

Appendix: source

Thrown at storm-core/src/jvm/storm/trident/fluent/ChainedAggregatorDeclarer.java:92

        Aggregator[] aggs = new Aggregator[_aggs.size()];
        int[] outSizes = new int[_aggs.size()];
        List<String> allOutFields = new ArrayList<String>();
        Set<String> allInFields = new HashSet<String>();
        for(int i=0; i<_aggs.size(); i++) {
            AggSpec spec = _aggs.get(i);
            Fields infields = spec.inFields;
            if(infields==null) infields = new Fields();
            Fields outfields = spec.outFields;
            if(outfields==null) outfields = new Fields();

            inputFields[i] = infields;
            aggs[i] = spec.agg;
            outSizes[i] = outfields.size();  
            allOutFields.addAll(outfields.toList());
            allInFields.addAll(infields.toList());
        }
        if(new HashSet(allOutFields).size() != allOutFields.size()) {
            throw new IllegalArgumentException("Output fields for chained aggregators must be distinct: " + allOutFields.toString());
        }
        
        Fields inFields = new Fields(new ArrayList<String>(allInFields));
        Fields outFields = new Fields(allOutFields);
        Aggregator combined = new ChainedAggregatorImpl(aggs, inputFields, new ComboList.Factory(outSizes));
        
        if(_type!=AggType.FULL) {
            _stream = _stream.partitionAggregate(inFields, combined, outFields);
        }
        if(_type!=AggType.PARTITION) {
            _stream = _globalScheme.aggPartition(_stream);
            BatchToPartition singleEmit = _globalScheme.singleEmitPartitioner();
            Aggregator toAgg = combined;
            if(singleEmit!=null) {
                toAgg = new SingleEmitAggregator(combined, singleEmit);
            }
            // this assumes that inFields and outFields are the same for combineragg
            // assumption also made above

View on GitHub (pinned to cdb116e942)