Consistent Hashing — Why One Node Broke 75% of Keys

import java.util.*;

/**
 * Demonstrates the rehash avalanche problem with naive modulo hashing.
 * Run this to see WHY consistent hashing was invented.
 */
public class NaiveModuloHashingDemo {

    // Simulates mapping a set of cache keys to servers using simple modulo
    static Map<String, String> assignKeysToServers(List<String> keys, List<String> servers) {
        Map<String, String> assignment = new LinkedHashMap<>();
        for (String key : keys) {
            // Standard modulo hash: server index = hash(key) mod server_count
            int serverIndex = Math.abs(key.hashCode()) % servers.size();
            assignment.put(key, servers.get(serverIndex));
        }
        return assignment;
    }

    public static void main(String[] args) {
        List<String> cacheKeys = List.of(
            "user:1001", "user:1002", "session:abc", "session:def",
            "product:55", "product:56", "order:789", "order:790",
            "cart:aaa", "cart:bbb"
        );

        // --- BEFORE: cluster has 3 servers ---
        List<String> servers3 = List.of("cache-server-A", "cache-server-B", "cache-server-C");
        Map<String, String> beforeScaleOut = assignKeysToServers(cacheKeys, servers3);

        // --- AFTER: we add one server, cluster now has 4 servers ---
        List<String> servers4 = List.of("cache-server-A", "cache-server-B", "cache-server-C", "cache-server-D");
        Map<String, String> afterScaleOut = assignKeysToServers(cacheKeys, servers4);

        // Count how many keys moved to a DIFFERENT server after the resize
        int keysMoved = 0;
        System.out.println("Key assignment changes after adding one server:");
        System.out.printf("%-20s %-20s %-20s %-10s%n", "Key", "Before (3 nodes)", "After (4 nodes)", "Moved?");
        System.out.println("-".repeat(75));

        for (String key : cacheKeys) {
            String before = beforeScaleOut.get(key);
            String after = afterScaleOut.get(key);
            boolean moved = !before.equals(after);
            if (moved) keysMoved++;
            System.out.printf("%-20s %-20s %-20s %-10s%n",
                key, before, after, moved ? "<-- MOVED" : "stable");
        }

        double percentMoved = (keysMoved * 100.0) / cacheKeys.size();
        System.out.println("-".repeat(75));
        System.out.printf("Keys moved: %d / %d (%.0f%%)%n", keysMoved, cacheKeys.size(), percentMoved);
        System.out.println("With modulo hashing, expect ~75%+ of keys to move on ANY cluster resize.");
        System.out.println("With consistent hashing, expect ~1/N keys to move (25% with 4 nodes).");
    }
}

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.*;

/**
 * A production-quality consistent hash ring implementation.
 * Supports virtual nodes (vnodes) to fix the uneven distribution problem.
 *
 * Key design decisions:
 *  - TreeMap for O(log N) clockwise-neighbor lookup via ceilingKey()
 *  - MD5 for strong hash distribution (swap for MurmurHash in perf-critical code)
 *  - Virtual nodes multiplied per physical server to smooth load distribution
 */
public class ConsistentHashRing {

    // The ring: maps a hash position (long) to a server label
    // TreeMap keeps positions sorted so we can do clockwise lookups efficiently
    private final TreeMap<Long, String> hashRing = new TreeMap<>();

    // How many virtual node positions each physical server gets on the ring
    private final int virtualNodesPerServer;

    // Track physical servers for clean removal later
    private final Set<String> physicalServers = new HashSet<>();

    public ConsistentHashRing(int virtualNodesPerServer) {
        this.virtualNodesPerServer = virtualNodesPerServer;
    }

