nathanmarz/storm · error · RuntimeException

Aggregate operation can only have one parent

Error message

Aggregate operation can only have one parent

What it means

AggregateProcessor.prepare requires the aggregate processor to have exactly one upstream parent tuple factory in the Trident processing graph. If tridentContext reports a parent count other than 1, prepare throws this RuntimeException. Aggregation in Trident consumes one merged input stream only, so a topology that feeds multiple (or zero) streams into an aggregate is invalid.

Solutions

1. Insert an operation that reduces the stream to one parent before aggregating, or restructure so aggregate is applied to a single Stream object.
2. If you need multi-stream aggregation, first merge/join streams and then aggregate on the merged Stream (Trident merges are single-parent chains themselves — apply aggregate after the merged result).
3. Check whether you actually want multiReduce/partitionAggregate on a properly joined Stream.
4. Review custom graph construction ensuring each aggregate node has in-degree 1.

Example fix

// before: aggregating a merged multi-parent stream
TridentStream merged = s1.merge(s2);
merged.aggregate(...) // invalid at graph level here
// after: aggregate each stream first, then merge results
TridentStream a = s1.aggregate(...) ;
TridentStream b = s2.aggregate(...);
TridentStream out = a.merge(b);

// Java: before aggregate, confirm the stream chain has a single parent
// (structural check at graph build time in custom planners)
if (node.getParents().size() != 1) {
    throw new IllegalArgumentException("aggregate requires exactly one parent stream, got " + node.getParents().size());
}

try { stream.aggregate(...); } catch (RuntimeException e) { if (e.getMessage().contains("can only have one parent")) { throw new TopologyStructureException("Move aggregate() onto a single-parent stream chain", e); } throw e; }

When it happens

Trigger: Calling Stream.aggregate (or persistentAggregate) at a point in the topology where the node has more than one parent stream — e.g. aggregating after a merge/join/shuffle connection of two Streams without an intervening single-parent operation.

Common situations: Aggregating directly on the output of a join or union in Trident; programmatically built graphs connecting two nodes into one aggregate node; misplacing aggregate after multi-stream bottlenecks.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• Each operation can only have one parent
• Projection processor can only have one parent
• Partition persist operation can only have one parent
• State query operation can only have one parent
• Trying to select non-existent field

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

Appendix: source

Thrown at storm-core/src/jvm/storm/trident/planner/processor/AggregateProcessor.java:49

public class AggregateProcessor implements TridentProcessor {
    Aggregator _agg;
    TridentContext _context;
    FreshCollector _collector;
    Fields _inputFields;
    ProjectionFactory _projection;

    public AggregateProcessor(Fields inputFields, Aggregator agg) {
        _agg = agg;
        _inputFields = inputFields;
    }
    
    @Override
    public void prepare(Map conf, TopologyContext context, TridentContext tridentContext) {
        List<Factory> parents = tridentContext.getParentTupleFactories();
        if(parents.size()!=1) {
            throw new RuntimeException("Aggregate operation can only have one parent");
        }
        _context = tridentContext;
        _collector = new FreshCollector(tridentContext);
        _projection = new ProjectionFactory(parents.get(0), _inputFields);
        _agg.prepare(conf, new TridentOperationContext(context, _projection));
    }

    @Override
    public void cleanup() {
        _agg.cleanup();
    }

    @Override
    public void startBatch(ProcessorContext processorContext) {
        _collector.setContext(processorContext);
        processorContext.state[_context.getStateIndex()] = _agg.init(processorContext.batchId, _collector);
    }    

View on GitHub (pinned to cdb116e942)