{"record":{"id":"145aaeba68d4a0eb","repo":"nathanmarz/storm","slug":"failed-to-emit-batch-for-transaction","errorCode":null,"errorMessage":"Failed to emit batch for transaction","messagePattern":"Failed to emit batch for transaction","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"storm-core/src/jvm/backtype/storm/transactional/TransactionalSpoutBatchExecutor.java","lineNumber":77,"sourceCode":"                    _activeTransactions.remove(attempt.getTransactionId());\n                    _collector.ack(input);\n                } else {\n                    _collector.fail(input);\n                }\n            } else { \n                _emitter.emitBatch(attempt, input.getValue(1), _collector);\n                _activeTransactions.put(attempt.getTransactionId(), attempt);\n                _collector.ack(input);\n                BigInteger committed = (BigInteger) input.getValue(2);\n                if(committed!=null) {\n                    // valid to delete before what's been committed since \n                    // those batches will never be accessed again\n                    _activeTransactions.headMap(committed).clear();\n                    _emitter.cleanupBefore(committed);\n                }\n            }\n        } catch(FailedException e) {\n            LOG.warn(\"Failed to emit batch for transaction\", e);\n            _collector.fail(input);\n        }\n    }\n\n    @Override\n    public void cleanup() {\n        _emitter.close();\n    }\n\n    @Override\n    public void declareOutputFields(OutputFieldsDeclarer declarer) {\n        _spout.declareOutputFields(declarer);\n    }\n\n    @Override\n    public Map<String, Object> getComponentConfiguration() {\n        return _spout.getComponentConfiguration();\n    }","sourceCodeStart":59,"sourceCodeEnd":95,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/backtype/storm/transactional/TransactionalSpoutBatchExecutor.java#L59-L95","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Inspect the logged FailedException cause — it identifies the emitter failure that must be fixed (source down, bad query, permissions).","Restore connectivity/health of the underlying data source; the failed tuples will be replayed automatically.","Add retry/backoff inside the emitter for transient source errors if replays are frequent.","Do not throw general RuntimeException for expected-transient problems — throw FailedException deliberately so the batch is replayed instead of killing the worker.","Monitor the spout's fail counts; persistent failures mean the source problem is not transient."],"exampleFix":"// before\npublic void emitBatch(TransactionAttempt tx, Map coordinatorMeta, BatchOutputCollector collector) {\n    ResultSet rs = conn.createStatement().executeQuery(query); // SQLException kills worker\n}\n// after\npublic void emitBatch(TransactionAttempt tx, Map coordinatorMeta, BatchOutputCollector collector) {\n    try {\n        ResultSet rs = conn.createStatement().executeQuery(query);\n    } catch (SQLException e) {\n        throw new FailedException(e); // batch is replayed instead of crashing\n    }\n}","handlingStrategy":"try-catch","validationCode":"// health-check the data source before the spout emits\nif (!isDataSourceHealthy(dataSource)) { throw new FailedException(\"data source unavailable\"); }","typeGuard":null,"tryCatchPattern":"public void emitBatch(TransactionAttempt tx, Map meta, BatchOutputCollector collector) {\n    try {\n        doEmit(tx, meta, collector);\n    } catch (TransientException e) {\n        throw new FailedException(e); // replay batch\n    }\n}","preventionTips":["Throw FailedException (not RuntimeException) for replayable batch failures.","Monitor spout fail/complete metrics to catch persistent source outages.","Add backoff and bounded retries in the emitter for transient backend errors.","Verify data-source credentials and connectivity as part of deploy checks."],"tags":["storm","transactional-spout","retry","tuple-fail","batch"],"backgroundTag":"network-request-failed","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"}