    // --- Hash a string to a long position on the ring ---
    private long hashToRingPosition(String input) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] digest = md.digest(input.getBytes());
            // Take first 8 bytes of MD5 to form a long
            // This gives us a 64-bit position space (0 to Long.MAX_VALUE)
            long position = 0;
            for (int i = 0; i < 8; i++) {
                position = (position << 8) | (digest[i] & 0xFFL);
            }
            // Ensure positive value for the ring
            return position & Long.MAX_VALUE;
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException("MD5 not available — this should never happen on JVM", e);
        }
    }

    /**
     * Adds a physical server to the ring by creating N virtual node positions.
     * Each vnode is hashed as "serverName#replicaIndex" to spread positions.
     */
    public void addServer(String serverName) {
        physicalServers.add(serverName);
        for (int replica = 0; replica < virtualNodesPerServer; replica++) {
            // Virtual node label: e.g. "cache-server-A#42"
            String vnodeLabel = serverName + "#" + replica;
            long ringPosition = hashToRingPosition(vnodeLabel);
            hashRing.put(ringPosition, serverName);
        }
        System.out.printf("[RING] Added server '%s' with %d virtual nodes%n",
            serverName, virtualNodesPerServer);
    }

    /**
     * Removes a physical server and all its virtual nodes from the ring.
     */
    public void removeServer(String serverName) {
        physicalServers.remove(serverName);
        for (int replica = 0; replica < virtualNodesPerServer; replica++) {
            String vnodeLabel = serverName + "#" + replica;
            long ringPosition = hashToRingPosition(vnodeLabel);
            hashRing.remove(ringPosition);
        }
        System.out.printf("[RING] Removed server '%s' and all its virtual nodes%n", serverName);
    }

    /**
     * Routes a key to its responsible server.
     * Walks clockwise from the key's hash position to find the first server.
     */
    public String getServerForKey(String key) {
        if (hashRing.isEmpty()) {
            throw new IllegalStateException("Hash ring is empty — add servers before routing keys");
        }
        long keyPosition = hashToRingPosition(key);

        // ceilingKey returns the smallest key >= keyPosition
        // This IS the clockwise-walk operation — O(log N)
        Map.Entry<Long, String> responsibleNode = hashRing.ceilingEntry(keyPosition);

        // If no entry >= keyPosition, wrap around to the start of the ring
        if (responsibleNode == null) {
            responsibleNode = hashRing.firstEntry();
        }
        return responsibleNode.getValue();
    }

    /**
     * Analyzes how evenly keys are distributed across physical servers.
     * This is your load balance health check.
     */
    public void printLoadDistribution(List<String> keysToAnalyze) {
        Map<String, Integer> loadCount = new TreeMap<>();
        physicalServers.forEach(s -> loadCount.put(s, 0));

        for (String key : keysToAnalyze) {
            String server = getServerForKey(key);
            loadCount.merge(server, 1, Integer::sum);
        }

        System.out.println("\n--- Load Distribution Report ---");
        System.out.printf("Total keys: %d | Servers: %d | VNodes per server: %d%n",
            keysToAnalyze.size(), physicalServers.size(), virtualNodesPerServer);
        System.out.printf("%-20s %-10s %-10s%n", "Server", "Keys", "Share");
        System.out.println("-".repeat(45));
        loadCount.forEach((server, count) -> {
            double share = (count * 100.0) / keysToAnalyze.size();
            System.out.printf("%-20s %-10d %.1f%%%n", server, count, share);
        });
        System.out.println("-".repeat(45));
    }

    public static void main(String[] args) {
        // Build a sample cluster with 3 cache servers
        // 150 virtual nodes per server is a common production starting point
        ConsistentHashRing ring = new ConsistentHashRing(150);
        ring.addServer("cache-server-A");
        ring.addServer("cache-server-B");
        ring.addServer("cache-server-C");

        // Generate 1000 representative cache keys
        List<String> cacheKeys = new ArrayList<>();
        for (int i = 1; i <= 1000; i++) {
            cacheKeys.add("key:" + i);
        }

        // Show initial distribution
        ring.printLoadDistribution(cacheKeys);

        // Record key assignments BEFORE adding a new server
        Map<String, String> assignmentBefore = new LinkedHashMap<>();
        for (String key : cacheKeys) {
            assignmentBefore.put(key, ring.getServerForKey(key));
        }

        // Scale out: add a 4th server
        System.out.println("\n[EVENT] Scaling out — adding cache-server-D...");
        ring.addServer("cache-server-D");

        // Show new distribution
        ring.printLoadDistribution(cacheKeys);

        // Count how many keys actually moved
        long keysMoved = cacheKeys.stream()
            .filter(key -> !assignmentBefore.get(key).equals(ring.getServerForKey(key)))
            .count();
        double percentMoved = (keysMoved * 100.0) / cacheKeys.size();
        System.out.printf("%n[RESULT] Keys moved after adding 1 server: %d / %d (%.1f%%)%n",
            keysMoved, cacheKeys.size(), percentMoved);
        System.out.printf("[EXPECTED] Ideal minimum: ~%.1f%% (1 / %d servers)%n",
            100.0 / 4, 4);
    }
}

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.*;

