Apache Ratis: Building Fault-Tolerant Distributed Systems with Raft Consensus

Introduction

In the world of distributed systems, achieving consensus across multiple nodes is one of the most challenging problems engineers face. Apache Ratis emerges as a powerful solution to this challenge, providing a robust, production-ready implementation of the Raft consensus algorithm in Java. Whether you’re building a distributed database, object storage system, or any application requiring strong consistency guarantees, Apache Ratis offers the foundation you need.

What is Apache Ratis?

Apache Ratis is an open source Java implementation of the RAFT consensus protocol. Raft is a consensus algorithm designed to be easy to understand, making it significantly more approachable than older consensus protocols like Paxos while maintaining the same guarantees of correctness and performance.

Apache Ratis graduated to become a top-level Apache project in February 2021, after being initially started in 2017. The project released its first GA version 1.0.0 in July 2020, followed by version 2.0.0 in March 2021, version 3.0.0 in November 2023, and the latest version 3.0.1 in January 2024.

Understanding Raft Consensus

Before diving into Apache Ratis, let’s understand how Raft achieves consensus:

┌──────────────────────────────────────────────────────────┐
│                    Raft Cluster                          │
│                                                          │
│  ┌─────────────┐         ┌─────────────┐                 │
│  │   Client    │────────▶│   Leader    │                 │
│  │             │         │   (Node 1)  │                 │
│  └─────────────┘         └──────┬──────┘                 │
│                                  │                       │
│                    ┌─────────────┼─────────────┐         │
│                    │             │             │         │
│                    ▼             ▼             ▼         │
│            ┌──────────┐   ┌──────────┐  ┌──────────┐     │
│            │ Follower │   │ Follower │  │ Follower │     │
│            │ (Node 2) │   │ (Node 3) │  │ (Node 4) │     │
│            └──────────┘   └──────────┘  └──────────┘     │
│                                                          │
│  Flow: Client → Leader → Log Replication → Followers     │
│        Leader waits for majority (3/4) → Commits         │
└──────────────────────────────────────────────────────────┘

Key Raft Mechanisms:

• Leader Election: One node is elected as the leader; followers replicate its log
• Log Replication: All operations are written to a replicated log before execution
• Consensus: A majority of nodes must agree before any operation is committed
• Strong Consistency: Once committed, data is guaranteed to be durable across the cluster

Core Features of Apache Ratis

Apache Ratis stands out for its pluggable architecture and high performance. Here’s a summary of its key components:

Let’s explore each feature in detail:

1. Pluggable Transport Layer

Ratis provides a pluggable transport layer with gRPC, Netty+Protobuf, and Apache Hadoop RPC based transports provided by default. This flexibility allows you to choose the communication protocol that best fits your infrastructure and performance requirements.

2. Pluggable State Machine

Ratis supports a Raft log and a state machine; the latter typically contains the data you want to make highly available. This design makes it straightforward to integrate your own business logic while Ratis handles the complex consensus coordination.

3. Pluggable Raft Log

The RAFT log is also pluggable, allowing users to provide their own log implementation while the default implementation stores log in local files. Applications can define policies for how and where data should be written.

4. Pluggable Metrics

Ratis provides a pluggable observation layer with default implementation using shaded Dropwizard 4, and users can provide their own metrics implementations. This enables comprehensive monitoring and observability tailored to your needs.

5. Leader Election and Log Replication

At its core, Ratis implements the fundamental Raft mechanisms for leader election and log replication, ensuring that your distributed system maintains consistency even in the face of node failures.

Real-World Use Cases

Apache Ratis has been adopted by several major Apache projects and enterprises for mission-critical production workloads. Let’s explore where and how it’s being used.

1. Apache Hadoop Ozone — Distributed Object Storage

Apache Ozone, a distributed object store, uses Ratis for both SCM/OM high availability and its write pipeline. Ozone metadata is persisted into a pluggable metadata store (RockDB by default) and replicated consistently across all instances using Apache Ratis when High Availability is configured.

Why Ozone Uses Ratis:

• When a client writes an object to Ozone, the object is automatically replicated to three datanodes
• Ensures strong consistency for metadata operations
• Provides fault tolerance for the Storage Container Manager (SCM) and Ozone Manager (OM)
• Enables seamless failover without data loss

