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:

  • 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