apache/kafka · critical · InvalidProducerEpochException

Producer with transactionalId '{transactionalId}' and {produ

Error message

Producer with transactionalId '{transactionalId}' and {producerIdAndEpoch} attempted to produce with an old epoch

What it means

Thrown by TransactionManager.maybeFailWithError when the cached lastError is an InvalidProducerEpochException, meaning this producer tried to send or commit using a producer epoch older than the one the transaction coordinator currently has on record. The coordinator bumps/fences epochs when another instance of the same transactionalId registers (via InitProducerId) or after a transaction timeout, so the broker rejects the batch with InvalidProducerEpoch and the producer stores it as a fatal error. Once set, every subsequent transactional call rethrows this exception via maybeFailWithError. It almost always indicates the producer instance is stale and must be closed and recreated.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:1192

    private void ensureTransactional() {
        if (!isTransactional())
            throw new IllegalStateException("Transactional method invoked on a non-transactional producer.");
    }

    private void maybeFailWithError() {
        if (!hasError()) {
            return;
        }
        // for ProducerFencedException, do not wrap it as a KafkaException
        // but create a new instance without the call trace since it was not thrown because of the current call
        if (lastError instanceof ProducerFencedException) {
            throw new ProducerFencedException("Producer with transactionalId '" + transactionalId
                    + "' and " + producerIdAndEpoch + " has been fenced by another producer " +
                    "with the same transactionalId");
        }
        if (lastError instanceof InvalidProducerEpochException) {
            throw new InvalidProducerEpochException("Producer with transactionalId '" + transactionalId
                    + "' and " + producerIdAndEpoch + " attempted to produce with an old epoch");
        }
        if (lastError instanceof IllegalStateException) {
            throw new IllegalStateException("Producer with transactionalId '" + transactionalId
                    + "' and " + producerIdAndEpoch + " cannot execute transactional method because of previous invalid state transition attempt", lastError);
        }
        throw new KafkaException("Cannot execute transactional method because we are in an error state", lastError);
    }

    private boolean maybeTerminateRequestWithError(TxnRequestHandler requestHandler) {
        if (hasError()) {
            if (hasAbortableError() && requestHandler instanceof FindCoordinatorHandler)
                // No harm letting the FindCoordinator request go through if we're expecting to abort
                return false;

            requestHandler.fail(lastError);
            return true;
        }

View on GitHub (pinned to c31c9215e1)

Solutions

1. If multiple instances share this transactionalId, ensure only ONE producer instance owns it at a time (unique transactional.id per logical instance, or use an instance-identity scheme). The fenced instance must close() and a new producer must call initTransactions().
2. Increase transaction.timeout.ms on the producer (and broker transaction.max.timeout.ms) so legitimate long transactions don't get fenced by the coordinator.
3. Never reuse a producer after catching ProducerFencedException/InvalidProducerEpochException — wrap usage in try-with-resources or close() and instantiate a fresh KafkaProducer.
4. Verify no zombie process from a previous deployment is still heartbeating/producing with the same transactional.id (kill stale pods/jobs).

Example fix

// before
try {
    producer.beginTransaction();
    producer.send(record);
    producer.commitTransaction();
} catch (ProducerFencedException | InvalidProducerEpochException e) {
    // reused the same producer
    producer.beginTransaction();  // throws InvalidProducerEpochException again
}

// after
try (var p = new KafkaProducer<>(props)) {
    p.initTransactions();
    p.beginTransaction();
    p.send(record);
    p.commitTransaction();
} catch (ProducerFencedException | InvalidProducerEpochException e) {
    // producer is unusable; closing via try-with-resources is correct, next run creates a new one
    throw e;
}

// Cannot reliably pre-validate: the broker invalidates epochs asynchronously.
// Ensure only one Producer instance per transactionalId is alive:
ProducerConfig cfg = new ProducerConfig(props);
String txnalId = (String) props.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
// Use an external lease (e.g. lock in coordinator/zk) keyed by transactionalId
// before constructing KafkaProducer so no two producers share it.

public static boolean isProducerEpochValid(org.apache.kafka.clients.producer.internals.TransactionManager tm) {
    // Reflective read only; the client does not expose producerId/epoch publicly.
    try {
        java.lang.reflect.Field f = tm.getClass().getDeclaredField("producerIdAndEpoch");
        f.setAccessible(true);
        Object epochHolder = f.get(tm);
        java.lang.reflect.Field e = epochHolder.getClass().getDeclaredField("epoch");
        e.setAccessible(true);
        return ((short) e.get(epochHolder)) >= 0;
    } catch (Exception ex) { return false; }
}

try {
    producer.send(record).get();
} catch (org.apache.kafka.common.errors.InvalidProducerEpochException e) {
    // This producer's epoch is stale: another producer with the same transactionalId
    // took over. Close this instance and surface the failure (do NOT retry on this object).
    producer.close(Duration.ZERO);
    throw new IllegalStateException("Producer fenced by a newer epoch; abort processing for "
        + "transactionalId and restart on a fresh producer", e);
} catch (org.apache.kafka.common.errors.ProducerFencedException e) {
    producer.close(Duration.ZERO);
    throw e;
}

When it happens

Trigger: Producing (KafkaProducer.send / flush), commitTransaction, abortTransaction, or sendOffsetsToTransaction on a transactional producer whose epoch has been superseded. Specifically: (a) two process instances share the same transactional.id and the newer one called initTransactions; (b) the transaction coordinator fenced this producer after transaction.timeout.ms expired and it was re-initialized; (c) the app caught the error once and reused the same producer instance for the next transaction without close()+new KafkaProducer.

Common situations: Deployment/rollback where both old and new pods run with the same transactional.id during a rolling restart; consumer-producer fan-out pipeline where a rebalance resurrects a 'dead' instance; long-running batch job whose transaction exceeds transaction.timeout.ms; client/cluster version mismatch where EOS semantics changed; accidentally hardcoding transactional.id per environment instead of using stable per-partition instance IDs.

Related errors

• MockProducer is fenced.
• Cannot set a transactional.id without also enabling idempote
• Cannot send offsets if a transaction is not in progress (cur
• Cannot add partition {topicPartition} to transaction before
• Cannot add partition {topicPartition} to transaction while i

AI-assisted analysis of apache/kafka@c31c9215e1 (2026-08-03). Data as JSON: /data/errors/77beeb3aaf8462e7.json. Report an issue: GitHub.