Fault-Tolerant Distributed Messaging System

A robust distributed messaging system implemented in Java using Apache ZooKeeper for coordination and a custom Raft consensus algorithm for strong consistency and fault tolerance.

Features

• Strong Consistency: Implements Raft consensus algorithm for linearizable message ordering
• High Availability: Automatic leader election and failover with sub-400ms recovery time
• Fault Tolerance: Tolerates up to (N-1)/2 node failures in an N-node cluster
• Message Durability: Persistent append-only logs with crash recovery
• Time Synchronization: NTP integration with Hybrid Logical Clock (HLC) for distributed ordering
• ZooKeeper Integration: Cluster membership management and leader discovery
• Real-time Messaging: HTTP-based client API with automatic leader discovery

Architecture

Core Components

1. ServerNode: Distributed server instances that form the messaging cluster
2. ClientNode: Client applications that send and retrieve messages
3. RaftConsensus: Implementation of Raft consensus algorithm
4. LogManager: Persistent message storage with crash recovery
5. TimeService: NTP synchronization with Hybrid Logical Clock
6. ZooKeeperManager: Cluster coordination and leader election
7. ReplicationManager: Background log synchronization between nodes

System Flow

Client → Leader → Followers → Consensus → Commit → Response
      ↘         ↗
        ZooKeeper (Leader Discovery)

Prerequisites

• Java 21 or higher
• Apache ZooKeeper 3.9.2 (for cluster coordination)
• Maven 3.8+ (for building)

Installation & Setup

1. Install ZooKeeper

Download and install Apache ZooKeeper from https://zookeeper.apache.org/

Set the ZOOKEEPER_HOME environment variable:

set ZOOKEEPER_HOME=C:\path\to\zookeeper

2. Build the Project

cd messageSys
mvn clean compile
mvn dependency:copy-dependencies -DoutputDirectory=target/lib

3. Start ZooKeeper

# Windows
scripts\start-zookeeper.bat

# Or manually
cd %ZOOKEEPER_HOME%\bin
zkServer.cmd

Running the System

Start Server Nodes

Open separate terminals for each server:

# Terminal 1 - Server 1
scripts\start-server1.bat

# Terminal 2 - Server 2  
scripts\start-server2.bat

# Terminal 3 - Server 3
scripts\start-server3.bat

Or manually:

java -cp "target\classes;target\lib\*" cs.ds.messagesys.DistributedServerNode config\server1.properties

Start Client

scripts\start-client.bat

Or manually:

java -cp "target\classes;target\lib\*" cs.ds.messagesys.ClientNode config\client.properties

Client Usage

The client provides an interactive CLI:

=== Distributed Messaging Client ===
Client ID: client-1
Commands:
  send <message>  - Send a message
  get             - Get new messages  
  get <index>     - Get messages from index
  status          - Show current status
  help            - Show this help
  quit            - Exit

client-1> send Hello, distributed world!
Message sent successfully

client-1> get
[1] client-1: Hello, distributed world! (term=1, ts=1697123456789)

client-1> status
Client ID: client-1
Current Leader: ServerNode{nodeId='server-1', address='localhost:8081', zkPath='null'}
Last Retrieved Index: 1

API Endpoints

Client Endpoints (Leader Only)

• POST /send - Send a message

  {
    "message": "Hello, World!",
    "sender": "client-1"
  }

• GET /messages?fromIndex=1 - Retrieve messages

• GET /status - Get node status

Raft Endpoints (Internal)

• POST /appendEntry - Raft AppendEntries RPC
• POST /requestVote - Raft RequestVote RPC
• GET /health - Health check

Configuration

Server Configuration (config/server1.properties)

node.id=server-1
node.host=localhost
node.port=8081
zookeeper.connect=localhost:2181

Client Configuration (config/client.properties)

client.id=client-1
zookeeper.connect=localhost:2181

Testing

Run Unit Tests

mvn test

Manual Testing Scenarios

1. Basic Messaging

1. Start 3 servers
2. Start client
3. Send messages: send Hello World
4. Retrieve messages: get

2. Leader Failure Test

1. Start 3 servers, identify leader
2. Send some messages
3. Kill leader process (Ctrl+C)
4. Send more messages (should work with new leader)
5. Verify all messages are present

3. Network Partition Test

1. Start 5 servers
2. Send messages
3. Stop 2 servers (minority)
4. Continue sending (should work)
5. Restart stopped servers
6. Verify consistency

4. Clock Skew Test

1. Start servers with different system times
2. Send messages from multiple clients
3. Verify HLC maintains ordering

Performance Characteristics

• Leader Election Time: < 400ms
• Message Throughput: ~1000 messages/second per leader
• Replication Latency: < 50ms for 3-node cluster
• Clock Drift Tolerance: ±40ms with NTP sync
• Recovery Time: < 5 seconds for follower rejoin

Monitoring

Check Node Status

curl http://localhost:8081/status

Response:

{
  "nodeId": "server-1",
  "role": "LEADER",
  "term": 3,
  "commitIndex": 42,
  "isLeader": true,
  "timestamp": 1697123456789,
  "ntpSynced": true,
  "ntpOffset": -12
}

Log Files

• Server logs: Console output with SLF4J
• Persistent logs: data/log_<nodeId>.json
• Raft state: data/raft_state_<nodeId>.json

Troubleshooting

Common Issues

1. ZooKeeper Connection Failed

   • Ensure ZooKeeper is running on port 2181
   • Check firewall settings

2. Leader Election Timeout

   • Verify all nodes can communicate
   • Check network connectivity between servers

3. Message Loss

   • Check that majority of nodes are running
   • Verify disk space for log files

4. Clock Skew Warnings

   • Synchronize system clocks with NTP
   • Check network latency between nodes

Debug Mode

Enable verbose logging:

java -Dorg.slf4j.simpleLogger.defaultLogLevel=debug -cp "target\classes;target\lib\*" cs.ds.messagesys.DistributedServerNode config\server1.properties

Architecture Details

Raft Implementation

• Leader Election: Randomized timeouts (150-300ms)
• Log Replication: Batched AppendEntries with heartbeats
• Safety: Log matching property ensures consistency
• Persistence: Current term, voted for, and log entries

Time Synchronization

• NTP Client: Queries pool.ntp.org, time.google.com
• Hybrid Logical Clock: Combines physical and logical time
• Drift Detection: Warns if clock drift exceeds 40ms

ZooKeeper Integration

• Membership: Ephemeral sequential nodes under /servers
• Leader Discovery: Leader info stored in /leader
• Failure Detection: ZooKeeper session timeouts

Contributing

1. Fork the repository
2. Create a feature branch
3. Implement changes with tests
4. Submit a pull request

License

This project is licensed under the MIT License - see the LICENSE file for details.

Authors

• Distributed Systems Team - Implementation of fault-tolerant messaging system

Acknowledgments

• Apache ZooKeeper team for coordination primitives
• Raft consensus algorithm by Diego Ongaro and John Ousterhout
• NTP protocol for time synchronization
• Hybrid Logical Clock concept by Kulkarni et al.