nathanmarz/storm · error · RuntimeException

Must declare exactly one stream from last bolt in…

Error message

Must declare exactly one stream from last bolt in LinearDRPCTopology

What it means

LinearDRPCTopologyBuilder.createTopology inspects the last bolt's declared output streams; the framework pipes its single output stream into JoinResult. If the last bolt declares zero or multiple streams, a RuntimeException is thrown because the linear DRPC wiring is impossible.

Solutions

Make the last bolt declare exactly one output stream (use plain declare(), not multiple stream() calls).
Move auxiliary streams (error streams etc.) to a bolt that is not last.
If multiple outputs are needed, add a final adapter bolt that merges them into one stream.

Example fix

// before
public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declareStream("result", new Fields("id", "result"));
    declarer.declareStream("errors", new Fields("id", "msg"));
}
// after
public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declare(new Fields("id", "result"));
}

Defensive patterns

Strategy: validation

Validate before calling

// verify last bolt declares exactly one stream before createTopology
OutputFieldsGetter g = new OutputFieldsGetter();
lastBolt.declareOutputFields(g);
if (g.getFieldsDeclaration().size() != 1) {
    throw new IllegalStateException("Last bolt must declare exactly one stream");
}

Prevention

In the final bolt always use plain declare(), never stream().
Keep LinearDRPC topologies' last bolt minimal and dedicated to emitting the result.
Add a unit test calling createLocalTopology to catch declaration issues early.

When it happens

Trigger: Implementing the final bolt's declareOutputFields with multiple outputFieldsDeclarer.stream(...) declarations, or none, in a LinearDRPCTopology.

Common situations: Refactoring a bolt to add a second (error/reporting) stream; copying a multi-stream bolt as the last stage; forgetting to declare any stream.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

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

Appendix: source

Thrown at storm-core/src/jvm/backtype/storm/drpc/LinearDRPCTopologyBuilder.java:152

                    prevId = PREPARE_ID;
                } else {
                    prevId = boltId(i-1);
                }
                for(InputDeclaration declaration: component.declarations) {
                    declaration.declare(prevId, declarer);
                }
            }
            if(i>0) {
                declarer.directGrouping(boltId(i-1), Constants.COORDINATED_STREAM_ID); 
            }
        }
        
        IRichBolt lastBolt = _components.get(_components.size()-1).bolt;
        OutputFieldsGetter getter = new OutputFieldsGetter();
        lastBolt.declareOutputFields(getter);
        Map<String, StreamInfo> streams = getter.getFieldsDeclaration();
        if(streams.size()!=1) {
            throw new RuntimeException("Must declare exactly one stream from last bolt in LinearDRPCTopology");
        }
        String outputStream = streams.keySet().iterator().next();
        List<String> fields = streams.get(outputStream).get_output_fields();
        if(fields.size()!=2) {
            throw new RuntimeException("Output stream of last component in LinearDRPCTopology must contain exactly two fields. The first should be the request id, and the second should be the result.");
        }

        builder.setBolt(boltId(i), new JoinResult(PREPARE_ID))
                .fieldsGrouping(boltId(i-1), outputStream, new Fields(fields.get(0)))
                .fieldsGrouping(PREPARE_ID, PrepareRequest.RETURN_STREAM, new Fields("request"));
        i++;
        builder.setBolt(boltId(i), new ReturnResults())
                .noneGrouping(boltId(i-1));
        return builder.createTopology();
    }
    
    private static String boltId(int index) {
        return "bolt" + index;

View on GitHub (pinned to cdb116e942)