{"record":{"id":"f8e49aefb110cf93","repo":"nathanmarz/storm","slug":"combiner-state-updater-should-receive-a-single-tuple","errorCode":null,"errorMessage":"Combiner state updater should receive a single tuple. Received: ${tuples}","messagePattern":"Combiner state updater should receive a single tuple\\. Received: (.+?)","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"storm-core/src/jvm/storm/trident/operation/impl/CombinerAggStateUpdater.java","lineNumber":42,"sourceCode":"import storm.trident.operation.TridentCollector;\nimport storm.trident.operation.TridentOperationContext;\nimport storm.trident.state.CombinerValueUpdater;\nimport storm.trident.state.StateUpdater;\nimport storm.trident.state.snapshot.Snapshottable;\nimport storm.trident.tuple.TridentTuple;\n\npublic class CombinerAggStateUpdater implements StateUpdater<Snapshottable> {\n    CombinerAggregator _agg;\n    \n    public CombinerAggStateUpdater(CombinerAggregator agg) {\n        _agg = agg;\n    }\n    \n\n    @Override\n    public void updateState(Snapshottable state, List<TridentTuple> tuples, TridentCollector collector) {\n        if(tuples.size()!=1) {\n            throw new IllegalArgumentException(\"Combiner state updater should receive a single tuple. Received: \" + tuples.toString());\n        }\n        Object newVal = state.update(new CombinerValueUpdater(_agg, tuples.get(0).getValue(0)));\n        collector.emit(new Values(newVal));\n    }\n\n    @Override\n    public void prepare(Map conf, TridentOperationContext context) {        \n    }\n\n    @Override\n    public void cleanup() {\n    }\n    \n}\n","sourceCodeStart":24,"sourceCodeEnd":57,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/storm/trident/operation/impl/CombinerAggStateUpdater.java#L24-L57","documentation":"CombinerAggStateUpdater performs state.update(new CombinerValueUpdater(agg, value)) which needs exactly one new value per call, so updateState asserts tuples.size()==1 and throws IllegalArgumentException otherwise. Combiner aggregators collapse a value, unlike reducers that accept a batch of tuples.","triggerScenarios":"A persistentAggregate(newMapState, ..., new CombinerAggregator..., new MapCombinerAggStateUpdater...) whose upstream grouping/projection delivers 0 or >1 tuples to a single updateState invocation.","commonSituations":"Mis-wiring a combiner updater into a reducer-style aggregate chain; upstream partitioning collapsing/expanding tuples so multiple values arrive at once; hand-rolled state update calls passing a batch list.","solutions":["Use ReducerAggStateUpdater / ReducerAggregator semantics if you must process multiple tuples per update, or pre-combine upstream","Ensure the stream feeding the combiner updater is grouped/partitioned so exactly one tuple reaches updateState","Switch the operation to a CombinerAggregator-based persistentAggregate which guarantees single-value updates"],"exampleFix":"// before\nstream.groupBy(g).persistentAggregate(state, new Fields(\"v\"), new ReducerAggregator(), new MapCombinerAggStateUpdater(...));\n// after\nstream.groupBy(g).persistentAggregate(state, new Fields(\"v\"), new Count(), new MapCombinerAggStateUpdater<>(new Count()));","handlingStrategy":"validation","validationCode":"if (tuples == null || tuples.size() != 1) {\n    throw new IllegalArgumentException(\"CombinerAggStateUpdater requires exactly one tuple, got \" + (tuples == null ? 0 : tuples.size()));\n}","typeGuard":"boolean isSingleTuple(List<TridentTuple> tuples) {\n    return tuples != null && tuples.size() == 1;\n}","tryCatchPattern":"try {\n    updater.updateState(state, tuples, collector);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"should receive a single tuple\")) {\n        throw new IllegalStateException(\"Use a Reducer-style updater for multi-tuple batches\", e);\n    } throw e;\n}","preventionTips":["Only pair CombinerAggregator with combiner state updaters","Use ReducerAggStateUpdater for batch-style aggregation","Test the persistentAggregate pipeline with real batch sizes"],"tags":["storm","trident","state","aggregator"],"backgroundTag":"invalid-argument-value","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"}