/**
 * Compares distribution quality across different vnode counts.
 * Shows WHY 1 vnode per server is dangerous and 100-200 is the sweet spot.
 *
 * Run this to see the standard deviation of server load shrink as vnodes increase.
 */
public class VirtualNodeDistributionAnalysis {

    private static long hashToPosition(String input) throws NoSuchAlgorithmException {
        MessageDigest md = MessageDigest.getInstance("MD5");
        byte[] digest = md.digest(input.getBytes());
        long position = 0;
        for (int i = 0; i < 8; i++) {
            position = (position << 8) | (digest[i] & 0xFFL);
        }
        return position & Long.MAX_VALUE;
    }

    /**
     * Simulates a consistent hash ring and returns per-server key counts.
     *
     * @param serverNames     physical server names
     * @param vnodesPerServer how many ring positions per physical server
     * @param totalKeys       how many test keys to route
     */
    static Map<String, Integer> simulateDistribution(
            List<String> serverNames, int vnodesPerServer, int totalKeys)
            throws NoSuchAlgorithmException {

        TreeMap<Long, String> ring = new TreeMap<>();
        // Place virtual nodes on the ring
        for (String server : serverNames) {
            for (int v = 0; v < vnodesPerServer; v++) {
                long pos = hashToPosition(server + "#vnode" + v);
                ring.put(pos, server);
            }
        }

        // Route keys and count per server
        Map<String, Integer> serverLoad = new TreeMap<>();
        serverNames.forEach(s -> serverLoad.put(s, 0));

        for (int k = 0; k < totalKeys; k++) {
            long keyPos = hashToPosition("testkey:" + k);
            Map.Entry<Long, String> node = ring.ceilingEntry(keyPos);
            if (node == null) node = ring.firstEntry();
            serverLoad.merge(node.getValue(), 1, Integer::sum);
        }
        return serverLoad;
    }

    static double standardDeviation(Collection<Integer> values) {
        double mean = values.stream().mapToInt(i -> i).average().orElse(0);
        double variance = values.stream()
            .mapToDouble(v -> Math.pow(v - mean, 2))
            .average().orElse(0);
        return Math.sqrt(variance);
    }

    public static void main(String[] args) throws NoSuchAlgorithmException {
        List<String> servers = List.of(
            "node-alpha", "node-beta", "node-gamma", "node-delta"
        );
        int totalKeys = 10_000;
        int[] vnodeCounts = {1, 5, 25, 100, 200, 500};

        System.out.println("Vnode Distribution Analysis — 4 servers, 10,000 keys");
        System.out.printf("%-15s %-15s %-15s %-20s%n",
            "VNodes/Server", "Min Load", "Max Load", "Std Dev (lower=better)");
        System.out.println("=" .repeat(70));

        for (int vnodesPerServer : vnodeCounts) {
            Map<String, Integer> distribution =
                simulateDistribution(servers, vnodesPerServer, totalKeys);

            int minLoad = Collections.min(distribution.values());
            int maxLoad = Collections.max(distribution.values());
            double stdDev = standardDeviation(distribution.values());

            System.out.printf("%-15d %-15d %-15d %.1f keys%n",
                vnodesPerServer, minLoad, maxLoad, stdDev);
        }

        System.out.println("\nConclusion: 100-200 vnodes is the practical sweet spot.");
        System.out.println("Beyond ~200, distribution improvement is marginal vs memory overhead.");
    }
}

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.*;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * Thread-safe consistent hash ring with replication support.
 *
 * Two critical additions over the basic version:
 *  1. getReplicaServers() returns N distinct physical servers for replication
 *  2. ReadWriteLock ensures safe concurrent access during topology changes
 *
 * This is the version you would actually build in a distributed datastore.
 */
