The Problem: All-or-Nothing Writes in Message Queues

Consider an e-commerce checkout pipeline. When a customer places an order, your backend must publish an event to orders and simultaneously emit a reservation to inventory.

If the network drops mid-flight and orders receives its message but inventory fails, you get an orphaned order with no stock allocated. In single relational databases, an ACID transaction guarantees that either both writes persist or neither does.

In distributed message queues, however, topics and partitions are traditionally sharded across independent consensus groups. Achieving cross-topic atomicity has historically required one of distributed systems' most expensive protocols: Two-Phase Commit (2PC).

The High Tax of Two-Phase Commit

In mature systems like Apache Kafka and Apache Pulsar, cross-partition atomicity relies on a dedicated Transaction Coordinator (TC). When a transactional producer initiates writes across topics, the system performs a multi-step orchestration:

• Register Partitions: The producer calls AddPartitionsToTxn for each topic partition involved, incurring network round-trips before message bodies are even sent.
• Durable State Logging: The coordinator persists transaction phase transitions to a dedicated internal log (e.g., __transaction_state).
• Prepare & Commit Markers: Once the client issues commitTransaction(), the coordinator writes prepare markers, waits for replication, and writes commit control markers to each individual partition log.
• Consumer Isolation Delays: Consumers running with isolation.level=read_committed maintain a per-partition Last Stable Offset (LSO), buffering and stalling messages until all control markers are finalized.

While 2PC provides correctness across arbitrary partitions, it incurs a steep price in latency: multiple round trips, synchronous disk flushes to coordination topics, and complex coordinator failure handling.

Raft Already Commits Atomically

Raft is an understandable consensus algorithm for replicating a log across a cluster. When a Raft leader replicates a command to a majority of followers and receives positive ACKs, the entire log entry is committed.

Here is the key observation behind DRMQ:

Instead of giving each topic its own separate Raft consensus group and using a coordinator to bridge them, DRMQ runs a unified Raft consensus log per cluster node. Multi-topic writes, single produces, and consumer group offset commits all pass through the same Raft log order:

// Client-side multi-topic produce API
Map<String, byte[]> topicMessages = new HashMap<>();
topicMessages.put("orders", orderEventBytes);
topicMessages.put("inventory", inventoryEventBytes);

// Single wire request containing all topic slices
producer.sendAtomic(topicMessages);

There is no transaction identifier, no beginTransaction() round-trip, and no partition registration. The wire request itself encodes the multi-topic transaction.

Leader-Side Cross-Request Aggregation

A naive implementation of single-group consensus would propose one Raft round for every client transaction, creating an immediate throughput bottleneck.

DRMQ solves this via Leader-Side Cross-Request Aggregation. The cluster leader runs a dedicated aggregator thread that continuously pulls waiting atomic requests from a bounded concurrent queue (with a configurable linger timeout) and merges them into a single contiguous Raft log entry:

// Algorithm: Leader-Side Cross-Request Aggregator
while (running) {
    // 1. Poll incoming atomic transactions with short linger timeout
    Request first = queue.poll(LINGER_MS);
    if (first == null) continue;

    List<Request> batch = queue.drainTo(MAX_DRAIN - 1);
    batch.add(0, first);

    // 2. Merge per-topic slices & record client response index positions
    Map<String, List<Entry>> merged = new HashMap<>();
    for (Request req : batch) {
        appendSlices(merged, req);
    }

    // 3. Pre-allocate contiguous monotonic global offsets
    long baseOffset = reserveOffsets(totalEntries);

    // 4. Propose ONE single Raft entry for dozens of client transactions
    RaftEntry entry = new RaftEntry(OpType.ATOMIC_BATCH, merged, baseOffset);
    proposeToRaft(entry);
}

Because contiguous global offsets are reserved upfront and embedded directly into the Raft entry, every replica deterministically computes identical per-topic offsets when applying the committed log.

Crash Safety: The Atomic-Intent Write-Ahead Log

Committing a decision in Raft guarantees that the cluster agreed on the transaction. But each individual broker still faces a local filesystem challenge: applying that decision requires persisting writes across multiple independent, append-only topic segment files (e.g., orders/00000.log and inventory/00000.log).

