nathanmarz/storm · warning

Failed to emit batch for transaction

Error message

Failed to emit batch for transaction

What it means

In TransactionalSpoutBatchExecutor.execute(), any FailedException raised while emitting a transactional batch is caught, logged as 'Failed to emit batch for transaction', and the input tuple is explicitly failed via _collector.fail(input). Failing the tuple tells Storm to replay the transactional batch from the spout, so this is the designed retry mechanism for transactional spouts rather than a fatal crash.

Solutions

1. Inspect the logged FailedException cause — it identifies the emitter failure that must be fixed (source down, bad query, permissions).
2. Restore connectivity/health of the underlying data source; the failed tuples will be replayed automatically.
3. Add retry/backoff inside the emitter for transient source errors if replays are frequent.
4. Do not throw general RuntimeException for expected-transient problems — throw FailedException deliberately so the batch is replayed instead of killing the worker.
5. Monitor the spout's fail counts; persistent failures mean the source problem is not transient.

Example fix

// before
public void emitBatch(TransactionAttempt tx, Map coordinatorMeta, BatchOutputCollector collector) {
    ResultSet rs = conn.createStatement().executeQuery(query); // SQLException kills worker
}
// after
public void emitBatch(TransactionAttempt tx, Map coordinatorMeta, BatchOutputCollector collector) {
    try {
        ResultSet rs = conn.createStatement().executeQuery(query);
    } catch (SQLException e) {
        throw new FailedException(e); // batch is replayed instead of crashing
    }
}

// health-check the data source before the spout emits
if (!isDataSourceHealthy(dataSource)) { throw new FailedException("data source unavailable"); }

public void emitBatch(TransactionAttempt tx, Map meta, BatchOutputCollector collector) {
    try {
        doEmit(tx, meta, collector);
    } catch (TransientException e) {
        throw new FailedException(e); // replay batch
    }
}

When it happens

Trigger: The transactional spout emitter (or the emit path in execute()) throws backtype.storm.topology.FailedException — e.g. the underlying data source is unavailable, the batch emit fails, or application code inside the emitter deliberately signals a replayable failure.

Common situations: Database/Kafka/data-source outage while reading the batch; emitter code hitting a transient error it deems retryable; misconfigured data source credentials causing the emitter to fail on read; long rebalance windows making the source temporarily unreachable.

Related errors

• Trying to initialize transaction for which there should be…
• Expecting previous txid state to be the previous transaction
• Expecting tx state to be initialized in strict order but…
• Failed to get metadata for a transaction
• Each element of the list

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

Appendix: source

Thrown at storm-core/src/jvm/backtype/storm/transactional/TransactionalSpoutBatchExecutor.java:77

                    _activeTransactions.remove(attempt.getTransactionId());
                    _collector.ack(input);
                } else {
                    _collector.fail(input);
                }
            } else { 
                _emitter.emitBatch(attempt, input.getValue(1), _collector);
                _activeTransactions.put(attempt.getTransactionId(), attempt);
                _collector.ack(input);
                BigInteger committed = (BigInteger) input.getValue(2);
                if(committed!=null) {
                    // valid to delete before what's been committed since 
                    // those batches will never be accessed again
                    _activeTransactions.headMap(committed).clear();
                    _emitter.cleanupBefore(committed);
                }
            }
        } catch(FailedException e) {
            LOG.warn("Failed to emit batch for transaction", e);
            _collector.fail(input);
        }
    }

    @Override
    public void cleanup() {
        _emitter.close();
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        _spout.declareOutputFields(declarer);
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return _spout.getComponentConfiguration();
    }

View on GitHub (pinned to cdb116e942)