nathanmarz/storm · error · RuntimeException

Output stream of last component in LinearDRPCTopology must…

Error message

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.

What it means

After requiring exactly one stream, createTopology checks that the last bolt's output stream has exactly two fields: the request id followed by the result, since JoinResult and the return-logic rely on that contract. Otherwise a RuntimeException is thrown.

Solutions

1. Emit exactly two values per tuple: the request id, then the result (emit(new Object[]{id, result})).
2. Declare fields as new Fields("id", "result") — names are yours, count must be 2.
3. Move extra data into the result object itself (e.g. a Map or POJO holding all results).

Example fix

// before
declarer.declare(new Fields("id", "result", "latency"));
// after
declarer.declare(new Fields("id", "result")); // put latency inside result object

OutputFieldsGetter g = new OutputFieldsGetter();
lastBolt.declareOutputFields(g);
Map<String, StreamInfo> streams = g.getFieldsDeclaration();
List<String> fields = streams.values().iterator().next().get_output_fields();
if (fields.size() != 2) {
    throw new IllegalStateException("Last DRPC bolt must output exactly 2 fields (id, result), got " + fields.size());
}

When it happens

Trigger: Declaring the last bolt's single stream with a field count != 2 (e.g. Fields("id","result","extra") or Fields("id")) in a LinearDRPCTopology.

Common situations: Adding extra diagnostic fields to the final emit; returning tuples/lists and forgetting the count contract; copying a normal (non-DRPC) bolt as the last component.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• Must declare exactly one stream from last bolt in…
• No DRPC servers configured for topology
• Cannot join DRPC stream with streams originating from other…
• Each element of the list
• Field must be an Iterable of

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

Appendix: source

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

                    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;
    }
    
    private static class Component {
        public IRichBolt bolt;
        public int parallelism;

View on GitHub (pinned to cdb116e942)