{"record":{"id":"1d65d487cf304654","repo":"nathanmarz/storm","slug":"could-not-find-fields-for-source-stream-sourcestream","errorCode":null,"errorMessage":"Could not find fields for source stream ${sourceStream}","messagePattern":"Could not find fields for source stream (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"storm-core/src/jvm/storm/trident/planner/SubtopologyBolt.java","lineNumber":136,"sourceCode":"                        stateIndex,\n                        batchCollector\n                        );\n                pn.processor.prepare(conf, context, triContext);\n                _outputFactories.put(n, pn.processor.getOutputFactory());\n            }   \n            stateIndex++;\n        }        \n        // 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    }","sourceCodeStart":118,"sourceCodeEnd":154,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/storm/trident/planner/SubtopologyBolt.java#L118-L154","documentation":"SubtopologyBolt.getSourceOutputFields resolves the output Fields of an upstream Trident source stream by scanning the topology context's this-sources map for a matching stream id. If no incoming GlobalStreamId matches the requested sourceStream, it throws this RuntimeException. It indicates the bolt was asked about a stream it never receives, meaning the Trident topology graph wiring is inconsistent.","triggerScenarios":"Called from SubtopologyBolt.prepare via getSourceOutputFields when the stream id passed in InitialReceiver setup does not appear in context.getThisSources(); i.e. a Trident topology was built where a subtopology bolt is expected to consume a stream that was never connected as its input.","commonSituations":"Building Trident topologies programmatically with custom spouts/nodes whose stream ids were renamed or mis-declared; upgrading Storm versions where stream wiring internals changed; hand-assembled Graphs/Nodes in custom Trident planners.","solutions":["Check that the Trident Stream feeding the subtopology is actually attached as an input to this bolt (verify Stream.parallelismHint/to/bolt wiring).","Verify the stream id string matches the id declared by the upstream spout/node; fix any renamed or mismatched stream ids.","Rebuild the topology through the standard Trident Stream/TridentTopology API instead of manual graph assembly.","If using a custom planner, ensure it registers all source streams in the TopologyContext before prepare."],"exampleFix":"// before (manual wiring with wrong stream id)\nHashMap<String, InitialReceiver> roots = new HashMap<>();\nroots.put(\"batch\", receiver);\n// after (use the actual stream id declared by the upstream source)\nString streamId = sourceStreamId; // from GlobalStreamId.get_streamId()\nroots.put(streamId, new InitialReceiver(streamId, outputFields));","handlingStrategy":"validation","validationCode":"// Java: before building/submitting the topology, verify each input stream is declared\nfor (GlobalStreamId g : topologyContext.getThisSources().keySet()) {\n    if (!expectedStreamIds.contains(g.get_streamId())) {\n        throw new IllegalStateException(\"Undeclared source stream: \" + g.get_streamId());\n    }\n}","typeGuard":null,"tryCatchPattern":"try { topology.build(); } catch (RuntimeException e) { if (e.getMessage().contains(\"Could not find fields for source stream\")) { log.error(\"Stream wiring mismatch: {}\", e.getMessage()); throw new TopologyConfigurationException(e); } throw e; }","preventionTips":["Always build Trident topologies through the TridentTopology Stream API, never manual node wiring","Keep stream id strings as constants shared by producer and consumer","After topology refactors, re-verify all stream connections before submitting"],"tags":["trident","topology-wiring","stream-not-found","runtime"],"backgroundTag":"resource-not-found","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"}