Production Impact: Tencent Big Data platform has deployed an Ozone cluster with over 1000 nodes as backend storage for big data applications, utilizing Ozone as a primary storage solution for private data warehouse projects. This demonstrates Ratis operating at massive scale in production environments.

2. Alluxio — Data Orchestration Platform

Alluxio, a data orchestration platform for the cloud originally called Tachyon, uses Ratis for managing all of Alluxio’s journaled state. This ensures that critical metadata about data location, access patterns, and system state remains consistent across the distributed system.

Key Benefits:

• Maintains consistency of the journal that tracks all metadata operations
• Enables high availability for the Alluxio master nodes
• Provides recovery mechanisms for system failures

3. Apache IoTDB — Time Series Database

Apache IoTDB, a database for Internet of Things, uses Ratis for replicating application data. IoTDB handles massive volumes of time-series data from sensors and devices, requiring robust replication and consistency guarantees.

Use Case Highlights:

• Replicates time-series data across multiple nodes
• Ensures data durability for IoT sensor readings
• Maintains consistency for queries across distributed nodes

4. Custom Distributed Systems

Beyond these well-known projects, Ratis is ideal for building custom distributed systems such as:

Distributed Configuration Management:

• Storing cluster configuration that must be consistent across all nodes
• Managing service discovery information
• Coordinating distributed locks and leader election

Distributed Databases:

• Implementing consensus for transaction logs
• Coordinating multi-master database systems
• Building strongly consistent key-value stores

Distributed Coordination Services:

• Building ZooKeeper-like coordination services
• Implementing distributed queues
• Creating leader election mechanisms for microservices

Log-Based Systems: Ratis provides a log service recipe with StateMachines to implement a distributed log service with a focused client API. This is perfect for building systems similar to Apache Kafka or distributed append-only logs.

When Should You Use Apache Ratis?

Apache Ratis is an excellent choice when you need:

Strong Consistency Guarantees: When your application cannot tolerate inconsistencies between replicas, Ratis provides linearizable consistency through the Raft protocol.

Fault Tolerance: If your system must continue operating despite node failures, Ratis handles leader election and failover automatically.

High Availability: Ratis has become a cornerstone technology for consensus in modern distributed systems, enabling systems to remain available even when some nodes fail.

Metadata Replication: When you need to replicate critical metadata (like database schemas, cluster configurations, or service registries) with strong consistency.

Write-Heavy Workloads: While maintaining consistency, organizations like Tencent have worked on optimizations like Multi-Raft to boost write performance for applications.

Architecture Considerations

When implementing Ratis in your system, consider these architectural points:

Cluster Sizing: Raft requires a minimum of 3 nodes to tolerate one failure (N=2F+1, where F is the number of failures to tolerate). For production systems, 5 or 7 nodes are common.

Network Requirements: Since consensus requires network communication between nodes, ensure low-latency, reliable network connectivity between Ratis cluster members.

Storage Performance: The Raft log is typically stored on disk, so fast storage (SSDs) can significantly improve performance, especially for write-heavy workloads.

State Machine Design: Your application state machine should be deterministic — given the same sequence of commands, it should always produce the same result.

Log Compaction: For long-lived clusters, implement periodic snapshots and log compaction to prevent the Raft log from growing indefinitely. Ratis supports snapshotting to capture the state machine’s current state and truncate old log entries.

Performance Characteristics

Apache Ratis has been optimized for production use with several performance enhancements:

• Batching: Multiple client requests can be batched into a single Raft log entry
• Pipelining: Requests can be pipelined to reduce latency
• Async Replication: Log replication to followers can happen asynchronously
• Streaming Support: Ozone’s write pipeline V2 implements Ratis Streaming for improved performance

Getting Started with Apache Ratis

To start using Apache Ratis in your project:

1. Add Dependencies: Include the Ratis libraries in your Maven or Gradle build
2. Implement State Machine: Create your application-specific state machine that extends Ratis’s base state machine
3. Configure Cluster: Define your Ratis cluster peers and configuration
4. Initialize Raft Server: Start Ratis servers on each node
5. Submit Requests: Use the Ratis client to submit operations to the cluster

Real-Time Example: Building a Distributed Counter

Let’s walk through a practical example of building a simple distributed counter service using Apache Ratis. This counter will be replicated across multiple nodes and remain consistent even if nodes fail.

