{"record":{"id":"a40bd8f37ba68f87","repo":"nathanmarz/storm","slug":"cannot-join-drpc-stream-with-streams-originating-from-other","errorCode":null,"errorMessage":"Cannot join DRPC stream with streams originating from other spouts","messagePattern":"Cannot join DRPC stream with streams originating from other spouts","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"storm-core/src/jvm/storm/trident/TridentTopology.java","lineNumber":488,"sourceCode":"        }\n        return null;\n    }\n    \n    private static void checkValidJoins(Collection<Node> g) {\n        boolean hasDRPCSpout = false;\n        boolean hasBatchSpout = false;\n        for(Node n: g) {\n            if(n instanceof SpoutNode) {\n                SpoutNode.SpoutType type = ((SpoutNode) n).type;\n                if(type==SpoutNode.SpoutType.BATCH) {\n                    hasBatchSpout = true;\n                } else if(type==SpoutNode.SpoutType.DRPC) {\n                    hasDRPCSpout = true;\n                }\n            }\n        }\n        if(hasBatchSpout && hasDRPCSpout) {\n            throw new RuntimeException(\"Cannot join DRPC stream with streams originating from other spouts\");\n        }\n    }\n    \n    private static boolean isSpoutGroup(Group g) {\n        return g.nodes.size() == 1 && g.nodes.iterator().next() instanceof SpoutNode;\n    }\n    \n    private static Collection<PartitionNode> uniquedSubscriptions(Set<PartitionNode> subscriptions) {\n        Map<String, PartitionNode> ret = new HashMap();\n        for(PartitionNode n: subscriptions) {\n            PartitionNode curr = ret.get(n.streamId);\n            if(curr!=null && !curr.thriftGrouping.equals(n.thriftGrouping)) {\n                throw new RuntimeException(\"Multiple subscriptions to the same stream with different groupings. Should be impossible since that is explicitly guarded against.\");\n            }\n            ret.put(n.streamId, n);\n        }\n        return ret.values();\n    }","sourceCodeStart":470,"sourceCodeEnd":506,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/storm/trident/TridentTopology.java#L470-L506","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before\nTridentTopology t = new TridentTopology();\nStream args = t.newDRPCStream(drpc);\nStream ref = t.newStream(\"ref\", new RefBatchSpout()); // batch spout\nargs.join(ref, ...).every(...).project(...);\n// after: load ref data via TridentState / static lookup instead of a batch spout\nTridentState state = t.newStaticState(new RefStateFactory());\nt.newDRPCStream(drpc).stateQuery(state, ...).project(...);","handlingStrategy":"validation","validationCode":"// Before completeDRPC, ensure only DRPC spouts exist in the topology:\n// audit all newStream(...) calls in the DRPC topology and remove any non-DRPC spout.\nassert topologyHasOnlyDrpcSpouts(tridentTopology) : \"DRPC topology must not contain batch spouts\";","typeGuard":null,"tryCatchPattern":"try {\n    topology.build();\n} catch (RuntimeException e) {\n    if (e.getMessage().contains(\"Cannot join DRPC stream\")) {\n        throw new IllegalStateException(\"Move the non-DRPC stream into a separate topology or use TridentState\", e);\n    } throw e;\n}","preventionTips":["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"],"tags":["storm","trident","drpc","join"],"backgroundTag":"conflicting-config-options","analyzedSha":"cdb116e942666973bc4eaa0df098d5bab82739e7","analyzedAt":"2026-09-12T14:30:00.714Z","contentChangedAt":"2026-09-12T14:30:00.714Z","schemaVersion":2},"datasetVersion":"2026-09-16T09:17:16.951Z"}