apache/beam · error

: closing producer after unrecoverable error. The work…

Error message

{} : closing producer {} after unrecoverable error. The work might have migrated. Committed id {}, current id {}.

What it means

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.

Solutions

1. Increase the Kafka broker's transaction.timeout.ms (and the producer's max.block.ms / request timeouts) so long transactions are not fenced.
2. Check for duplicate pipeline workers/instances sharing the same shard key and eliminate the contention (scale down or fix the shard assignment).
3. Retry the affected bundle; the sink re-initializes a fresh producer and work migrates.
4. Upgrade the Kafka client/Beam version if fencing after worker migration is not being resolved correctly.

Example fix

// before
props.put("transaction.timeout.ms", "60000"); // 1 min, too short for big bundles
// after
props.put("transaction.timeout.ms", "900000"); // 15 min

// Job-level: rerun failed bundles; the sink re-initializes a fresh producer per retry
pipeline.run().waitUntilFinish(); // inspect PipelineResult failures and resubmit

When it happens

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

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

Related errors

• Kafka metadata exists for shard
• : consumer thread is interrupted
• consumerPollingTimeout should be > 0.
• Couldn't infer Coder from
• Error while parsing the element

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/184181a6b83c11b3. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java:439

            iter =
                Iterators.mergeSorted(
                    ImmutableList.of(iter, buffered.iterator()), new KV.OrderByKey<>());
          }
        }

        writer.commitTxn(nextId - 1, numTransactions);
        nextIdState.write(nextId);

      } catch (ProducerSpEL.UnrecoverableProducerException e) {
        // Producer JavaDoc says these are not recoverable errors and producer should be closed.

        // Close the producer and a new producer will be initialized in retry.
        // It is possible that a rough worker keeps retrying and ends up fencing off
        // active producers. How likely this might be or how well such a scenario is handled
        // depends on the runner. For now we will leave it to upper layers, will need to revisit.

        LOG.warn(
            "{} : closing producer {} after unrecoverable error. The work might have migrated."
                + " Committed id {}, current id {}.",
            writer.shard,
            writer.producerName,
            writer.committedId,
            nextId - 1,
            e);

        writer.producer.close();
        writer = null; // No need to cache it.
        throw e;
      } finally {
        if (writer != null) {
          cache.insert(shard, writer);
        }
      }
    }

View on GitHub (pinned to 12126d8942)