public class ReplicatedConsistentHashRing {

    private final TreeMap<Long, String> ring = new TreeMap<>();
    private final Set<String> liveServers = new HashSet<>();
    private final int vnodesPerServer;

    // ReadWriteLock: many readers (key lookups) + rare writers (add/remove server)
    private final ReadWriteLock ringLock = new ReentrantReadWriteLock();

    public ReplicatedConsistentHashRing(int vnodesPerServer) {
        this.vnodesPerServer = vnodesPerServer;
    }

    private long hash(String input) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] d = md.digest(input.getBytes());
            long h = 0;
            for (int i = 0; i < 8; i++) h = (h << 8) | (d[i] & 0xFFL);
            return h & Long.MAX_VALUE;
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException(e);
        }
    }

    public void addServer(String serverName) {
        // Exclusive write lock — stops all readers while ring topology changes
        ringLock.writeLock().lock();
        try {
            liveServers.add(serverName);
            for (int v = 0; v < vnodesPerServer; v++) {
                ring.put(hash(serverName + "#" + v), serverName);
            }
        } finally {
            ringLock.writeLock().unlock(); // Always release in finally!
        }
    }

    public void removeServer(String serverName) {
        ringLock.writeLock().lock();
        try {
            liveServers.remove(serverName);
            for (int v = 0; v < vnodesPerServer; v++) {
                ring.remove(hash(serverName + "#" + v));
            }
        } finally {
            ringLock.writeLock().unlock();
        }
    }

    /**
     * Returns an ordered list of `replicationFactor` distinct physical servers
     * responsible for this key — starting with the primary, then replicas.
     *
     * Walk clockwise, collecting servers until we have `replicationFactor`
     * distinct physical servers (skip extra vnodes of servers already collected).
     */
    public List<String> getReplicaServers(String key, int replicationFactor) {
        // Shared read lock — many threads can look up keys concurrently
        ringLock.readLock().lock();
        try {
            if (ring.isEmpty()) throw new IllegalStateException("Ring is empty");
            if (replicationFactor > liveServers.size()) {
                throw new IllegalArgumentException(
                    "Replication factor " + replicationFactor +
                    " exceeds live server count " + liveServers.size());
            }

            List<String> replicas = new ArrayList<>(replicationFactor);
            long keyPos = hash(key);

            // Start clockwise walk from key position
            NavigableMap<Long, String> clockwiseView = ring.tailMap(keyPos, true);

            // Iterate the entire ring clockwise using two passes (tail + head)
            Iterator<String> iter = new Iterator<>() {
                private final Iterator<String> tailIter = clockwiseView.values().iterator();
                private final Iterator<String> headIter = ring.values().iterator();
                private boolean onHead = false;

                @Override public boolean hasNext() {
                    return tailIter.hasNext() || (!onHead && headIter.hasNext());
                }

                @Override public String next() {
                    if (tailIter.hasNext()) return tailIter.next();
                    onHead = true;
                    return headIter.next();
                }
            };

            // Collect distinct physical servers as we walk clockwise
            while (iter.hasNext() && replicas.size() < replicationFactor) {
                String candidate = iter.next();
                // Only add if this physical server is not already in our replica set
                if (!replicas.contains(candidate)) {
                    replicas.add(candidate);
                }
            }
            return replicas;
        } finally {
            ringLock.readLock().unlock();
        }
    }

    public static void main(String[] args) {
        ReplicatedConsistentHashRing ring = new ReplicatedConsistentHashRing(100);
        ring.addServer("node-1");
        ring.addServer("node-2");
        ring.addServer("node-3");
        ring.addServer("node-4");
        ring.addServer("node-5");

        String[] keysToRoute = {"user:alice", "user:bob", "session:xyz", "product:42"};
        int replicationFactor = 3; // Write to 3 servers for fault tolerance

        System.out.println("Replica placement with RF=3 (primary first, then replicas):");
        System.out.printf("%-20s %-40s%n", "Key", "Responsible Servers (in order)");
        System.out.println("-".repeat(65));

        for (String key : keysToRoute) {
            List<String> servers = ring.getReplicaServers(key, replicationFactor);
            System.out.printf("%-20s %s%n", key, servers);
        }

        // Simulate node failure: remove node-3
        System.out.println("\n[FAILURE] node-3 goes down...");
        ring.removeServer("node-3");

        System.out.println("\nReplica placement AFTER node-3 failure:");
        System.out.printf("%-20s %-40s%n", "Key", "Responsible Servers (in order)");
        System.out.println("-".repeat(65));
        for (String key : keysToRoute) {
            List<String> servers = ring.getReplicaServers(key, replicationFactor);
            System.out.printf("%-20s %s%n", key, servers);
        }
    }
}

