{"record":{"id":"7079872bca7409b9","repo":"nathanmarz/storm","slug":"can-only-do-an-identity-grouping-when-source-and-target-have","errorCode":null,"errorMessage":"Can only do an identity grouping when source and target have same number of tasks","messagePattern":"Can only do an identity grouping when source and target have same number of tasks","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"storm-core/src/jvm/storm/trident/partition/IdentityGrouping.java","lineNumber":41,"sourceCode":"import java.util.ArrayList;\nimport java.util.Arrays;\nimport java.util.Collections;\nimport java.util.HashMap;\nimport java.util.List;\nimport java.util.Map;\n\n\npublic class IdentityGrouping implements CustomStreamGrouping {\n\n    List<Integer> ret = new ArrayList<Integer>();\n    Map<Integer, List<Integer>> _precomputed = new HashMap();\n    \n    @Override\n    public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> tasks) {\n        List<Integer> sourceTasks = new ArrayList<Integer>(context.getComponentTasks(stream.get_componentId()));\n        Collections.sort(sourceTasks);\n        if(sourceTasks.size()!=tasks.size()) {\n            throw new RuntimeException(\"Can only do an identity grouping when source and target have same number of tasks\");\n        }\n        tasks = new ArrayList<Integer>(tasks);\n        Collections.sort(tasks);\n        for(int i=0; i<sourceTasks.size(); i++) {\n            int s = sourceTasks.get(i);\n            int t = tasks.get(i);\n            _precomputed.put(s, Arrays.asList(t));\n        }\n    }\n\n    @Override\n    public List<Integer> chooseTasks(int task, List<Object> values) {\n        List<Integer> ret = _precomputed.get(task);\n        if(ret==null) {\n            throw new RuntimeException(\"Tuple emitted by task that's not part of this component. Should be impossible\");\n        }\n        return ret;\n    }","sourceCodeStart":23,"sourceCodeEnd":59,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/storm/trident/partition/IdentityGrouping.java#L23-L59","documentation":"IdentityGrouping routes tuples so source task i sends to target task i, preserving task-index identity. prepare() compares the number of source component tasks with the target tasks; if the counts differ, identity pairing is impossible and a RuntimeException is thrown during worker prepare (via makeContext).","triggerScenarios":"Declaring a stream with identityGrouping (e.g. Trident's partitionBy/identity paths or a topology .customGrouping(..., new IdentityGrouping())) where the upstream component's parallelism differs from the target's parallelism.","commonSituations":"Setting parallelismHint(4) on the target of an identity grouping while the source has 2 executors; topology rebalancing/numWorkers changes that shift one side's task count; forgetting that identity grouping requires matching parallelism.","solutions":["Give the source and target components the same parallelism (matching parallelismHint)","Use shuffle/shuffleGrouping instead if equal task counts cannot be guaranteed","Repartition with partitionBy on a key rather than identity grouping"],"exampleFix":"// before\nbuilder.setBolt(\"b\", bolt, 4).identityGrouping(\"a\", stream); // source a has 2 tasks\n// after\nbuilder.setBolt(\"b\", bolt, 2).identityGrouping(\"a\", stream); // matches source parallelism","handlingStrategy":"validation","validationCode":"// before declaring identity grouping, ensure equal parallelism\nint srcTasks = conf.getComponentTasks(sourceComponentId).size();\nint dstTasks = targetParallelism;\nif (srcTasks != dstTasks) throw new IllegalStateException(\"identityGrouping requires equal task counts\");","typeGuard":null,"tryCatchPattern":"try {\n    prepare(context, stream, tasks);\n} catch (RuntimeException e) {\n    if (e.getMessage().contains(\"identity grouping\")) {\n        throw new IllegalStateException(\"Match source and target parallelism or use shuffle grouping\", e);\n    } throw e;\n}","preventionTips":["Pin matching parallelismHint on both ends of an identity grouping","Avoid identity grouping when topology may rebalance task counts","Prefer shuffleGrouping/partitionBy unless task-index identity is required"],"tags":["storm","grouping","parallelism","topology"],"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"}