nathanmarz/storm · error · RuntimeException

Could not find fields for source stream

Error message

Could not find fields for source stream ${sourceStream}

What it means

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.

Solutions

1. Check that the Trident Stream feeding the subtopology is actually attached as an input to this bolt (verify Stream.parallelismHint/to/bolt wiring).
2. Verify the stream id string matches the id declared by the upstream spout/node; fix any renamed or mismatched stream ids.
3. Rebuild the topology through the standard Trident Stream/TridentTopology API instead of manual graph assembly.
4. If using a custom planner, ensure it registers all source streams in the TopologyContext before prepare.

Example fix

// before (manual wiring with wrong stream id)
HashMap<String, InitialReceiver> roots = new HashMap<>();
roots.put("batch", receiver);
// after (use the actual stream id declared by the upstream source)
String streamId = sourceStreamId; // from GlobalStreamId.get_streamId()
roots.put(streamId, new InitialReceiver(streamId, outputFields));

// Java: before building/submitting the topology, verify each input stream is declared
for (GlobalStreamId g : topologyContext.getThisSources().keySet()) {
    if (!expectedStreamIds.contains(g.get_streamId())) {
        throw new IllegalStateException("Undeclared source stream: " + g.get_streamId());
    }
}

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; }

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• Received unexpected tuple
• Trying to select non-existent field
• Regular rich spouts not supported yet... try wrapping in a…
• Cannot join DRPC stream with streams originating from other…
• Parallelism is fixed to

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

Appendix: source

Thrown at storm-core/src/jvm/storm/trident/planner/SubtopologyBolt.java:136

                        stateIndex,
                        batchCollector
                        );
                pn.processor.prepare(conf, context, triContext);
                _outputFactories.put(n, pn.processor.getOutputFactory());
            }   
            stateIndex++;
        }        
        // TODO: get prepared one time into executor data... need to avoid the ser/deser
        // for each task (probably need storm to support boltfactory)
    }

    private Fields getSourceOutputFields(TopologyContext context, String sourceStream) {
        for(GlobalStreamId g: context.getThisSources().keySet()) {
            if(g.get_streamId().equals(sourceStream)) {
                return context.getComponentOutputFields(g);
            }
        }
        throw new RuntimeException("Could not find fields for source stream " + sourceStream);
    }
    
    @Override
    public void execute(BatchInfo batchInfo, Tuple tuple) {
        String sourceStream = tuple.getSourceStreamId();
        InitialReceiver ir = _roots.get(sourceStream);
        if(ir==null) {
            throw new RuntimeException("Received unexpected tuple " + tuple.toString());
        }
        ir.receive((ProcessorContext) batchInfo.state, tuple);
    }

    @Override
    public void finishBatch(BatchInfo batchInfo) {
        for(TridentProcessor p: _myTopologicallyOrdered.get(batchInfo.batchGroup)) {
            p.finishBatch((ProcessorContext) batchInfo.state);
        }
    }

View on GitHub (pinned to cdb116e942)