{"record":{"id":"5ca11f5ff1a6b144","repo":"nathanmarz/storm","slug":"aggregate-operation-can-only-have-one-parent","errorCode":null,"errorMessage":"Aggregate operation can only have one parent","messagePattern":"Aggregate operation can only have one parent","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"storm-core/src/jvm/storm/trident/planner/processor/AggregateProcessor.java","lineNumber":49,"sourceCode":"\n\npublic class AggregateProcessor implements TridentProcessor {\n    Aggregator _agg;\n    TridentContext _context;\n    FreshCollector _collector;\n    Fields _inputFields;\n    ProjectionFactory _projection;\n\n    public AggregateProcessor(Fields inputFields, Aggregator agg) {\n        _agg = agg;\n        _inputFields = inputFields;\n    }\n    \n    @Override\n    public void prepare(Map conf, TopologyContext context, TridentContext tridentContext) {\n        List<Factory> parents = tridentContext.getParentTupleFactories();\n        if(parents.size()!=1) {\n            throw new RuntimeException(\"Aggregate operation can only have one parent\");\n        }\n        _context = tridentContext;\n        _collector = new FreshCollector(tridentContext);\n        _projection = new ProjectionFactory(parents.get(0), _inputFields);\n        _agg.prepare(conf, new TridentOperationContext(context, _projection));\n    }\n\n    @Override\n    public void cleanup() {\n        _agg.cleanup();\n    }\n\n    @Override\n    public void startBatch(ProcessorContext processorContext) {\n        _collector.setContext(processorContext);\n        processorContext.state[_context.getStateIndex()] = _agg.init(processorContext.batchId, _collector);\n    }    \n","sourceCodeStart":31,"sourceCodeEnd":67,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/storm/trident/planner/processor/AggregateProcessor.java#L31-L67","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Insert an operation that reduces the stream to one parent before aggregating, or restructure so aggregate is applied to a single Stream object.","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).","Check whether you actually want multiReduce/partitionAggregate on a properly joined Stream.","Review custom graph construction ensuring each aggregate node has in-degree 1."],"exampleFix":"// before: aggregating a merged multi-parent stream\nTridentStream merged = s1.merge(s2);\nmerged.aggregate(...) // invalid at graph level here\n// after: aggregate each stream first, then merge results\nTridentStream a = s1.aggregate(...) ;\nTridentStream b = s2.aggregate(...);\nTridentStream out = a.merge(b);","handlingStrategy":"validation","validationCode":"// Java: before aggregate, confirm the stream chain has a single parent\n// (structural check at graph build time in custom planners)\nif (node.getParents().size() != 1) {\n    throw new IllegalArgumentException(\"aggregate requires exactly one parent stream, got \" + node.getParents().size());\n}","typeGuard":null,"tryCatchPattern":"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; }","preventionTips":["Call aggregate() only on Stream objects produced by a linear each/spout chain or a single join output","Never apply aggregate directly to multi-parent merged nodes","Draw the DAG before adding aggregations to joins/unions"],"tags":["trident","topology-graph","aggregate","invalid-wiring"],"backgroundTag":"incompatible-source-type","analyzedSha":"cdb116e942666973bc4eaa0df098d5bab82739e7","analyzedAt":"2026-09-12T14:30:00.714Z","contentChangedAt":"2026-09-12T14:30:00.714Z","schemaVersion":2},"datasetVersion":"2026-09-16T09:17:16.951Z"}