import java.util.*;

/**
 * Demonstration of hash slot assignment (like Redis Cluster) using a consistent hash ring.
 * This is NOT a full implementation, but shows the mapping concept.
 */
public class HashSlotRing {

    // Fixed number of slots (like Redis Cluster's 16384)
    private static final int NUM_SLOTS = 16384;

    // Map slot -> physical server
    private final String[] slotOwners = new String[NUM_SLOTS];

    // Set of servers
    private final Set<String> servers = new HashSet<>();

    /**
     * Initializes the ring: distributes slots evenly across servers.
     * A simple round-robin assignment for demonstration.
     */
    public void initializeCluster(List<String> serverList) {
        servers.addAll(serverList);
        for (int slot = 0; slot < NUM_SLOTS; slot++) {
            int serverIndex = slot % serverList.size();
            slotOwners[slot] = serverList.get(serverIndex);
        }
    }

    /**
     * Assigns a key to a slot using a hash function, then returns the owning server.
     */
    public String getServerForKey(String key) {
        int slot = Math.abs(key.hashCode()) % NUM_SLOTS;
        return slotOwners[slot];
    }

    /**
     * Migrates a set of slots from one server to another (e.g., during cluster expansion).
     */
    public void migrateSlots(List<Integer> slots, String fromServer, String toServer) {
        // In production, this would be transactional with data migration
        for (int slot : slots) {
            if (!slotOwners[slot].equals(fromServer)) {
                throw new IllegalStateException("Slot " + slot + " not owned by " + fromServer);
            }
            slotOwners[slot] = toServer;
        }
        System.out.println("Migrated " + slots.size() + " slots to " + toServer);
    }

    public static void main(String[] args) {
        HashSlotRing ring = new HashSlotRing();
        List<String> servers = List.of("node-1", "node-2", "node-3");
        ring.initializeCluster(servers);

        System.out.println("Initial assignment for sample keys:");
        String[] testKeys = {"user:1001", "product:42", "order:789"};
        for (String key : testKeys) {
            System.out.println(key + " -> " + ring.getServerForKey(key));
        }

        // Add a new server
        System.out.println("\nAdding node-4 and migrating slots...");
        // Migrate 4000 slots from each existing node to node-4 (simplistic)
        for (int i = 0; i < 4000; i++) {
            // Each existing node gives up slots proportionally
            int slotToMigrate = i; // simplistic
            String currentOwner = ring.slotOwners[slotToMigrate];
            if (!currentOwner.equals("node-4")) {
                ring.migrateSlots(List.of(slotToMigrate), currentOwner, "node-4");
            }
        }

        System.out.println("\nAfter migration:");
        for (String key : testKeys) {
            System.out.println(key + " -> " + ring.getServerForKey(key));
        }
    }
}