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

try
{
    // Start the transaction
    await producer.BeginTransactionAsync();

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

    // Commit - all messages become visible atomically
    await producer.CommitTransactionAsync();
}
catch (Exception ex)
{
    // Abort - none of the messages become visible
    await producer.AbortTransactionAsync();
    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 producer.BeginTransactionAsync();

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

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

        await producer.CommitTransactionAsync();
    }
    catch (Exception ex)
    {
        await producer.AbortTransactionAsync();
        _logger.LogError(ex, "Failed to process message, transaction aborted");
    }
}

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)
.WithIsolationLevel(IsolationLevel.ReadUncommitted)

// Only read committed messages
.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:

try
{
    await producer.BeginTransactionAsync();
    // ... produce messages ...
    await producer.CommitTransactionAsync();
}
catch (ProducerFencedException)
{
    // Another producer with the same transactional ID took over
    // This producer is no longer valid - must recreate
    throw;
}
catch (TransactionAbortedException)
{
    // Transaction was aborted by Kafka (timeout, etc.)
    // Can retry with a new transaction
    await producer.AbortTransactionAsync();
}
catch (Exception ex)
{
    // Other errors - abort and possibly retry
    await producer.AbortTransactionAsync();
    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 producer.BeginTransactionAsync();
await producer.ProduceAsync("topic", key, value);
await producer.CommitTransactionAsync();

// ❌ Bad - long-running transaction
await producer.BeginTransactionAsync();
foreach (var item in millionsOfItems)  // Too many items!
{
    await producer.ProduceAsync("topic", item.Key, item.Value);
}
await producer.CommitTransactionAsync();

2. Unique Transactional IDs

Use instance-specific IDs to avoid fencing:

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

// ❌ Bad - will cause fencing when scaled
var transactionalId = "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
    continue;
}

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 _producer.BeginTransactionAsync();

                var offsets = new List<TopicPartitionOffset>();

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

                await _producer.SendOffsetsToTransactionAsync(
                    _consumer.ConsumerGroupMetadata!,
                    offsets
                );

                await _producer.CommitTransactionAsync();
                _logger.LogInformation("Committed batch of {Count} messages", batch.Count);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Transaction failed, aborting");
                await _producer.AbortTransactionAsync();
            }
        }
    }

    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​