apache/kafka · error · IllegalStateException

Cannot add partition {topicPartition} to transaction before

Error message

Cannot add partition {topicPartition} to transaction before completing a call to initTransactions

What it means

Thrown by TransactionManager.maybeAddPartition (invoked on send) when the producer is transactional but has no valid producerId/epoch yet — i.e. initTransactions() has not completed. Without a producerId the broker cannot attribute the partition to a transaction, so the client blocks the send rather than producing orphan records. The exception leaves the state machine untouched so the caller can retry after init.

Source

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

                            .setTransactionalId(transactionalId)
                            .setProducerId(producerIdAndEpoch.producerId)
                            .setProducerEpoch(producerIdAndEpoch.epoch)
                            .setGroupId(groupMetadata.groupId())
            );
            handler = new AddOffsetsToTxnHandler(builder, offsets, groupMetadata);
        }

        enqueueRequest(handler);
        return handler.result;
    }

    public synchronized void maybeAddPartition(TopicPartition topicPartition) {
        maybeFailWithError();
        throwIfPendingState(TransactionOperation.SEND);

        if (isTransactional()) {
            if (!hasProducerId()) {
                throw new IllegalStateException("Cannot add partition " + topicPartition +
                    " to transaction before completing a call to initTransactions");
            } else if (currentState != State.IN_TRANSACTION) {
                throw new IllegalStateException("Cannot add partition " + topicPartition +
                    " to transaction while in state  " + currentState);
            } else if (isTransactionV2Enabled()) {
                txnPartitionMap.getOrCreate(topicPartition);
                partitionsInTransaction.add(topicPartition);
                transactionStarted = true;
            } else if (transactionContainsPartition(topicPartition) || isPartitionPendingAdd(topicPartition)) {
                return;
            } else {
                log.debug("Begin adding new partition {} to transaction", topicPartition);
                txnPartitionMap.getOrCreate(topicPartition);
                newPartitionsInTransaction.add(topicPartition);
            }
        }
    }

View on GitHub (pinned to c31c9215e1)

Solutions

Call producer.initTransactions() once at startup and block until it completes (it returns a future that must resolve).
Do not call send() until initTransactions() has succeeded; gate sends behind an AtomicBoolean initialised flag.
Call initTransactions() only once per producer lifetime — not per transaction. Subsequent transactions start with beginTransaction().
If the producerId was lost due to an error, re-create the producer and re-run initTransactions() rather than sending on the broken instance.

Example fix

// before
props.put("transactional.id", "tx-1");
KafkaProducer<String,String> p = new KafkaProducer<>(props);
p.send(new ProducerRecord<>("t", "k", "v")); // throws

// after
props.put("transactional.id", "tx-1");
KafkaProducer<String,String> p = new KafkaProducer<>(props);
p.initTransactions();
p.beginTransaction();
p.send(new ProducerRecord<>("t", "k", "v"));
p.commitTransaction();

Defensive patterns

Strategy: validation

Validate before calling

private final java.util.concurrent.atomic.AtomicBoolean initialized = new java.util.concurrent.atomic.AtomicBoolean();
...
producer.initTransactions();           // call exactly once at startup
initialized.set(true);
...
if (!initialized.get()) throw new IllegalStateException("initTransactions must complete before beginTransaction");
producer.beginTransaction();

Try / catch

try {
    producer.beginTransaction();
} catch (IllegalStateException ise) {
    // initTransactions not yet completed; call and await it, then retry beginTransaction
}

Prevention

Call and await producer.initTransactions() exactly once at startup before any transactional operation.
Treat initTransactions failure as fatal; do not proceed to send on an uninitialized producer.
Gate every beginTransaction on a flag set only after initTransactions returns successfully.

When it happens

Trigger: Calling producer.send() on a transactional producer (transactional.id set) before producer.initTransactions() has returned. initTransactions() is asynchronous internally; sending during the INITIALIZING window before the producerId is set also trips this guard via hasProducerId()==false.

Common situations: Forgetting initTransactions() in application bootstrap; calling initTransactions() but not waiting on its future before sending; hot-restart code paths that skip init on a reused producer; library code that wraps send() but not init().

Related errors

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