nathanmarz/storm · critical · RuntimeException

Anchored onto after ack/fail

Error message

Anchored onto ${o} after ack/fail

What it means

When a ShellBolt emits an anchored tuple, each anchor id must still resolve to a pending input tuple in _inputs. If the shell script anchors onto an id that was already acked/failed (removed from the map) or never received, Storm throws this RuntimeException because the anchor tuple cannot be resolved.

Solutions

1. Only anchor on tuple ids from the CURRENT execute/nextTuple message that have not been acked/failed yet.
2. Remove any retry/re-emit logic that reuses previously acked tuple ids as anchors.
3. Emit unanchored (anchors=[] ) if delayed emission is intended, accepting no anchoring guarantees.
4. Keep anchor emission in the same processing step as the input tuple, before acking it.
5. Log anchor ids in the script to identify which id is stale.

Example fix

# before
process(tuple)
ack(tuple['id'])
emit([tuple['id']], [value])  # anchor already acked

# after
emit([tuple['id']], [value])  # anchor while still pending
ack(tuple['id'])

# before emitting with anchors:
anchors = [i for i in anchor_ids if i in pending]

def is_anchorable(anchor_id):
    return anchor_id in pending

When it happens

Trigger: Shell process calls emit with an 'anchor' list containing a tuple id that has already been acked/failed, or an id that was never passed to the script, e.g. emitting with anchors=['<old id>'] after acking that tuple.

Common situations: Multilang scripts that cache tuple ids and re-emit them later (retry logic), scripts anchoring on ids from previous execute() calls, or scripts that generate anchor ids instead of using the ones received.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Failed a non-existent or already acked/failed id
• Error when launching multilang subprocess
• Error when launching multilang subprocess
• Error during multilang processing
• Acked a non-existent or already acked/failed id

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

Appendix: source

Thrown at storm-core/src/jvm/backtype/storm/task/ShellBolt.java:226

        String msg = (String) action.get("msg");
        _collector.reportError(new Exception("Shell Process Exception: " + msg));
    }

    private void handleEmit(Map action) throws InterruptedException {
        String stream = (String) action.get("stream");
        if(stream==null) stream = Utils.DEFAULT_STREAM_ID;
        Long task = (Long) action.get("task");
        List<Object> tuple = (List) action.get("tuple");
        List<Tuple> anchors = new ArrayList<Tuple>();
        Object anchorObj = action.get("anchors");
        if(anchorObj!=null) {
            if(anchorObj instanceof String) {
                anchorObj = Arrays.asList(anchorObj);
            }
            for(Object o: (List) anchorObj) {
                Tuple t = _inputs.get((String) o);
                if (t == null) {
                    throw new RuntimeException("Anchored onto " + o + " after ack/fail");
                }
                anchors.add(t);
            }
        }
        if(task==null) {
            List<Integer> outtasks = _collector.emit(stream, anchors, tuple);
            Object need_task_ids = action.get("need_task_ids");
            if (need_task_ids == null || ((Boolean) need_task_ids).booleanValue()) {
                _pendingWrites.put(outtasks);
            }
        } else {
            _collector.emitDirect((int)task.longValue(), stream, anchors, tuple);
        }
    }

    private void die(Throwable exception) {
        _exception = exception;
    }

View on GitHub (pinned to cdb116e942)