Transactions

Redpanda supports Apache Kafka®-compatible transaction semantics and APIs. For example, you can fetch messages starting from the last consumed offset and transactionally process them one by one, updating the last consumed offset and producing events at the same time.

A transaction can span partitions from different topics, and a topic can be deleted while there are active transactions on one or more of its partitions. In-flight transactions can detect deletion events, remove the deleted partitions (and related messages) from the transaction scope, and commit changes to the remaining partitions.

If a producer is sending multiple messages to the same or different partitions, and network connectivity or broker failure cause the transaction to fail, then it’s guaranteed that either all messages are written to the partitions or none. This is important for applications that require strict guarantees, like financial services transactions.

Transactions guarantee both exactly-once semantics (EOS) and atomicity:

  • EOS helps developers avoid the anomalies of at-most-once processing (with potential lost events) and at-least-once processing (with potential duplicated events). Redpanda supports EOS when transactions are used in combination with idempotent producers.

  • Atomicity additionally commits a set of messages across partitions as a unit: either all messages are committed or none. Encapsulated data received or sent across multiple topics in a single operation can only succeed or fail globally.

Use transactions

By default, the enable_transactions cluster configuration property is set to true. However, in the following use cases, clients must explicitly use the Transactions API to perform operations within a transaction:

When you use transactions, you must set the transactional.id property in the producer configuration. This property uniquely identifies the producer and enables reliable semantics across multiple producer sessions. It ensures that all transactions issued by a given producer are completed before any new transactions are started.

Atomic publishing of multiple messages

A banking IT system with an event-sourcing microservice architecture illustrates why transactions are necessary. In this system, each bank branch is implemented as an independent microservice that manages its own distinct set of accounts. Every branch maintains its own transaction history, stored as a Redpanda partition. When a branch starts, it replays the transaction history to reconstruct its current state.

Financial transactions such as money transfers require the following guarantees:

  • A sender can’t withdraw more than the account withdrawal limit.

  • A recipient receives exactly the same amount sent.

  • A transaction is fast and is run at most once.

  • If a transaction fails, the system rolls back to the initial state.

  • Without withdrawals and deposits, the amount of money in the system remains constant with any history of money transfers.

These requirements are easy to satisfy when the sender and the recipient of a financial transaction are hosted by the same branch. The operation doesn’t leave the consistency domain, and all checks and locks can be performed within a single service (ledger).

Things get more complex with cross-branch financial transactions, because they involve several ledgers, and the operations should be performed atomically (all or nothing). The default approach (saga pattern) breaks a transaction into a sequence of reversible idempotent steps; however, this violates the isolation principle and adds complexity, making the application responsible for orchestrating the steps.

Redpanda natively supports transactions, so it’s possible to atomically update several ledgers at the same time. For example:

Show multi-ledger transaction example: 
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "...");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "app-id");

Producer<String, String> producer = null;

while (true) {
    // waiting for somebody to initiate a financial transaction
    var sender_branch = ...;
    var sender_account = ...;
    var recipient_branch = ...;
    var recipient_account = ...;
    var amount = 42;

    if (producer == null) {
        try {
            producer = new KafkaProducer<>(props);
            producer.initTransactions();
        } catch (Exception e1) {
            // TIP: log error for further analysis
            try {
                if (producer != null) {
                    producer.close();
                }
            } catch(Exception e2) { }
            producer = null;
            // TIP: notify the initiator of a transaction about the failure
            continue;
        }
    }

    producer.beginTransaction();
    try {
        var f1 = producer.send(new ProducerRecord<String, String>("ledger", sender_branch, sender_account, "" + (-amount)));
        var f2 = producer.send(new ProducerRecord<String, String>("ledger", recipient_branch, recipient_account, "" + amount));
        f1.get();
        f2.get();
    } catch (Exception e1) {
        // TIP: log error for further analysis
        try {
            producer.abortTransaction();
        } catch (Exception e2) {
            // TIP: log error for further analysis
            try {
                producer.close();
            } catch (Exception e3) { }
            producer = null;
        }
        // TIP: notify the initiator of a transaction about the failure
        continue;
    }

    try {
        producer.commitTransaction();
    } catch (Exception e1) {
        try {
            producer.close();
        } catch (Exception e3) {}
        producer = null;
        // TIP: notify the initiator of a transaction about the failure
        continue;
    }

    // TIP: notify the initiator of a transaction about the success
}

When a transaction fails before a commitTransaction attempt completes, you can assume that it is not executed. When a transaction fails after a commitTransaction attempt completes, the true transaction status is unknown. Redpanda only guarantees that there isn’t a partial result: either the transaction is committed and complete, or it is fully rolled back.

Exactly-once stream processing

Redpanda is commonly used as a pipe connecting different applications and storage systems. An application could use an OLTP database and then rely on change data capture to deliver the changes to a data warehouse.

Redpanda transactions let you use streams as a smart pipe in your applications, building complex atomic operations that transform, aggregate, or otherwise process data transiting between external applications and storage systems.

For example, here is the regular pipe flow:

Postgresql -> topic -> warehouse

Here is the smart pipe flow, with a transformation in topic(1) -> topic(2):

Postgresql -> topic(1) transform topic(2) -> warehouse

The transformation reads a record from topic(1), processes it, and writes it to topic(2). Without transactions, an intermittent error can cause a message to be lost or processed several times. With transactions, Redpanda guarantees exactly-once semantics. For example:

Show exactly-once processing example: 
var source = "source-topic";
var target = "target-topic";