Step 1: Add Maven Dependencies

<dependency>
    <groupId>org.apache.ratis</groupId>
    <artifactId>ratis-server</artifactId>
    <version>3.0.1</version>
</dependency>
<dependency>
    <groupId>org.apache.ratis</groupId>
    <artifactId>ratis-netty</artifactId>
    <version>3.0.1</version>
</dependency>
<dependency>
    <groupId>org.apache.ratis</groupId>
    <artifactId>ratis-grpc</artifactId>
    <version>3.0.1</version>
</dependency>

Step 2: Implement the State Machine

import org.apache.ratis.proto.RaftProtos;
import org.apache.ratis.protocol.Message;
import org.apache.ratis.statemachine.TransactionContext;
import org.apache.ratis.statemachine.impl.BaseStateMachine;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;

public class CounterStateMachine extends BaseStateMachine {

    private final AtomicLong counter = new AtomicLong(0);

    @Override
    public CompletableFuture<Message> applyTransaction(TransactionContext trx) {
        final RaftProtos.LogEntryProto entry = trx.getLogEntry();
        final String command = entry.getStateMachineLogEntry()
            .getLogData()
            .toString(StandardCharsets.UTF_8);

        long result;
        switch (command) {
            case "INCREMENT":
                result = counter.incrementAndGet();
                break;
            case "DECREMENT":
                result = counter.decrementAndGet();
                break;
            case "GET":
                result = counter.get();
                break;
            default:
                result = counter.get();
        }

        final String response = String.valueOf(result);
        return CompletableFuture.completedFuture(
            Message.valueOf(response)
        );
    }
}

Step 3: Create the Ratis Server

import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.grpc.GrpcConfigKeys;
import org.apache.ratis.protocol.RaftGroup;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftPeer;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.RaftServerConfigKeys;
import org.apache.ratis.util.NetUtils;

import java.io.File;
import java.util.Collections;
import java.util.UUID;

public class CounterServer {

    public static void main(String[] args) throws Exception {

        // Define server ID and address
        String serverId = "n1";
        String serverAddress = "127.0.0.1:6000";

        // Create Raft peer
        RaftPeerId peerId = RaftPeerId.valueOf(serverId);
        RaftPeer peer = RaftPeer.newBuilder()
            .setId(peerId)
            .setAddress(serverAddress)
            .build();

        // Create Raft group (single peer for simplicity, use 3+ in production)
        RaftGroupId groupId = RaftGroupId.valueOf(
            UUID.fromString("02511d47-d67c-49a3-9011-abb3109a44c1")
        );
        RaftGroup raftGroup = RaftGroup.valueOf(
            groupId,
            Collections.singletonList(peer)
        );

        // Configure Raft properties
        RaftProperties properties = new RaftProperties();

        // Set storage directory
        File storageDir = new File("./raft-data/" + serverId);
        RaftServerConfigKeys.setStorageDir(properties, 
            Collections.singletonList(storageDir));

        // Configure gRPC port
        GrpcConfigKeys.Server.setPort(properties, 6000);

        // Build and start server
        RaftServer raftServer = RaftServer.newBuilder()
            .setGroup(raftGroup)
            .setProperties(properties)
            .setServerId(peerId)
            .setStateMachine(new CounterStateMachine())
            .build();

        raftServer.start();

        System.out.println("Counter server started on " + serverAddress);

        // Keep server running
        Thread.currentThread().join();
    }
}

Step 4: Create the Client

import org.apache.ratis.client.RaftClient;
import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.protocol.*;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

public class CounterClient {

