nathanmarz/storm · error · RuntimeException

Received unexpected tuple

Error message

Received unexpected tuple ${tuple}

What it means

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.

Solutions

1. Log tuple.getSourceStreamId() and compare against the stream ids registered in the roots map; correct whichever side is wrong.
2. Ensure every stream connected to the bolt is registered in the _roots map during prepare with a matching InitialReceiver.
3. Check the upstream spout/function is not emitting on an extra stream id.
4. Rebuild the topology using TridentTopology.newStream/each/partitionPersist so wiring is generated consistently.

Example fix

// before
roots.put("spout-stream", new InitialReceiver("spout-stream", fields));
// after: derive from the actual GlobalStreamId of each source
for (GlobalStreamId g : context.getThisSources().keySet()) {
    roots.put(g.get_streamId(), new InitialReceiver(g.get_streamId(), sourceFields));
}

// Java: assert every source stream has a registered InitialReceiver before activate
for (GlobalStreamId g : context.getThisSources().keySet()) {
    if (!_roots.containsKey(g.get_streamId())) {
        throw new IllegalStateException("No InitialReceiver for stream " + g.get_streamId());
    }
}

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

When it happens

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

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

Related errors

• Could not find fields for source stream
• 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/c56408e9925012d6. Report an issue: GitHub.

Appendix: source

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

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

    @Override
    public Object initBatchState(String batchGroup, Object batchId) {
        ProcessorContext ret = new ProcessorContext(batchId, new Object[_nodes.size()]);
        for(TridentProcessor p: _myTopologicallyOrdered.get(batchGroup)) {
            p.startBatch(ret);
        }
        return ret;

View on GitHub (pinned to cdb116e942)