Properties pprops = new Properties();
pprops.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "...");
pprops.put(ProducerConfig.ACKS_CONFIG, "all");
pprops.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
pprops.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, UUID.randomUUID().toString());

Properties cprops = new Properties();
cprops.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "...");
cprops.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
cprops.put(ConsumerConfig.GROUP_ID_CONFIG, "app-id");
cprops.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
cprops.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

Consumer<String, String> consumer = null;
Producer<String, String> producer = null;

boolean should_reset = false;

while (true) {
    if (should_reset) {
        should_reset = false;

        if (consumer != null) {
            try {
                consumer.close();
            } catch(Exception e) {}
            consumer = null;
        }

        if (producer != null) {
            try {
                producer.close();
            } catch (Exception e2) {}
            producer = null;
        }
    }

    try {
        if (consumer == null) {
            consumer = new KafkaConsumer<>(cprops);
            consumer.subscribe(Collections.singleton(source));
        }
    } catch (Exception e1) {
        // TIP: log error for further analysis
        should_reset = true;
        continue;
    }

    try {
        if (producer == null) {
            producer = new KafkaProducer<>(pprops);
            producer.initTransactions();
        }
    } catch (Exception e1) {
        // TIP: log error for further analysis
        should_reset = true;
        continue;
    }

    ConsumerRecords<String, String> records = null;
    try {
        records = consumer.poll(Duration.ofMillis(10000));
    } catch (Exception e1) {
        // TIP: log error for further analysis
        should_reset = true;
        continue;
    }

    var it = records.iterator();
    while (it.hasNext()) {
        var record = it.next();

        // transformation
        var old_value = record.value();
        var new_value = old_value.toUpperCase();

        try {
            producer.beginTransaction();
            producer.send(new ProducerRecord<String, String>(target, record.key(), new_value));
            var offsets = new HashMap<TopicPartition, OffsetAndMetadata>();
            offsets.put(new TopicPartition(source, record.partition()), new OffsetAndMetadata(record.offset() + 1));
            producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
        } catch (Exception e1) {
            // TIP: log error for further analysis
            try {
                producer.abortTransaction();
            } catch (Exception e2) { }
            should_reset = true;
            break;
        }

        try {
            producer.commitTransaction();
        } catch (Exception e1) {
            // TIP: log error for further analysis
            should_reset = true;
            break;
        }
    }
}

Exactly-once processing configuration requirements

Redpanda’s default configuration supports exactly-once processing. To preserve this capability, ensure the following settings are maintained:

  • enable_idempotence = true

  • enable_transactions = true

  • transaction_coordinator_delete_retention_ms is greater than or equal to transactional_id_expiration_ms

Best practices

To help avoid common pitfalls and optimize performance, consider the following when configuring transactional workloads in Redpanda:

Tune producer ID limits

For production environments with heavy producer usage, configure both max_concurrent_producer_ids and transactional_id_expiration_ms to prevent out-of-memory (OOM) crashes. Setting limits on producer IDs helps manage memory usage in high-throughput environments, particularly when using transactions or idempotent producers.

If you have`kafka_connections_max` configured, you can determine an appropriate value for max_concurrent_producer_ids based on your connection patterns.

  • Lower bound: kafka_connections_max / number_of_shards, assuming each producer connects to only one shard.

  • Upper bound: topic_partitions_per_shard * kafka_connections_max, assuming producers connect to all shards.

If kafka_connections_max is not configured, estimate the value for max_concurrent_producer_ids based on your application patterns. A conservative approach is to start with 1000-5000 per shard, then monitor and adjust as needed. Applications with many partitions per producer typically require higher values, such as 10000 or more per shard.

Tune transactional_id_expiration_ms based on your application’s transaction patterns. Calculate this value by taking your longest expected transaction time and adding a safety buffer. For example, if transactions typically run for 30 minutes, consider setting this to 2-4 hours. Short-lived transactions can use values between 1-4 hours, while batch processing applications should match their batch interval plus buffer time. Interactive applications may benefit from shorter values to free up memory faster.

Client applications should minimize producer ID churn. Reuse producer instances when possible, instead of creating new ones for each operation. Avoid using random transactional IDs, as some Flink configurations do, because this creates excessive producer ID churn. Instead, use consistent transactional IDs that can be resumed across application restarts.

Configure transaction timeouts and limits

  • If a consumer is configured to use the read_committed isolation level, it can only process successfully committed transactions. As a result, an ongoing transaction with a large timeout that becomes stuck could prevent the consumer from processing other committed transactions.

    To avoid this, don’t set the transaction timeout client setting (transaction.timeout.ms in the Kafka Java client implementation) to a value that is too high. The longer the timeout, the longer consumers may be blocked.

Handle transaction failures

Different transactions require different approaches to handling failures within the application. Consider the approaches to failed or timed-out transactions in the provided use cases:

  • Publishing of multiple messages: The request came from outside the system, and it is the application’s responsibility to discover the true status of a timed-out transaction. (This example doesn’t use consumer groups to distribute partitions between consumers.)

  • Exactly-once streaming (consume-transform-loop): This is a closed system. Upon re-initialization of the consumer and producer, the system automatically discovers the moment it was interrupted and continues from that place. Additionally, this automatically scales by the number of partitions. Run another instance of the application, and it starts processing its share of partitions in the source topic.

Transactions with compacted segments

Transactions are supported on topics with compaction configured. The compaction process removes the aborted transaction’s data and all transactional control markers from the log. The resulting compacted segment contains only committed data batches (and potentially harmless gaps in the offsets due to skipped batches).

Suggested reading