{"record":{"id":"c56408e9925012d6","repo":"nathanmarz/storm","slug":"received-unexpected-tuple-tuple","errorCode":null,"errorMessage":"Received unexpected tuple ${tuple}","messagePattern":"Received unexpected tuple (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"storm-core/src/jvm/storm/trident/planner/SubtopologyBolt.java","lineNumber":144,"sourceCode":"        // TODO: get prepared one time into executor data... need to avoid the ser/deser\n        // for each task (probably need storm to support boltfactory)\n    }\n\n    private Fields getSourceOutputFields(TopologyContext context, String sourceStream) {\n        for(GlobalStreamId g: context.getThisSources().keySet()) {\n            if(g.get_streamId().equals(sourceStream)) {\n                return context.getComponentOutputFields(g);\n            }\n        }\n        throw new RuntimeException(\"Could not find fields for source stream \" + sourceStream);\n    }\n    \n    @Override\n    public void execute(BatchInfo batchInfo, Tuple tuple) {\n        String sourceStream = tuple.getSourceStreamId();\n        InitialReceiver ir = _roots.get(sourceStream);\n        if(ir==null) {\n            throw new RuntimeException(\"Received unexpected tuple \" + tuple.toString());\n        }\n        ir.receive((ProcessorContext) batchInfo.state, tuple);\n    }\n\n    @Override\n    public void finishBatch(BatchInfo batchInfo) {\n        for(TridentProcessor p: _myTopologicallyOrdered.get(batchInfo.batchGroup)) {\n            p.finishBatch((ProcessorContext) batchInfo.state);\n        }\n    }\n\n    @Override\n    public Object initBatchState(String batchGroup, Object batchId) {\n        ProcessorContext ret = new ProcessorContext(batchId, new Object[_nodes.size()]);\n        for(TridentProcessor p: _myTopologicallyOrdered.get(batchGroup)) {\n            p.startBatch(ret);\n        }\n        return ret;","sourceCodeStart":126,"sourceCodeEnd":162,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/storm/trident/planner/SubtopologyBolt.java#L126-L162","documentation":"SubtopologyBolt.execute looks up the InitialReceiver registered for the incoming tuple's source stream in the _roots map. If the tuple arrives on a stream for which no root receiver was registered during prepare, it throws this RuntimeException. It means the bolt received a tuple on an unexpected stream, so the topology wiring diverges from what the planner expected.","triggerScenarios":"A tuple is delivered to SubtopologyBolt.execute whose getSourceStreamId() is not a key in _roots; e.g. an extra/direct stream feeds the bolt, or the stream id in _roots was set differently from the actual incoming stream id.","commonSituations":"Custom Trident nodes emitting on misnamed streams; accidental multi-stream inputs to a subtopology bolt; topology rebuilt after modifying stream names without re-registering roots; mixing Trident with raw Storm bolts feeding the same bolt.","solutions":["Log tuple.getSourceStreamId() and compare against the stream ids registered in the roots map; correct whichever side is wrong.","Ensure every stream connected to the bolt is registered in the _roots map during prepare with a matching InitialReceiver.","Check the upstream spout/function is not emitting on an extra stream id.","Rebuild the topology using TridentTopology.newStream/each/partitionPersist so wiring is generated consistently."],"exampleFix":"// before\nroots.put(\"spout-stream\", new InitialReceiver(\"spout-stream\", fields));\n// after: derive from the actual GlobalStreamId of each source\nfor (GlobalStreamId g : context.getThisSources().keySet()) {\n    roots.put(g.get_streamId(), new InitialReceiver(g.get_streamId(), sourceFields));\n}","handlingStrategy":"validation","validationCode":"// Java: assert every source stream has a registered InitialReceiver before activate\nfor (GlobalStreamId g : context.getThisSources().keySet()) {\n    if (!_roots.containsKey(g.get_streamId())) {\n        throw new IllegalStateException(\"No InitialReceiver for stream \" + g.get_streamId());\n    }\n}","typeGuard":null,"tryCatchPattern":"try { ir.receive(ctx, tuple); } catch (RuntimeException e) { if (e.getMessage().startsWith(\"Received unexpected tuple\")) { log.warn(\"Dropping tuple on unknown stream: {}\", tuple.getSourceStreamId()); return; } throw e; }","preventionTips":["Register one InitialReceiver per actual source stream id from getThisSources()","Avoid feeding raw Storm streams into Trident subtopology bolts","Cover spout-to-bolt routing with integration tests"],"tags":["trident","tuple-routing","unexpected-input","runtime"],"backgroundTag":"unexpected-response-shape","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"}