    public static void main(String[] args) throws Exception {

        // Create Raft peer (server address)
        RaftPeer peer = RaftPeer.newBuilder()
            .setId(RaftPeerId.valueOf("n1"))
            .setAddress("127.0.0.1:6000")
            .build();

        // Create Raft group
        RaftGroupId groupId = RaftGroupId.valueOf(
            UUID.fromString("02511d47-d67c-49a3-9011-abb3109a44c1")
        );
        RaftGroup raftGroup = RaftGroup.valueOf(
            groupId,
            peer
        );

        // Build client
        RaftProperties properties = new RaftProperties();
        RaftClient client = RaftClient.newBuilder()
            .setProperties(properties)
            .setRaftGroup(raftGroup)
            .setClientRpc(
                GrpcFactory.newRaftClientRpc(properties)
            )
            .build();

        // Send INCREMENT command (write operation)
        RaftClientReply reply1 = client.io().send(
            Message.valueOf("INCREMENT")
        );
        System.out.println("Counter after increment: " + 
            reply1.getMessage().getContent().toString(StandardCharsets.UTF_8));

        // Send another INCREMENT
        RaftClientReply reply2 = client.io().send(
            Message.valueOf("INCREMENT")
        );
        System.out.println("Counter after increment: " + 
            reply2.getMessage().getContent().toString(StandardCharsets.UTF_8));

        // Send GET command using read-only operation for efficiency
        // Note: For production, use sendReadOnly() for GET operations
        // to avoid unnecessary Raft log writes
        RaftClientReply reply3 = client.io().sendReadOnly(
            Message.valueOf("GET")
        );
        System.out.println("Current counter value: " + 
            reply3.getMessage().getContent().toString(StandardCharsets.UTF_8));

        client.close();
    }
}

Step 5: Running the Example

Terminal 1 — Start Server:

java -cp target/counter-example.jar CounterServer

Terminal 2 — Run Client:

java -cp target/counter-example.jar CounterClient

Expected Output:

Counter after increment: 1
Counter after increment: 2
Current counter value: 2

Production Configuration (3-Node Cluster)

For a production setup with fault tolerance, configure 3 servers:

// Server 1: n1 at 127.0.0.1:6000
// Server 2: n2 at 127.0.0.1:6001
// Server 3: n3 at 127.0.0.1:6002

List<RaftPeer> peers = Arrays.asList(
    RaftPeer.newBuilder().setId("n1").setAddress("127.0.0.1:6000").build(),
    RaftPeer.newBuilder().setId("n2").setAddress("127.0.0.1:6001").build(),
    RaftPeer.newBuilder().setId("n3").setAddress("127.0.0.1:6002").build()
);

RaftGroup raftGroup = RaftGroup.valueOf(groupId, peers);

Now even if one server fails, the system continues operating with the remaining two servers maintaining consensus.

More Complex Real-World Example: Distributed Key-Value Store

Apache Ratis includes a FileStore example that demonstrates a high-performance file service supporting read, write, and delete operations using an asynchronous event-driven model.

You can find complete working examples in the Apache Ratis repository:

• GitHub Examples: https://github.com/apache/ratis/tree/master/ratis-examples

Available Examples:

• FileStore: High-performance distributed file service with async I/O
• Arithmetic: Calculator service with replicated state
• Counter: Simple counter implementation (similar to our example above)

These examples show how to build production-ready distributed systems with Ratis handling all the complex consensus coordination while you focus on your business logic.

Apache Ratis vs. Other Consensus Systems

For those evaluating different consensus implementations, here’s how Apache Ratis compares:

Feature Apache Ratis etcd Consul Language Java Go Go Use Case Embeddable consensus library Standalone distributed key-value store Service mesh & service discovery Pluggability Highly pluggable (transport, log, state machine) Fixed implementation Fixed implementation Integration Deep integration with Hadoop ecosystem Kubernetes-native HashiCorp ecosystem Best For Building custom distributed systems Configuration management Service networking

Apache Ratis shines when you need to embed consensus into your Java application and want fine-grained control over the implementation details.

Conclusion

Apache Ratis provides a production-ready, high-performance implementation of the Raft consensus algorithm that has been battle-tested in some of the largest distributed systems in the world. From managing petabytes of data in Hadoop Ozone clusters with thousands of nodes to coordinating IoT time-series databases, Ratis has proven its reliability and scalability.

If you’re building a distributed system that requires strong consistency, fault tolerance, and high availability, Apache Ratis offers a mature, well-documented solution backed by an active Apache community. Its pluggable architecture allows you to customize nearly every aspect of the system while benefiting from a proven consensus implementation.

Whether you’re building the next generation of distributed databases, object stores, or coordination services, Apache Ratis provides the solid foundation of consensus you need to ensure your system remains consistent and available in the face of failures.

Resources

• Apache Ratis Website: https://ratis.apache.org/
• GitHub Repository: https://github.com/apache/ratis
• Documentation: Available on the official website
• Community: Join the Apache Ratis mailing lists for support and discussion