nathanmarz/storm · error · RuntimeException

Can only do an identity grouping when source and target…

Error message

Can only do an identity grouping when source and target have same number of tasks

What it means

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).

Solutions

1. Give the source and target components the same parallelism (matching parallelismHint)
2. Use shuffle/shuffleGrouping instead if equal task counts cannot be guaranteed
3. Repartition with partitionBy on a key rather than identity grouping

Example fix

// before
builder.setBolt("b", bolt, 4).identityGrouping("a", stream); // source a has 2 tasks
// after
builder.setBolt("b", bolt, 2).identityGrouping("a", stream); // matches source parallelism

// before declaring identity grouping, ensure equal parallelism
int srcTasks = conf.getComponentTasks(sourceComponentId).size();
int dstTasks = targetParallelism;
if (srcTasks != dstTasks) throw new IllegalStateException("identityGrouping requires equal task counts");

try {
    prepare(context, stream, tasks);
} catch (RuntimeException e) {
    if (e.getMessage().contains("identity grouping")) {
        throw new IllegalStateException("Match source and target parallelism or use shuffle grouping", e);
    } throw e;
}

When it happens

Trigger: 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.

Common situations: 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.

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

• Parallelism is fixed to
• A single worker should have 1 SystemBolt instance.
• No output fields defined for component:stream
• Fields for already set
• Bolt has already been declared for id

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

Appendix: source

Thrown at storm-core/src/jvm/storm/trident/partition/IdentityGrouping.java:41

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


public class IdentityGrouping implements CustomStreamGrouping {

    List<Integer> ret = new ArrayList<Integer>();
    Map<Integer, List<Integer>> _precomputed = new HashMap();
    
    @Override
    public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> tasks) {
        List<Integer> sourceTasks = new ArrayList<Integer>(context.getComponentTasks(stream.get_componentId()));
        Collections.sort(sourceTasks);
        if(sourceTasks.size()!=tasks.size()) {
            throw new RuntimeException("Can only do an identity grouping when source and target have same number of tasks");
        }
        tasks = new ArrayList<Integer>(tasks);
        Collections.sort(tasks);
        for(int i=0; i<sourceTasks.size(); i++) {
            int s = sourceTasks.get(i);
            int t = tasks.get(i);
            _precomputed.put(s, Arrays.asList(t));
        }
    }

    @Override
    public List<Integer> chooseTasks(int task, List<Object> values) {
        List<Integer> ret = _precomputed.get(task);
        if(ret==null) {
            throw new RuntimeException("Tuple emitted by task that's not part of this component. Should be impossible");
        }
        return ret;
    }

View on GitHub (pinned to cdb116e942)