nathanmarz/storm · error · RuntimeException

Cannot join DRPC stream with streams originating from other…

Error message

Cannot join DRPC stream with streams originating from other spouts

What it means

TridentTopology.checkValidJoins() (invoked from completeDRPC) scans spout nodes and rejects topologies that mix a DRPC-type spout with any other batch spout, because a DRPC join requires the DRPC stream to be the only external input. Mixing makes the distributed join semantics undefined, so build fails with this RuntimeException.

Solutions

Remove the non-DRPC spout from the DRPC topology and serve lookup data another way (e.g. static state, TridentState, or an in-function lookup)
Split into two topologies: one DRPC query topology and one batch/state topology that persists data the query reads via TridentState
Restructure the query to not join DRPC args with batch-spout streams

Example fix

// before
TridentTopology t = new TridentTopology();
Stream args = t.newDRPCStream(drpc);
Stream ref = t.newStream("ref", new RefBatchSpout()); // batch spout
args.join(ref, ...).every(...).project(...);
// after: load ref data via TridentState / static lookup instead of a batch spout
TridentState state = t.newStaticState(new RefStateFactory());
t.newDRPCStream(drpc).stateQuery(state, ...).project(...);

Defensive patterns

Strategy: validation

Validate before calling

// Before completeDRPC, ensure only DRPC spouts exist in the topology:
// audit all newStream(...) calls in the DRPC topology and remove any non-DRPC spout.
assert topologyHasOnlyDrpcSpouts(tridentTopology) : "DRPC topology must not contain batch spouts";

Try / catch

try {
    topology.build();
} catch (RuntimeException e) {
    if (e.getMessage().contains("Cannot join DRPC stream")) {
        throw new IllegalStateException("Move the non-DRPC stream into a separate topology or use TridentState", e);
    } throw e;
}

Prevention

Keep DRPC query topologies free of regular newStream() spouts
Serve reference data via TridentState/static state instead of batch spouts in DRPC flows
Call topology.build() in tests to catch this before deployment

When it happens

Trigger: In a DRPC topology, calling completeDRPC() while the same topology also creates a stream from a regular IBatchSpout/ITridentSpout (SpoutType.BATCH), so both hasBatchSpout and hasDRPCSpout are true.

Common situations: Adding a lookup/reference data stream from a normal spout into a DRPC query topology that joins with the DRPC args stream; combining an online query pipeline with a batch enrichment source in one TridentTopology.

Understand the failure class

Background: Conflicting config options: "cannot be used together" — configuration validation errors across open-source libraries — this error's family across 162 libraries.

Related errors

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

Appendix: source

Thrown at storm-core/src/jvm/storm/trident/TridentTopology.java:488

        }
        return null;
    }
    
    private static void checkValidJoins(Collection<Node> g) {
        boolean hasDRPCSpout = false;
        boolean hasBatchSpout = false;
        for(Node n: g) {
            if(n instanceof SpoutNode) {
                SpoutNode.SpoutType type = ((SpoutNode) n).type;
                if(type==SpoutNode.SpoutType.BATCH) {
                    hasBatchSpout = true;
                } else if(type==SpoutNode.SpoutType.DRPC) {
                    hasDRPCSpout = true;
                }
            }
        }
        if(hasBatchSpout && hasDRPCSpout) {
            throw new RuntimeException("Cannot join DRPC stream with streams originating from other spouts");
        }
    }
    
    private static boolean isSpoutGroup(Group g) {
        return g.nodes.size() == 1 && g.nodes.iterator().next() instanceof SpoutNode;
    }
    
    private static Collection<PartitionNode> uniquedSubscriptions(Set<PartitionNode> subscriptions) {
        Map<String, PartitionNode> ret = new HashMap();
        for(PartitionNode n: subscriptions) {
            PartitionNode curr = ret.get(n.streamId);
            if(curr!=null && !curr.thriftGrouping.equals(n.thriftGrouping)) {
                throw new RuntimeException("Multiple subscriptions to the same stream with different groupings. Should be impossible since that is explicitly guarded against.");
            }
            ret.put(n.streamId, n);
        }
        return ret.values();
    }

View on GitHub (pinned to cdb116e942)