{"record":{"id":"184181a6b83c11b3","repo":"apache/beam","slug":"closing-producer-after-unrecoverable-error-the-work-might","errorCode":null,"errorMessage":"{} : closing producer {} after unrecoverable error. The work might have migrated. Committed id {}, current id {}.","messagePattern":"(.+?) : closing producer (.+?) after unrecoverable error\\. The work might have migrated\\. Committed id (.+?), current id (.+?)\\.","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java","lineNumber":439,"sourceCode":"\n            iter =\n                Iterators.mergeSorted(\n                    ImmutableList.of(iter, buffered.iterator()), new KV.OrderByKey<>());\n          }\n        }\n\n        writer.commitTxn(nextId - 1, numTransactions);\n        nextIdState.write(nextId);\n\n      } catch (ProducerSpEL.UnrecoverableProducerException e) {\n        // Producer JavaDoc says these are not recoverable errors and producer should be closed.\n\n        // Close the producer and a new producer will be initialized in retry.\n        // It is possible that a rough worker keeps retrying and ends up fencing off\n        // active producers. How likely this might be or how well such a scenario is handled\n        // depends on the runner. For now we will leave it to upper layers, will need to revisit.\n\n        LOG.warn(\n            \"{} : closing producer {} after unrecoverable error. The work might have migrated.\"\n                + \" Committed id {}, current id {}.\",\n            writer.shard,\n            writer.producerName,\n            writer.committedId,\n            nextId - 1,\n            e);\n\n        writer.producer.close();\n        writer = null; // No need to cache it.\n        throw e;\n      } finally {\n        if (writer != null) {\n          cache.insert(shard, writer);\n        }\n      }\n    }\n","sourceCodeStart":421,"sourceCodeEnd":457,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java#L421-L457","documentation":"Warning from KafkaExactlyOnceSink when the producer hits an unrecoverable error and must be closed. The log records which shard/producer was closed, the committed producer ID, and the current producer ID. Work may migrate to another worker; how the runner handles the fencing of active producers is runner-dependent.","triggerScenarios":"A Kafka writeExactlyOnce() sink producer receives an unrecoverable (non-retryable) transactional error (e.g. ProducerFencedException, transaction timeout) in processElement, so the writer closes the producer and retries with a new one (nextId incremented).","commonSituations":"Long-running streaming pipelines exceeding Kafka transaction.timeout.ms, two workers contending for the same shard/transactional ID after rebalances or speculative execution, or broker-side fencing of stale producers.","solutions":["Increase the Kafka broker's transaction.timeout.ms (and the producer's max.block.ms / request timeouts) so long transactions are not fenced.","Check for duplicate pipeline workers/instances sharing the same shard key and eliminate the contention (scale down or fix the shard assignment).","Retry the affected bundle; the sink re-initializes a fresh producer and work migrates.","Upgrade the Kafka client/Beam version if fencing after worker migration is not being resolved correctly."],"exampleFix":"// before\nprops.put(\"transaction.timeout.ms\", \"60000\"); // 1 min, too short for big bundles\n// after\nprops.put(\"transaction.timeout.ms\", \"900000\"); // 15 min","handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Job-level: rerun failed bundles; the sink re-initializes a fresh producer per retry\npipeline.run().waitUntilFinish(); // inspect PipelineResult failures and resubmit","preventionTips":["Set transaction.timeout.ms (default 60s) large enough for your largest bundle, e.g. 900000.","Avoid running duplicate workers with the same shard/transactional IDs.","Monitor for ProducerFencedException and transaction timeouts in broker/client metrics.","Keep Beam and Kafka client versions current for exactly-once fencing fixes."],"tags":["kafka","transactions","exactly-once","producer","beam-io"],"backgroundTag":"producer-fenced","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}