Transactions

Kafka transactions enable exactly-once semantics (EOS) by allowing you to atomically write to multiple partitions and topics. Either all messages in a transaction are committed, or none are.

When to Use Transactions

Transactions are useful when you need to:

Atomically write to multiple topics - All succeed or all fail
Implement exactly-once processing - Consume, process, produce without duplicates
Maintain consistency - Ensure related messages are visible together

Setting Up a Transactional Producer

Create a producer with a transactional ID:

using Dekaf;

await using var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithTransactionalId("my-service-instance-1")  // Must be unique per instance
    .BuildAsync();

// Initialize transactions (required before first transaction)
await producer.InitTransactionsAsync();

By default, code after await transaction.ProduceAsync(...) resumes inline on the broker sender thread for maximum throughput. That thread also sends, retries, and handles timeouts for other work using the same broker connection, so continuation code must not block or do long-running synchronous work.

If your continuation code cannot guarantee that, isolate it from broker processing:

await using var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithTransactionalId("my-service-instance-1")
    .WithInlineTransactionCompletions(false)
    .BuildAsync();

This schedules transactional produce continuations asynchronously and prevents a blocking continuation from stalling sends, retries, or timeout handling on that broker. It adds one continuation dispatch per transactional produce.

warning

The transactional ID must be unique per producer instance. If two producers use the same ID, one will be fenced (killed) by Kafka.

Basic Transaction Flow

await using var transaction = producer.BeginTransaction();
try
{
    // Send messages within the transaction
    await transaction.ProduceAsync("orders", orderId, orderJson);
    await transaction.ProduceAsync("audit-log", orderId, auditEntry);
    await transaction.ProduceAsync("notifications", userId, notification);

    // Commit - all messages become visible atomically
    await transaction.CommitAsync();
}
catch (Exception ex)
{
    // Abort - none of the messages become visible
    await transaction.AbortAsync();
    throw;
}

Two-Phase Commit Participation

For external transaction coordinators, enable KIP-939 two-phase commit participation:

await using var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithTransactionalId("orders-2pc")
    .WithTwoPhaseCommit()
    .BuildAsync();

await producer.InitTransactionsAsync();

await using var transaction = producer.BeginTransaction();
await transaction.ProduceAsync(new ProducerMessage<string, string>
{
    Topic = "orders",
    Key = orderId,
    Value = orderJson
});

var prepared = await transaction.PrepareAsync();

// Store prepared.ToString() with the external transaction decision.
await producer.CompletePreparedTransactionAsync(prepared, committed: true);

If the process restarts after prepare, initialize with keepPreparedTransaction: true and complete using the stored state plus the external coordinator's decision:

await producer.InitTransactionsAsync(keepPreparedTransaction: true);
var prepared = PreparedTransactionState.Parse(storedPreparedState);
await producer.CompletePreparedTransactionAsync(prepared, committed: shouldCommit);

This requires broker support for transaction.version 3 and InitProducerId v6. After PrepareAsync, only commit, abort, dispose, or CompletePreparedTransactionAsync are valid until the prepared transaction is finished.

Exactly-Once Processing (Consume-Transform-Produce)

The most common use case for transactions is exactly-once stream processing:

using Dekaf;

await using var consumer = await Kafka.CreateConsumer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("processor-group")
    .WithOffsetCommitMode(OffsetCommitMode.Manual)  // We'll commit via transaction
    .WithIsolationLevel(IsolationLevel.ReadCommitted)  // Only read committed messages
    .SubscribeTo("input-topic")
    .BuildAsync();

await using var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithTransactionalId($"processor-{Environment.MachineName}")
    .BuildAsync();

await producer.InitTransactionsAsync();

await foreach (var message in consumer.ConsumeAsync(ct))
{
    try
    {
        await using var transaction = producer.BeginTransaction();

        // Process and produce output
        var result = message.Value.ToUpperInvariant();
        await transaction.ProduceAsync("output-topic", message.Key, result);

        // Commit offsets within the transaction
        await transaction.SendOffsetsToTransactionAsync(
            new[] { new TopicPartitionOffset(message.Topic, message.Partition, message.Offset + 1) },
            consumer.ConsumerGroupMetadata
        );

        await transaction.CommitAsync();
    }
    catch (Exception ex)
    {
        _logger.LogError(ex, "Failed to process message; transaction disposal aborts it");
    }
}

This pattern guarantees:

Each input message is processed exactly once
Output messages are produced exactly once
Consumer offsets are committed atomically with outputs

Transaction Isolation Levels

