{"record":{"id":"41dc75e40248a4c3","repo":"nathanmarz/storm","slug":"trident-does-not-support-direct-streams-tridentutils","errorCode":null,"errorMessage":"Trident does not support direct streams","messagePattern":"Trident does not support direct streams","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"storm-core/src/jvm/storm/trident/util/TridentUtils.java","lineNumber":75,"sourceCode":"        List<String> toKeep = new ArrayList<String>();\n        for(String s: all.toList()) {\n            if(!removeSet.contains(s)) {\n                toKeep.add(s);\n            }\n        }\n        return new Fields(toKeep);\n    }\n    \n    public static Fields getSingleOutputStreamFields(IComponent component) {\n        OutputFieldsGetter getter = new OutputFieldsGetter();\n        component.declareOutputFields(getter);\n        Map<String, StreamInfo> declaration = getter.getFieldsDeclaration();\n        if(declaration.size()!=1) {\n            throw new RuntimeException(\"Trident only supports components that emit a single stream\");\n        }\n        StreamInfo si = declaration.values().iterator().next();\n        if(si.is_direct()) {\n            throw new RuntimeException(\"Trident does not support direct streams\");\n        }\n        return new Fields(si.get_output_fields());        \n    }\n    \n    /**\n     * Assumes edge contains an index\n     */\n    public static <T> List<T> getParents(DirectedGraph g, T n) {\n        List<IndexedEdge> incoming = new ArrayList(g.incomingEdgesOf(n));\n        Collections.sort(incoming);\n        List<T> ret = new ArrayList();\n        for(IndexedEdge e: incoming) {\n            ret.add((T)e.source);\n        }        \n        return ret;\n    }\n\n    public static <T> List<T> getChildren(DirectedGraph g, T n) {","sourceCodeStart":57,"sourceCodeEnd":93,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/storm/trident/util/TridentUtils.java#L57-L93","documentation":"Trident does not support direct streams (streams declared with direct grouping, where the downstream decides routing). After checking that a component declares a single stream, getSingleOutputStreamFields inspects StreamInfo.is_direct(); a direct declaration is rejected with this RuntimeException at TridentUtils.java:75.","triggerScenarios":"Passing to Trident any component whose declareOutputFields uses declarer.declareStream(name, true, fields) or otherwise marks the stream as direct, then invoking Trident topology construction (newStream, each, etc.).","commonSituations":"Reusing Storm bolts that use direct streams for custom grouping inside Trident; porting components written for partialKeyGrouping or custom routing patterns; copying Storm examples into a Trident codebase.","solutions":["Remove the direct flag from the component's stream declaration (declare non-direct fields).","Replace direct-grouping logic with a Trident-supported grouping (shuffle, fields, etc.).","Move the component to a plain Storm topology if direct streams are required."],"exampleFix":"// before\ndeclarer.declareStream(\"out\", true, new Fields(\"word\"));\n// after\ndeclarer.declare(new Fields(\"word\"));","handlingStrategy":"validation","validationCode":"OutputFieldsGetter getter = new OutputFieldsGetter();\ncomponent.declareOutputFields(getter);\nfor (StreamInfo si : getter.getFieldsDeclaration().values()) {\n    if (si.is_direct()) throw new IllegalArgumentException(\"Direct streams are not supported by Trident\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Never pass `true` as the direct flag in components used with Trident.","Prefer standard Trident groupings over custom direct routing.","Audit reused Storm bolts for direct declarations before adding them to Trident topologies."],"tags":["trident","direct-stream","stream-declaration","storm"],"backgroundTag":"unsupported-operation","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"}