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: Post-Write Completion Markers

Committing a decision in Raft guarantees that the cluster reached consensus 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 disk is not an atomic operation. If a broker loses power halfway through writing to orders before finishing inventory, its local storage would be left in an inconsistent, partially applied state upon reboot.

To solve this problem, an atomic intent file was initially introduced. When a cross-topic message has been persisted to the Raft log, before it is written to each topic's append-only segment, it is first written to the intent file. If there is a crash and the broker restarts, it checks the intent file to replay and complete any interrupted writes. However, this redundant pre-write log introduces unnecessary disk overhead. DRMQ recognizes that the committed Raft log entry itself is already the authoritative write-ahead log. To guarantee local crash consistency, DRMQ simply uses a lightweight post-write completion marker:

// Algorithm: Atomic Apply and Post-Write Marker Crash Recovery
procedure applyAtomicBatch(entry) {
    // 1. Deterministic lock ordering prevents deadlocks between threads
    topics = sort(distinctTopics(entry.slices));
    for (topic in topics) { acquire(lock[topic]); }

    try {
        // 2. Append to independent per-topic append-only segment files
        for ((topic, messages) in entry.slices) {
            seg = activeSegment(topic);
            if (seg.isFull()) seg = rollNewSegment(topic);
            seg.append(messages, entry.raftIndex); // tagged with raftIndex
            indexMessage(topic, messages);
        }
    } finally {
        for (topic in topics) { release(lock[topic]); }
    }

    // 3. Fsync empty completion marker to confirm all topics were written
    write(".atomic-done-" + entry.raftIndex);
    fsync(dataDirectory);

    // 4. Two-phase visibility: make offsets visible to consumers only now
    advanceTopicHeadOffsets(entry.slices);
    cleanupMarkers(priorTo: entry.raftIndex);
}

// Executed once on broker startup before serving client traffic
procedure recover() {
    lastIndex = readLastAppliedRaftIndex();
    if (lastIndex < 0) return;

    // Marker present: all topic segments were durably persisted
    if (exists(".atomic-done-" + lastIndex)) return;

    // Marker missing: crash occurred mid-apply. Truncate partial writes
    // so the Raft state machine can deterministically re-apply the full batch.
    for (topic in allTopics()) {
        truncatePartialApply(topic, atRaftIndex: lastIndex);
    }
    decrementLastAppliedRaftIndex();
    cleanupAllMarkers();
}

When a broker restarts, it inspects its last applied Raft index. If .atomic-done-{lastIndex} exists, the transaction completed cleanly. If the marker is missing, the broker knows an unclean shutdown interrupted an atomic write mid-flight. Rather than guessing or repairing in place, it truncates partial writes tagged with that Raft index from affected segments, decrements its applied index counter, and lets Raft's authoritative consensus log cleanly re-apply the full batch.

This design yields two major architectural advantages:

• Zero redundant write amplification: No secondary pre-write intent log is ever flushed before appending messages. The Raft log remains the single source of truth.
• Read isolation without coordination: Consumers only observe messages once the completion marker is flushed and head offsets advance — preventing clients from ever reading partial cross-topic transactions.

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)	55 ms	8 ms	85.5% faster
Commit Latency (p95)	93 ms	11 ms	88.2% faster
Commit Latency (p99)	129 ms	16 ms	87.6% faster
Tail Latency (p999)	198 ms	23 ms	88.4% lower tail
Serial Producer Throughput	19 TPS	127 TPS	~6.6× throughput
10 Concurrent Producers	181 TPS	1,621 TPS	~8.9× throughput
Client Batching (1MB Payload)	—	4,878 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

DRMQ shows that getting atomic writes across multiple topics doesn't have to mean paying the heavy latency cost of Two-Phase Commit. By putting multi-topic writes directly into a single unified Raft log, the standard consensus protocol just handles the atomicity for you.

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

→ github.com/samuel025/DRMQ