Modifying multiple files on a filesystem is not an atomic operation. If a server loses power halfway through writing to orders before it finishes inventory, the local broker storage would be left in an inconsistent, partially applied state upon reboot.

To make this step crash-safe without introducing a bloated, general-purpose transaction manager, DRMQ uses a lightweight Atomic-Intent Write-Ahead Log (.atomic-intent):

// Algorithm 2: Atomic Apply and Local Crash Recovery
procedure applyAtomicBatch(entry) {
    // 1. Synchronously persist intent to disk before touching topic files
    write(".atomic-intent", entry.slices, entry.baseOffsets);
    fsync(".atomic-intent");

    // 2. Acquire topic locks in sorted order to guarantee deadlock freedom
    topics = sort(distinctTopics(entry.slices));
    for (topic in topics) { acquire(lock[topic]); }

    // 3. Append to independent append-only segment files
    for ((topic, messages) in entry.slices) {
        seg = activeSegment(topic);
        if (seg.isFull()) seg = rollNewSegment(topic);
        seg.append(messages);
        updateIndexAndCache(topic, messages);
    }
    for (topic in topics) { release(lock[topic]); }

    // 4. Once all per-topic files are durable, clear the intent file
    delete(".atomic-intent");
}

// Crash recovery executed once on broker startup before serving traffic
procedure recover() {
    if (!exists(".atomic-intent")) return;

    // Replay pending writes from the intent WAL
    intent = read(".atomic-intent");
    for ((topic, messages) in intent.slices) {
        head = currentHeadOffset(topic);
        pending = { m in messages : m.offset > head };
        if (pending != empty) appendToSegment(topic, pending);
    }
    delete(".atomic-intent");
}

Because offsets are pre-allocated deterministically during Raft entry creation, recovery is always an idempotent redo (never an undo). Any message whose offset was already written to disk before the crash is cleanly skipped. The broker finishes applying remaining messages to all topics before opening its port to traffic — guaranteeing that local broker failure never creates an inconsistent cross-topic view.

Benchmark Evaluation: DRMQ vs Apache Kafka

To evaluate the architecture, we benchmarked DRMQ directly against Apache Kafka’s transactional producer under identical cluster conditions (3-node cluster, Replication Factor = 3, acks=all / Raft quorum).

Metric / Percentile	Apache Kafka (2PC)	DRMQ (Unified Raft)	Observed Improvement
Commit Latency (p50)	49 ms	13 ms	73.5% faster
Commit Latency (p95)	77 ms	23 ms	70.1% faster
Commit Latency (p99)	88 ms	40 ms	54.5% faster
Tail Latency (p999)	116 ms	65 ms	44.0% lower tail
Serial Producer Throughput	19 – 22 TPS	68 – 80 TPS	~3.5× throughput
20 Concurrent Producers	253 TPS	1,339 TPS	5.3× throughput
Client Batching (1MB Payload)	—	3,247 TPS	High concurrency scale

Engineering Tradeoffs & Limitations

No architectural decision in distributed systems comes without trade-offs. DRMQ is not intended as a universal drop-in replacement for Apache Kafka’s entire ecosystem. Rather, it explores an alternative optimization point:

• Single Leader Bottleneck: In DRMQ, all writes pass through a single Raft leader node. Kafka avoids this by distributing partition leadership across all brokers. For massive multi-terabyte non-transactional streaming, Kafka scales horizontally across more machines.
• No Application-Level Abort: In DRMQ, once an atomic batch passes acceptance checks and enters the Raft consensus pipeline, it commits upon majority replication. It does not currently support an interactive abortTransaction() halfway through.
• Visibility Synchronization: Message publication visibility across different topics in a batch occurs as the apply loop executes. In rare corner cases, a fast consumer could observe messages in Topic A a few microseconds before Topic B is marked visible.

Conclusion & Source Code

By rethinking where transaction coordination belongs, DRMQ demonstrates that cross-topic atomicity does not inevitably require the latency overhead of Two-Phase Commit coordinators. When multi-topic writes are embraced as first-class citizens in a unified Raft log, consensus and atomicity become one and the same.

The complete Java implementation, client SDK, and reproducible benchmark harness are open source:

→ github.com/samuel025/DRMQ