Consumers can choose whether to read uncommitted messages:

// Read all messages, including uncommitted (default)
var readUncommitted = Kafka.CreateConsumer<string, string>()
    .WithIsolationLevel(IsolationLevel.ReadUncommitted);

// Only read committed messages
var readCommitted = Kafka.CreateConsumer<string, string>()
    .WithIsolationLevel(IsolationLevel.ReadCommitted);

tip

Use ReadCommitted when consuming from topics that receive transactional writes to avoid seeing messages that might be aborted.

Transaction Timeouts

Transactions have a timeout to prevent hanging transactions:

using Dekaf;

var producer = await Kafka.CreateProducer<string, string>()
    .WithBootstrapServers("localhost:9092")
    .WithTransactionalId("my-service")
    .WithTransactionTimeout(TimeSpan.FromMinutes(2))  // Default is 1 minute
    .BuildAsync();

If a transaction isn't committed or aborted within the timeout, Kafka will abort it automatically.

Error Handling

Different errors require different handling:

await using var transaction = producer.BeginTransaction();
try
{
    // ... produce messages ...
    await transaction.CommitAsync();
}
catch (FatalTransactionException)
{
    // Another producer with the same transactional ID took over
    // This producer is no longer valid - must recreate
    throw;
}
catch (AbortableTransactionException)
{
    // Transaction was aborted by Kafka (timeout, etc.)
    // Can retry with a new transaction
    await transaction.AbortAsync();
}
catch (Exception ex)
{
    // Other errors - abort and possibly retry
    await transaction.AbortAsync();
    throw;
}

Best Practices

1. Keep Transactions Short

Long-running transactions:

Increase memory usage on brokers
May timeout
Block consumers using ReadCommitted

// ✅ Good - quick transaction
await using (var shortTransaction = producer.BeginTransaction())
{
    await shortTransaction.ProduceAsync("topic", key, value);
    await shortTransaction.CommitAsync();
}

// ❌ Bad - long-running transaction
await using (var longTransaction = producer.BeginTransaction())
{
    foreach (var item in millionsOfItems)  // Too many items!
    {
        await longTransaction.ProduceAsync("topic", item, item);
    }
    await longTransaction.CommitAsync();
}

2. Unique Transactional IDs

Use instance-specific IDs to avoid fencing:

// ✅ Good - unique per instance
var uniqueTransactionalId = $"order-processor-{Environment.MachineName}-{Guid.NewGuid():N}";

// ❌ Bad - will cause fencing when scaled
var sharedTransactionalId = "order-processor";  // Same ID for all instances!

3. Idempotent Operations

Even with transactions, make your processing idempotent when possible:

// Use deterministic IDs
var outputKey = $"{inputMessage.Topic}-{inputMessage.Partition}-{inputMessage.Offset}";

// Check if already processed (in your database, cache, etc.)
if (await IsAlreadyProcessedAsync(outputKey))
{
    // Skip - this handles edge cases during recovery
    Console.WriteLine("Already processed");
}

Complete Example

public class ExactlyOnceProcessor
{
    private readonly IKafkaConsumer<string, string> _consumer;
    private readonly IKafkaProducer<string, string> _producer;
    private readonly ILogger _logger;

    public async Task RunAsync(CancellationToken ct)
    {
        await _producer.InitTransactionsAsync();

        await foreach (var batch in _consumer.ConsumeAsync(ct).Batch(100))
        {
            try
            {
                await using var transaction = _producer.BeginTransaction();

                var offsets = new List<TopicPartitionOffset>();

                foreach (var msg in batch)
                {
                    var result = Transform(msg.Value);
                    await transaction.ProduceAsync("output", msg.Key, result);
                    offsets.Add(new TopicPartitionOffset(msg.Topic, msg.Partition, msg.Offset + 1));
                }

                await transaction.SendOffsetsToTransactionAsync(
                    offsets,
                    _consumer.ConsumerGroupMetadata
                );

                await transaction.CommitAsync();
                _logger.LogInformation("Committed batch of {Count} messages", batch.Count);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Transaction failed; transaction disposal aborts it");
            }
        }
    }

    private string Transform(string input) => input.ToUpperInvariant();
}

When to Use Transactions​

Setting Up a Transactional Producer​

Basic Transaction Flow​

Two-Phase Commit Participation​

Exactly-Once Processing (Consume-Transform-Produce)​

Transaction Isolation Levels​

Transaction Timeouts​

Error Handling​

Best Practices​

1. Keep Transactions Short​

2. Unique Transactional IDs​

3. Idempotent Operations​

Complete Example​