+---------------------------------------------------------------+
|              FAILURE PROBABILITY AT SCALE                     |
+---------------------------------------------------------------+
|                                                               |
|  Single Machine (99.9% uptime):                               |
|  ┌────────────────────────────────────────┐                   |
|  │ Downtime: ~8.7 hours/year             │                   |
|  │ Reliability: Good for single server    │                   |
|  └────────────────────────────────────────┘                   |
|                                                               |
|  1000-Node Cluster (each 99.9% uptime):                       |
|  ┌────────────────────────────────────────┐                   |
|  │ At any moment:                         │                   |
|  │ • 1 node down on average               │                   |
|  │ • Multiple failures per day likely     │                   |
|  │ • Correlated failures possible         │                   |
|  │                                        │                   |
|  │ Without fault tolerance:               │                   |
|  │ ✗ Jobs fail constantly                 │                   |
|  │ ✗ Data loss is inevitable              │                   |
|  │ ✗ System becomes unusable              │                   |
|  └────────────────────────────────────────┘                   |
|                                                               |
|  With Hadoop Fault Tolerance:                                 |
|  ┌────────────────────────────────────────┐                   |
|  │ ✓ Jobs complete despite failures       │                   |
|  │ ✓ Data is never lost                   │                   |
|  │ ✓ System remains operational           │                   |
|  │ ✓ Transparent recovery                 │                   |
|  └────────────────────────────────────────┘                   |
|                                                               |
+---------------------------------------------------------------+

+---------------------------------------------------------------+
|                 HDFS REPLICATION STRATEGY                     |
+---------------------------------------------------------------+
|                                                               |
|  Block Placement Policy (Rack-Aware):                         |
|                                                               |
|    Rack 1              Rack 2              Rack 3             |
|  ┌─────────┐        ┌─────────┐        ┌─────────┐           |
|  │ Node A  │        │ Node C  │        │ Node E  │           |
|  │  Blk 1  │◄───────│  Blk 1  │        │         │           |
|  │ [Replica│   Copy │ [Replica│        │         │           |
|  │    1]   │        │    2]   │        │         │           |
|  └─────────┘        └─────────┘        └─────────┘           |
|  ┌─────────┐        ┌─────────┐        ┌─────────┐           |
|  │ Node B  │        │ Node D  │        │ Node F  │           |
|  │  Blk 1  │        │         │        │         │           |
|  │ [Replica│        │         │        │         │           |
|  │    3]   │        │         │        │         │           |
|  └─────────┘        └─────────┘        └─────────┘           |
|      ▲                                                        |
|      │                                                        |
|      └──────────────────────────────────┐                     |
|                                         │                     |
|  Placement Strategy:                    │                     |
|  1. First replica: Same node as writer (or random)            |
|  2. Second replica: Different rack                            |
|  3. Third replica: Same rack as #2, different node            |
|                                                               |
|  Benefits:                                                    |
|  • Survives rack failure (lose 1 replica)                     |
|  • Survives node failures (lose 1-2 replicas)                 |
|  • Fast local reads when possible                             |
|  • Balanced network usage                                     |
|                                                               |
+---------------------------------------------------------------+

// HDFS replication configuration examples

// 1. Global default (hdfs-site.xml)
<property>
  <name>dfs.replication</name>
  <value>3</value>
  <description>Default block replication factor</description>
</property>

// 2. Per-file replication at write time
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);

// Set replication for specific file
Path filePath = new Path("/data/important.txt");
FSDataOutputStream out = fs.create(
    filePath,
    true,        // overwrite
    4096,        // buffer size
    (short) 5,   // replication factor = 5 (critical data)
    134217728    // block size (128MB)
);

// 3. Change replication for existing file
fs.setReplication(new Path("/data/logs.txt"), (short) 2);

// 4. Directory-level replication policy
// All files in this directory get replication=2
fs.setReplication(new Path("/data/temp/"), (short) 2);

// 5. Programmatic replication management
public class ReplicationManager {

    public void adjustReplicationByImportance(
            FileSystem fs,
            Path path,
            String importance) throws IOException {

        short replication;
        switch (importance) {
            case "CRITICAL":
                replication = 5;  // High durability
                break;
            case "IMPORTANT":
                replication = 3;  // Standard
                break;
            case "TEMPORARY":
                replication = 2;  // Reduce storage
                break;
            case "CACHE":
                replication = 1;  // Minimal (risky)
                break;
            default:
                replication = 3;
        }

        fs.setReplication(path, replication);
    }

    // Monitor and maintain replication levels
    public void ensureReplication(FileSystem fs, Path path)
            throws IOException {

        FileStatus status = fs.getFileStatus(path);
        BlockLocation[] blocks = fs.getFileBlockLocations(
            status, 0, status.getLen()
        );

        for (BlockLocation block : blocks) {
            int currentReplicas = block.getHosts().length;
            int targetReplicas = status.getReplication();

            if (currentReplicas < targetReplicas) {
                System.err.println(
                    "Under-replicated block: " +
                    block.getNames()[0] +
                    " (has " + currentReplicas +
                    ", needs " + targetReplicas + ")"
                );
            }
        }
    }
}

REPLICATION (3x) vs. ERASURE CODING (RS-6-3)
─────────────────────────────────────────────

REPLICATION:
[Block A] [Block A] [Block A]
Total: 3x Storage

ERASURE CODING (RS-6-3):
[d1] [d2] [d3] [d4] [d5] [d6] | [p1] [p2] [p3]
Total: 1.5x Storage (50% overhead)

+---------------------------------------------------------------+
|              HDFS BLOCK RECOVERY WORKFLOW                     |
+---------------------------------------------------------------+
|                                                               |
|  Step 1: Failure Detection                                    |
|  ┌────────────────────────────────────────┐                   |
|  │ NameNode monitors DataNodes via        │                   |
|  │ heartbeats (every 3 seconds)           │                   |
|  │                                        │                   |
|  │ DataNode Timeout:                      │                   |
|  │ • Heartbeat interval: 3s               │                   |
|  │ • Timeout threshold: 10 min            │                   |
|  │ • Recheck interval: 5 min              │                   |
|  │                                        │                   |
|  │ If no heartbeat → Node marked DEAD     │                   |
|  └────────────────────────────────────────┘                   |
|           ↓                                                   |
|  Step 2: Identify Under-Replicated Blocks                     |
|  ┌────────────────────────────────────────┐                   |
|  │ NameNode scans block metadata:         │                   |
|  │                                        │                   |
|  │ Block B1:                              │                   |
|  │   Expected replicas: 3                 │                   |
|  │   Current replicas: 2 (Node3 died)     │                   |
|  │   Status: UNDER-REPLICATED             │                   |
|  │                                        │                   |
|  │ Priority Queue:                        │                   |
|  │   1. Blocks with 0 replicas (HIGHEST)  │                   |
|  │   2. Blocks with 1 replica             │                   |
|  │   3. Blocks below target               │                   |
|  └────────────────────────────────────────┘                   |
|           ↓                                                   |
|  Step 3: Select Replication Source & Target                   |
|  ┌────────────────────────────────────────┐                   |
|  │ Source Selection:                      │                   |
|  │ • Choose healthy DataNode with copy    │                   |
|  │ • Prefer nodes with low load           │                   |
|  │ • Consider network topology            │                   |
|  │                                        │                   |
|  │ Target Selection:                      │                   |
|  │ • Choose node with free space          │                   |
|  │ • Follow rack-awareness policy         │                   |
|  │ • Balance cluster utilization          │                   |
|  └────────────────────────────────────────┘                   |
|           ↓                                                   |
|  Step 4: Copy Block                                           |
|  ┌────────────────────────────────────────┐                   |
|  │   Source DN          →        Target DN│                   |
|  │  ┌──────────┐              ┌──────────┐│                   |
|  │  │ Block B1 │─────Copy────→│ Block B1 ││                   |
|  │  │ (Replica)│              │  (New)   ││                   |
|  │  └──────────┘              └──────────┘│                   |
|  │                                        │                   |
|  │ • Stream block data                    │                   |
|  │ • Verify checksum                      │                   |
|  │ • Report to NameNode                   │                   |
|  └────────────────────────────────────────┘                   |
|           ↓                                                   |
|  Step 5: Update Metadata                                      |
|  ┌────────────────────────────────────────┐                   |
|  │ NameNode updates:                      │                   |
|  │                                        │                   |
|  │ Block B1:                              │                   |
|  │   Replicas: 3 ✓                        │                   |
|  │   Locations: [Node1, Node2, Node5]     │                   |
|  │   Status: HEALTHY                      │                   |
|  └────────────────────────────────────────┘                   |
|                                                               |
+---------------------------------------------------------------+

## Deep Dive: The Replication Pipeline

When a client writes a block, it doesn't send data to all three DataNodes simultaneously. Instead, it uses a **Pipeline** to minimize network bottlenecks.

### 1. Chain Replication
The client sends the data in 64KB packets to the first DataNode in the pipeline. The first DataNode receives the packet, writes it to its local disk, and immediately forwards it to the second DataNode. This continues until the last DataNode in the chain.

- **Pipeline Ordering**: The NameNode orders the pipeline to optimize for distance (Local -> Same Rack -> Remote Rack).
- **Acknowledge Pipeline**: ACKs flow in the reverse direction. The client only considers the packet "done" once it receives an ACK from the first DataNode (meaning all nodes have the packet).

### 2. Handling Pipeline Failures
If a DataNode fails during the write:
1.  **Close Pipeline**: The client stops sending data and closes the current pipeline.
2.  **Report Failure**: The client notifies the NameNode of the failed node.
3.  **Bypass/Resume**: The client removes the bad node from the list and opens a new pipeline with the remaining healthy nodes.
4.  **Re-replicate**: After the file is closed, the NameNode notices the replication factor is below target (e.g., 2 instead of 3) and triggers a background re-replication to a new node.

## Deep Dive: Heartbeats, Block Reports, and Scaling the NameNode

The NameNode's primary job is keeping the **Block Map** in sync with reality. It relies on three communication channels from every DataNode:

### 1. Heartbeats (Liveness)
- **Frequency**: Every 3 seconds (`dfs.heartbeat.interval`).
- **Content**: Node status (Disk space, CPU, Load).
- **Failure Detection**: If no heartbeat is received for 10 minutes, the NameNode marks the DataNode as "Dead".

### 2. Full Block Reports (Initial & Periodic)
- **Frequency**: Every 6 hours (`dfs.blockreport.intervalMsec`).
- **Content**: A complete list of ALL blocks stored on that DataNode.
- **Purpose**: This is the source of truth for the NameNode. On startup, the NameNode enters **Safe Mode** until a threshold percentage of blocks are reported by DataNodes.

### 3. Incremental Block Reports (Dynamic)
- **Trigger**: Whenever a block is received, deleted, or moved.
- **Content**: Only the changes since the last report.
- **Why it matters**: Full block reports are massive and can crash a NameNode if thousands of nodes report at once (the "Block Report Storm"). Incremental reports allow the NameNode to stay updated with minimal CPU overhead.

### NameNode High Availability

```text
+---------------------------------------------------------------+
|           NAMENODE HA ARCHITECTURE (QJM)                      |
+---------------------------------------------------------------+
|                                                               |
|  Active NameNode              Standby NameNode                |
|  ┌─────────────────┐          ┌─────────────────┐            |
|  │  • Serves reads │          │  • Warm standby │            |
|  │  • Handles writes│         │  • Applies edits│            |
|  │  • Updates edits│          │  • Ready to take│            |
|  │  • Sends to QJM │          │    over         │            |
|  └────────┬────────┘          └────────┬────────┘            |
|           │                            │                     |
|           │  Write                Read │                     |
|           │  Edits                Edits│                     |
|           ▼                            ▼                     |
|  ┌──────────────────────────────────────────┐                |
|  │      Quorum Journal Manager (QJM)        │                |
|  │  ┌──────┐    ┌──────┐    ┌──────┐       │                |
|  │  │ JN 1 │    │ JN 2 │    │ JN 3 │       │                |
|  │  │      │    │      │    │      │       │                |
|  │  │ Edit │    │ Edit │    │ Edit │       │                |
|  │  │ Log  │    │ Log  │    │ Log  │       │                |
|  │  └──────┘    └──────┘    └──────┘       │                |
|  │                                          │                |
|  │  Quorum: Majority must acknowledge       │                |
|  │  (2 out of 3 in this example)            │                |
|  └──────────────────────────────────────────┘                |
|           │                            │                     |
|           ▼                            ▼                     |
|  ┌─────────────────┐          ┌─────────────────┐            |
|  │  ZooKeeper      │          │  ZooKeeper      │            |
|  │  Failover       │◄────────►│  Failover       │            |
|  │  Controller     │  Monitor │  Controller     │            |
|  │  (ZKFC)         │          │  (ZKFC)         │            |
|  └─────────────────┘          └─────────────────┘            |
|           │                            │                     |
|           └──────────┬─────────────────┘                     |
|                      ▼                                       |
|           ┌──────────────────────┐                           |
|           │  ZooKeeper Ensemble  │                           |
|           │  ┌────┐ ┌────┐ ┌────┐│                           |
|           │  │ ZK │ │ ZK │ │ ZK ││                           |
|           │  │ 1  │ │ 2  │ │ 3  ││                           |
|           │  └────┘ └────┘ └────┘│                           |
|           │  (Leader Election)   │                           |
|           └──────────────────────┘                           |
|                                                               |
+---------------------------------------------------------------+

Failover Process:
1. Active NN fails
2. ZKFC detects failure (via health checks)
3. ZKFC releases ZooKeeper lock
4. Standby ZKFC acquires lock
5. Standby promotes itself to Active
6. Reads latest edits from QJM
7. Begins serving requests
8. Total failover time: ~seconds
9. **Epoch Fencing**: To prevent a "Zombie" Active NameNode (one that is slow but not dead) from corrupting the edits, QJM uses **Epoch Numbers**.
    - When a NameNode is promoted, it is assigned a unique, monotonically increasing **Epoch Number** (e.g., Epoch 5).
    - It sends this number to all JournalNodes (JNs).
    - The JNs will reject any write requests from NameNodes with an Epoch < 5.
    - If the old Active NN (Epoch 4) tries to write, it receives an "Unauthorized" error and immediately shuts itself down (fencing).

<!-- hdfs-site.xml for NameNode HA -->

<!-- HA Configuration -->
<property>
  <name>dfs.nameservices</name>
  <value>mycluster</value>
  <description>Logical name for the cluster</description>
</property>

<property>
  <name>dfs.ha.namenodes.mycluster</name>
  <value>nn1,nn2</value>
  <description>Unique identifiers for each NameNode</description>
</property>

<!-- NameNode RPC addresses -->
<property>
  <name>dfs.namenode.rpc-address.mycluster.nn1</name>
  <value>namenode1.example.com:8020</value>
</property>

<property>
  <name>dfs.namenode.rpc-address.mycluster.nn2</name>
  <value>namenode2.example.com:8020</value>
</property>

<!-- NameNode HTTP addresses -->
<property>
  <name>dfs.namenode.http-address.mycluster.nn1</name>
  <value>namenode1.example.com:50070</value>
</property>

<property>
  <name>dfs.namenode.http-address.mycluster.nn2</name>
  <value>namenode2.example.com:50070</value>
</property>

<!-- Journal Node configuration -->
<property>
  <name>dfs.namenode.shared.edits.dir</name>
  <value>qjournal://jn1.example.com:8485;jn2.example.com:8485;jn3.example.com:8485/mycluster</value>
  <description>URIs of Journal Nodes</description>
</property>

<property>
  <name>dfs.journalnode.edits.dir</name>
  <value>/data/hadoop/journal</value>
  <description>Local path for Journal Node edits</description>
</property>

<!-- Client failover configuration -->
<property>
  <name>dfs.client.failover.proxy.provider.mycluster</name>
  <value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>
</property>

<!-- Automatic failover -->
<property>
  <name>dfs.ha.automatic-failover.enabled</name>
  <value>true</value>
</property>

<!-- Fencing configuration (prevents split-brain) -->
<property>
  <name>dfs.ha.fencing.methods</name>
  <value>
    sshfence
    shell(/bin/true)
  </value>
</property>

<property>
  <name>dfs.ha.fencing.ssh.private-key-files</name>
  <value>/home/hadoop/.ssh/id_rsa</value>
</property>

<!-- ZooKeeper configuration -->
<property>
  <name>ha.zookeeper.quorum</name>
  <value>zk1.example.com:2181,zk2.example.com:2181,zk3.example.com:2181</value>
</property>

// HDFS checksum mechanisms

public class ChecksumVerification {

    /**
     * HDFS Checksum Architecture:
     * - Every 512 bytes of data → 4 bytes CRC32C checksum
     * - Checksums stored separately from data
     * - Verified on read and write
     */

    // 1. Write with checksum verification
    public void writeWithChecksumVerification() throws IOException {
        Configuration conf = new Configuration();

        // Enable checksum verification (enabled by default)
        conf.setBoolean("dfs.client.read.checksum.verify", true);

        FileSystem fs = FileSystem.get(conf);
        Path path = new Path("/data/verified-file.txt");

        // Write data - checksums automatically created
        FSDataOutputStream out = fs.create(path);

        byte[] data = "Important data".getBytes();
        out.write(data);

        // Before close, checksums are computed and stored
        out.close();

        System.out.println("File written with checksums");
    }

    // 2. Read with checksum verification
    public void readWithChecksumVerification() throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path("/data/verified-file.txt");

        FSDataInputStream in = fs.open(path);

        try {
            byte[] buffer = new byte[1024];
            int bytesRead;

            // Each read verifies checksums
            while ((bytesRead = in.read(buffer)) != -1) {
                // If checksum fails, IOException thrown
                processData(buffer, bytesRead);
            }

            System.out.println("File read successfully, all checksums valid");

        } catch (ChecksumException e) {
            // Checksum mismatch detected
            System.err.println("Checksum verification failed: " + e.getMessage());

            // HDFS will automatically try another replica
            // and report corrupted block to NameNode
        } finally {
            in.close();
        }
    }

    // 3. Disable checksum for performance (risky)
    public void readWithoutChecksum() throws IOException {
        Configuration conf = new Configuration();
        conf.setBoolean("dfs.client.read.checksum.verify", false);

        FileSystem fs = FileSystem.get(conf);

        // Raw read without checksum verification
        // Faster but no corruption detection
        FSDataInputStream in = fs.open(new Path("/data/file.txt"));
        // ... read data ...
        in.close();
    }

    // 4. Manual checksum verification
    public void manualChecksumCheck() throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path dataPath = new Path("/data/file.txt");

        // Get checksum file
        FileChecksum checksum = fs.getFileChecksum(dataPath);

        System.out.println("Algorithm: " + checksum.getAlgorithmName());
        System.out.println("Length: " + checksum.getLength());
        System.out.println("Bytes: " +
            java.util.Arrays.toString(checksum.getBytes()));
    }

    // 5. Handle corrupted blocks
    public void handleCorruption() throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path("/data/possibly-corrupt.txt");

        // Set number of retries for checksum failures
        conf.setInt("dfs.client.read.checksum.retries", 3);

        FSDataInputStream in = fs.open(path);

        try {
            byte[] buffer = new byte[4096];
            in.read(buffer);

        } catch (ChecksumException e) {
            System.err.println("Corruption detected: " + e.getPos());

            // HDFS automatically:
            // 1. Tries other replicas
            // 2. Reports bad block to NameNode
            // 3. Schedules re-replication from good replica
            // 4. Marks corrupted replica for deletion
        }
    }

    private void processData(byte[] buffer, int length) {
        // Process the data
    }
}

+---------------------------------------------------------------+
|           MAPREDUCE TASK FAILURE RECOVERY                     |
+---------------------------------------------------------------+
|                                                               |
|  Scenario 1: Map Task Failure                                |
|  ┌────────────────────────────────────────┐                   |
|  │  Node 1: Running Map Task M1           │                   |
|  │  ┌──────────┐                          │                   |
|  │  │  Map M1  │ ─ CRASH! ─► X            │                   |
|  │  │ Progress │                          │                   |
|  │  │   47%    │                          │                   |
|  │  └──────────┘                          │                   |
|  └────────────────────────────────────────┘                   |
|           │                                                   |
|           ▼                                                   |
|  ┌────────────────────────────────────────┐                   |
|  │  ApplicationMaster Detects Failure     │                   |
|  │  • No heartbeat from task              │                   |
|  │  • Task timeout (10 min default)       │                   |
|  │  • Mark task as FAILED                 │                   |
|  └────────────────────────────────────────┘                   |
|           │                                                   |
|           ▼                                                   |
|  ┌────────────────────────────────────────┐                   |
|  │  Retry Decision:                       │                   |
|  │                                        │                   |
|  │  Attempt 1: FAILED                     │                   |
|  │  Max Attempts: 4 (configurable)        │                   |
|  │  Decision: RETRY on different node     │                   |
|  └────────────────────────────────────────┘                   |
|           │                                                   |
|           ▼                                                   |
|  ┌────────────────────────────────────────┐                   |
|  │  Node 2: Retry Map Task M1             │                   |
|  │  ┌──────────┐                          │                   |
|  │  │  Map M1  │ ─ SUCCESS ─► ✓           │                   |
|  │  │ (Attempt │                          │                   |
|  │  │    2)    │                          │                   |
|  │  └──────────┘                          │                   |
|  └────────────────────────────────────────┘                   |
|                                                               |
|  Scenario 2: Reduce Task Failure                             |
|  ┌────────────────────────────────────────┐                   |
|  │  Reduce Task R1 fails at 85%           │                   |
|  │                                        │                   |
|  │  Recovery:                             │                   |
|  │  1. Fetch phase restarts (get map     │                   |
|  │     outputs again)                     │                   |
|  │  2. Sort phase restarts                │                   |
|  │  3. Reduce phase restarts from 0%      │                   |
|  │                                        │                   |
|  │  Note: Must re-fetch all map outputs   │                   |
|  │  (Map outputs may be on failed nodes)  │                   |
|  └────────────────────────────────────────┘                   |
|                                                               |
+---------------------------------------------------------------+

+---------------------------------------------------------------+
|        APPLICATIONMASTER FAILURE & RECOVERY                   |
+---------------------------------------------------------------+
|                                                               |
|  Normal Operation:                                            |
|  ┌──────────────────────────────────────────────────┐         |
|  │           ApplicationMaster (AM)                 │         |
|  │  ┌────────────────────────────────────┐          │         |
|  │  │ Job State:                         │          │         |
|  │  │ • Task tracking                    │          │         |
|  │  │ • Progress monitoring              │          │         |
|  │  │ • Resource negotiation             │          │         |
|  │  │ • Intermediate data locations      │          │         |
|  │  └────────────────────────────────────┘          │         |
|  │              │                                   │         |
|  │              ▼ Checkpoint to HDFS               │         |
|  │  ┌────────────────────────────────────┐          │         |
|  │  │ /tmp/hadoop/job_12345/state.dat    │          │         |
|  │  └────────────────────────────────────┘          │         |
|  └──────────────────────────────────────────────────┘         |
|                      │                                        |
|                      │ CRASH!                                 |
|                      ▼                                        |
|  ┌──────────────────────────────────────────────────┐         |
|  │  ResourceManager Detects AM Failure              │         |
|  │  • No heartbeat received                         │         |
|  │  • Timeout threshold exceeded                    │         |
|  └──────────────────────────────────────────────────┘         |
|                      │                                        |
|                      ▼                                        |
|  ┌──────────────────────────────────────────────────┐         |
|  │  Recovery Process:                               │         |
|  │                                                  │         |
|  │  1. Start new AM on different node               │         |
|  │                                                  │         |
|  │  2. New AM reads checkpoint from HDFS:           │         |
|  │     • Completed tasks → Don't re-run             │         |
|  │     • Running tasks → Re-schedule                │         |
|  │     • Failed tasks → Retry                       │         |
|  │     • Pending tasks → Schedule                   │         |
|  │                                                  │         |
|  │  3. New AM reconstructs job state:               │         |
|  │     • Contact running tasks                      │         |
|  │     • Verify task status                         │         |
|  │     • Re-fetch map output locations              │         |
|  │                                                  │         |
|  │  4. Continue job execution                       │         |
|  │                                                  │         |
|  │  Max AM failures: 2 (configurable)               │         |
|  │  If exceeded → Job FAILED                        │         |
|  └──────────────────────────────────────────────────┘         |
|                                                               |
+---------------------------------------------------------------+

+---------------------------------------------------------------+
|              SPECULATIVE EXECUTION MECHANISM                  |
+---------------------------------------------------------------+
|                                                               |
|  Problem: Slow Tasks (Stragglers)                             |
|  ┌────────────────────────────────────────┐                   |
|  │  10 Map Tasks Running:                 │                   |
|  │                                        │                   |
|  │  Tasks 1-9:  [████████████] 100% ✓    │                   |
|  │  Task 10:    [███░░░░░░░░░]  25%      │                   |
|  │              (Slow disk, high load)    │                   |
|  │                                        │                   |
|  │  Without speculation:                  │                   |
|  │  • Job waits for Task 10               │                   |
|  │  • Entire job delayed                  │                   |
|  │  • Resources underutilized             │                   |
|  └────────────────────────────────────────┘                   |
|           │                                                   |
|           ▼                                                   |
|  ┌────────────────────────────────────────┐                   |
|  │  Detection: AM monitors progress       │                   |
|  │                                        │                   |
|  │  Average task time: 2 minutes          │                   |
|  │  Task 10 time: 5 minutes (still 25%)   │                   |
|  │                                        │                   |
|  │  Heuristic:                            │                   |
|  │  • Task significantly slower than avg  │                   |
|  │  • Not due to different input size     │                   |
|  │  • Resources available for duplicate   │                   |
|  │                                        │                   |
|  │  Decision: Launch speculative copy     │                   |
|  └────────────────────────────────────────┘                   |
|           │                                                   |
|           ▼                                                   |
|  ┌────────────────────────────────────────┐                   |
|  │  Speculative Execution:                │                   |
|  │                                        │                   |
|  │  Node A:  Task 10    [███░░░░░░░░░]   │                   |
|  │           (Original)      25%          │                   |
|  │                                        │                   |
|  │  Node B:  Task 10    [░░░░░░░░░░░░]   │                   |
|  │           (Backup)        0%           │                   |
|  │                                        │                   |
|  │  Both tasks process same input         │                   |
|  │  First to complete wins                │                   |
|  │  Slower task is killed                 │                   |
|  └────────────────────────────────────────┘                   |
|           │                                                   |
|           ▼                                                   |
|  ┌────────────────────────────────────────┐                   |
|  │  Outcome:                              │                   |
|  │                                        │                   |
|  │  Node B completes first: ✓             │                   |
|  │  Node A task killed                    │                   |
|  │                                        │                   |
|  │  Result: Job completes faster          │                   |
|  │  Cost: Extra resource usage            │                   |
|  │  (Usually worth it for job completion) │                   |
|  └────────────────────────────────────────┘                   |
|                                                               |
+---------------------------------------------------------------+

// MapReduce fault tolerance settings

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.MRJobConfig;

public class FaultToleranceConfig {

    public static Job configureFaultTolerance() throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Fault Tolerant Job");

        // ============================================
        // TASK FAILURE HANDLING
        // ============================================

        // Max attempts for map tasks (default: 4)
        conf.setInt(MRJobConfig.MAP_MAX_ATTEMPTS, 4);

        // Max attempts for reduce tasks (default: 4)
        conf.setInt(MRJobConfig.REDUCE_MAX_ATTEMPTS, 4);

        // Max % of failed maps before job fails (default: 0)
        // 0 means any map failure causes job failure
        // Set to 5 = allow 5% of maps to fail
        conf.setInt(MRJobConfig.MAP_FAILURES_MAX_PERCENT, 0);

        // Max % of failed reduces before job fails
        conf.setInt(MRJobConfig.REDUCE_FAILURES_MAXPERCENT, 0);

        // Max failed task attempts before blacklisting node
        conf.setInt(MRJobConfig.MAX_TASK_FAILURES_PER_TRACKER, 3);

        // ============================================
        // TIMEOUT CONFIGURATION
        // ============================================

        // Task timeout (milliseconds)
        // Kill task if no progress for this duration
        // Default: 600000 (10 minutes)
        conf.setLong(MRJobConfig.TASK_TIMEOUT, 600000);

        // Set to 0 to disable timeout (risky)
        // conf.setLong(MRJobConfig.TASK_TIMEOUT, 0);

        // For long-running tasks, increase timeout
        conf.setLong(MRJobConfig.TASK_TIMEOUT, 1800000); // 30 min

        // ============================================
        // SPECULATIVE EXECUTION
        // ============================================

        // Enable speculative execution for maps
        conf.setBoolean(MRJobConfig.MAP_SPECULATIVE, true);

        // Enable speculative execution for reduces
        conf.setBoolean(MRJobConfig.REDUCE_SPECULATIVE, true);

        // Disable speculation for non-idempotent tasks
        // (tasks with side effects)
        // conf.setBoolean(MRJobConfig.MAP_SPECULATIVE, false);
        // conf.setBoolean(MRJobConfig.REDUCE_SPECULATIVE, false);

        // Speculation threshold (default: 1.0)
        // Launch backup if task is this many times slower than average
        conf.setDouble("mapreduce.job.speculative.slowtaskthreshold", 1.0);

        // ============================================
        // APPLICATION MASTER FAILURE
        // ============================================

        // Max AM attempts (default: 2)
        conf.setInt(MRJobConfig.MR_AM_MAX_ATTEMPTS, 2);

        // For critical jobs, allow more AM retries
        conf.setInt(MRJobConfig.MR_AM_MAX_ATTEMPTS, 4);

        // ============================================
        // DATA LOCALITY
        // ============================================

        // Node-local attempts before rack-local
        conf.setInt(MRJobConfig.MAP_MAX_ATTEMPTS, 5);

        // Rack-local attempts before off-rack
        // Helps ensure tasks run near data even with failures

        // ============================================
        // MEMORY SETTINGS (Prevent OOM failures)
        // ============================================

        // Map memory (MB)
        conf.setInt(MRJobConfig.MAP_MEMORY_MB, 2048);

        // Reduce memory (MB)
        conf.setInt(MRJobConfig.REDUCE_MEMORY_MB, 4096);

        // Java heap size (as fraction of container memory)
        conf.setFloat(MRJobConfig.MAP_JAVA_OPTS, 0.8f);
        conf.setFloat(MRJobConfig.REDUCE_JAVA_OPTS, 0.8f);

        // ============================================
        // INTERMEDIATE DATA PROTECTION
        // ============================================

        // Number of fetch failures before blacklisting map output
        conf.setInt(MRJobConfig.SHUFFLE_FETCH_FAILURES, 10);

        // Re-run map tasks if reduce cannot fetch output
        // (map node may have failed)

        // ============================================
        // JOB RECOVERY
        // ============================================

        // Enable job recovery (AM restart)
        conf.setBoolean(MRJobConfig.MR_AM_JOB_RECOVERY_ENABLE, true);

        // Store recovery state in HDFS
        conf.set(MRJobConfig.MR_AM_STAGING_DIR,
                 "/tmp/hadoop-yarn/staging");

        return job;
    }

    /**
     * Example: Critical job with aggressive fault tolerance
     */
    public static Job criticalJobConfig() throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Critical Job");

        // Allow multiple retries
        conf.setInt(MRJobConfig.MAP_MAX_ATTEMPTS, 8);
        conf.setInt(MRJobConfig.REDUCE_MAX_ATTEMPTS, 8);

        // Longer timeout for complex operations
        conf.setLong(MRJobConfig.TASK_TIMEOUT, 3600000); // 1 hour

        // More AM retry attempts
        conf.setInt(MRJobConfig.MR_AM_MAX_ATTEMPTS, 5);

        // Enable speculation
        conf.setBoolean(MRJobConfig.MAP_SPECULATIVE, true);
        conf.setBoolean(MRJobConfig.REDUCE_SPECULATIVE, true);

        // Extra memory to prevent OOM
        conf.setInt(MRJobConfig.MAP_MEMORY_MB, 4096);
        conf.setInt(MRJobConfig.REDUCE_MEMORY_MB, 8192);

        return job;
    }

    /**
     * Example: Fast job with minimal overhead
     */
    public static Job fastJobConfig() throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Fast Job");

        // Fewer retries for fast failure
        conf.setInt(MRJobConfig.MAP_MAX_ATTEMPTS, 2);
        conf.setInt(MRJobConfig.REDUCE_MAX_ATTEMPTS, 2);

        // Shorter timeout
        conf.setLong(MRJobConfig.TASK_TIMEOUT, 300000); // 5 min

        // Disable speculation (save resources)
        conf.setBoolean(MRJobConfig.MAP_SPECULATIVE, false);
        conf.setBoolean(MRJobConfig.REDUCE_SPECULATIVE, false);

        return job;
    }
}

+---------------------------------------------------------------+
|           YARN NODEMANAGER FAILURE HANDLING                   |
+---------------------------------------------------------------+
|                                                               |
|  Healthy Cluster:                                             |
|  ┌─────────────────────────────────────────┐                  |
|  │         ResourceManager (RM)            │                  |
|  │  ┌───────────────────────────┐          │                  |
|  │  │ Node Health Tracker       │          │                  |
|  │  │ • Monitors all NMs        │          │                  |
|  │  │ • Heartbeat every 1 sec   │          │                  |
|  │  └───────────────────────────┘          │                  |
|  └──────────┬──────────────────────────────┘                  |
|             │                                                 |
|        Heartbeats                                             |
|             │                                                 |
|    ┌────────┼────────┬───────────┐                            |
|    ▼        ▼        ▼           ▼                            |
|  ┌────┐  ┌────┐  ┌────┐      ┌────┐                          |
|  │ NM │  │ NM │  │ NM │  ... │ NM │                          |
|  │ 1  │  │ 2  │  │ 3  │      │ N  │                          |
|  └────┘  └────┘  └────┘      └────┘                          |
|    ✓       ✓       ✗           ✓                             |
|                    │                                          |
|                  CRASH!                                       |
|                    │                                          |
|                    ▼                                          |
|  ┌─────────────────────────────────────────┐                  |
|  │  RM Detection (NM3 failed):             │                  |
|  │                                         │                  |
|  │  1. No heartbeat for 10 minutes         │                  |
|  │     (dfs.heartbeat.interval × 10)       │                  |
|  │                                         │                  |
|  │  2. Mark NM3 as LOST                    │                  |
|  │                                         │                  |
|  │  3. Identify affected containers:       │                  |
|  │     • Container C1 (AM for Job1)        │                  |
|  │     • Container C2 (Map task)           │                  |
|  │     • Container C3 (Reduce task)        │                  |
|  └─────────────────────────────────────────┘                  |
|                    │                                          |
|                    ▼                                          |
|  ┌─────────────────────────────────────────┐                  |
|  │  Recovery Actions:                      │                  |
|  │                                         │                  |
|  │  For ApplicationMaster (C1):            │                  |
|  │  ├─ Restart AM on different node        │                  |
|  │  ├─ Restore state from checkpoint       │                  |
|  │  └─ Continue job execution              │                  |
|  │                                         │                  |
|  │  For Map Task (C2):                     │                  |
|  │  ├─ Mark task as FAILED                 │                  |
|  │  ├─ Re-schedule on healthy node         │                  |
|  │  └─ Restart from beginning              │                  |
|  │                                         │                  |
|  │  For Reduce Task (C3):                  │                  |
|  │  ├─ Mark task as FAILED                 │                  |
|  │  ├─ Re-schedule on healthy node         │                  |
|  │  ├─ Re-fetch map outputs                │                  |
|  │  └─ Restart reduce phase                │                  |
|  └─────────────────────────────────────────┘                  |
|                                                               |
+---------------------------------------------------------------+

+---------------------------------------------------------------+
|          RESOURCEMANAGER HA ARCHITECTURE                      |
+---------------------------------------------------------------+
|                                                               |
|  Active RM                    Standby RM                      |
|  ┌─────────────────┐         ┌─────────────────┐             |
|  │  • Serves clients│         │  • Monitors     │             |
|  │  • Schedules     │         │  • Warm standby │             |
|  │  • Tracks apps   │         │  • Ready to take│             |
|  │  • Writes state  │         │    over         │             |
|  └────────┬────────┘         └────────┬────────┘             |
|           │                           │                      |
|           │   Write State    Read State                      |
|           │                           │                      |
|           ▼                           ▼                      |
|  ┌──────────────────────────────────────────┐                |
|  │      State Store (ZooKeeper)             │                |
|  │  ┌────────────────────────────────┐      │                |
|  │  │ Application State:             │      │                |
|  │  │ • Running applications         │      │                |
|  │  │ • Completed applications       │      │                |
|  │  │ • Node membership              │      │                |
|  │  │ • Security tokens              │      │                |
|  │  │ • Delegation tokens            │      │                |
|  │  └────────────────────────────────┘      │                |
|  └──────────────────────────────────────────┘                |
|           │                           │                      |
|           │                           │                      |
|           └───────────┬───────────────┘                      |
|                       ▼                                      |
|           ┌──────────────────────┐                           |
|           │  ZooKeeper Ensemble  │                           |
|           │  ┌────┐ ┌────┐ ┌────┐│                           |
|           │  │ ZK │ │ ZK │ │ ZK ││                           |
|           │  │ 1  │ │ 2  │ │ 3  ││                           |
|           │  └────┘ └────┘ └────┘│                           |
|           │  (Active RM election)│                           |
|           └──────────────────────┘                           |
|                                                               |
|  Failover Process:                                            |
|  1. Active RM fails                                           |
|  2. ZK detects missing heartbeat                              |
|  3. Standby RM promoted to Active                             |
|  4. New Active loads state from ZK                            |
|  5. Clients automatically redirect                            |
|  6. Jobs continue without interruption                        |
|                                                               |
|  Failover time: < 1 minute                                    |
|                                                               |
+---------------------------------------------------------------+

<!-- yarn-site.xml for ResourceManager HA -->

<!-- Enable RM HA -->
<property>
  <name>yarn.resourcemanager.ha.enabled</name>
  <value>true</value>
</property>

<!-- RM cluster ID -->
<property>
  <name>yarn.resourcemanager.cluster-id</name>
  <value>yarn-cluster</value>
</property>

<!-- RM IDs -->
<property>
  <name>yarn.resourcemanager.ha.rm-ids</name>
  <value>rm1,rm2</value>
</property>

<!-- RM1 addresses -->
<property>
  <name>yarn.resourcemanager.hostname.rm1</name>
  <value>resourcemanager1.example.com</value>
</property>

<property>
  <name>yarn.resourcemanager.address.rm1</name>
  <value>resourcemanager1.example.com:8032</value>
</property>

<property>
  <name>yarn.resourcemanager.scheduler.address.rm1</name>
  <value>resourcemanager1.example.com:8030</value>
</property>

<property>
  <name>yarn.resourcemanager.webapp.address.rm1</name>
  <value>resourcemanager1.example.com:8088</value>
</property>

<!-- RM2 addresses -->
<property>
  <name>yarn.resourcemanager.hostname.rm2</name>
  <value>resourcemanager2.example.com</value>
</property>

<property>
  <name>yarn.resourcemanager.address.rm2</name>
  <value>resourcemanager2.example.com:8032</value>
</property>

<property>
  <name>yarn.resourcemanager.scheduler.address.rm2</name>
  <value>resourcemanager2.example.com:8030</value>
</property>

<property>
  <name>yarn.resourcemanager.webapp.address.rm2</name>
  <value>resourcemanager2.example.com:8088</value>
</property>

<!-- ZooKeeper configuration -->
<property>
  <name>yarn.resourcemanager.zk-address</name>
  <value>zk1.example.com:2181,zk2.example.com:2181,zk3.example.com:2181</value>
</property>

<property>
  <name>yarn.resourcemanager.zk-state-store.parent-path</name>
  <value>/rmstore</value>
</property>

<!-- State store configuration -->
<property>
  <name>yarn.resourcemanager.store.class</name>
  <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore</value>
</property>

<!-- Enable automatic failover -->
<property>
  <name>yarn.resourcemanager.ha.automatic-failover.enabled</name>
  <value>true</value>
</property>

<!-- Embedded leader election -->
<property>
  <name>yarn.resourcemanager.ha.automatic-failover.embedded</name>
  <value>true</value>
</property>

<!-- Client failover -->
<property>
  <name>yarn.client.failover-proxy-provider</name>
  <value>org.apache.hadoop.yarn.client.ConfiguredRMFailoverProxyProvider</value>
</property>

<!-- Retry settings -->
<property>
  <name>yarn.client.failover-max-attempts</name>
  <value>15</value>
</property>

<property>
  <name>yarn.client.failover-sleep-base-ms</name>
  <value>500</value>
</property>

<property>
  <name>yarn.client.failover-sleep-max-ms</name>
  <value>2000</value>
</property>

<!-- Work-preserving RM restart -->
<property>
  <name>yarn.resourcemanager.recovery.enabled</name>
  <value>true</value>
</property>

<property>
  <name>yarn.resourcemanager.work-preserving-recovery.enabled</name>
  <value>true</value>
</property>

<!-- NodeManager recovery -->
<property>
  <name>yarn.nodemanager.recovery.enabled</name>
  <value>true</value>
</property>

<property>
  <name>yarn.nodemanager.recovery.dir</name>
  <value>/var/log/hadoop-yarn/nodemanager/recovery-state</value>
</property>

// Designing idempotent MapReduce operations

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import java.io.IOException;
import java.util.*;

/**
 * Idempotent operations produce the same result
 * when executed multiple times.
 * Critical for fault tolerance with task retries.
 */
public class IdempotentDesign {

    // ============================================
    // GOOD: Idempotent Mapper
    // ============================================

    public static class IdempotentMapper
            extends Mapper<LongWritable, Text, Text, IntWritable> {

        private final IntWritable one = new IntWritable(1);
        private Text word = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            // Parse input
            String line = value.toString();
            String[] words = line.split("\\s+");

            // Emit each word
            for (String w : words) {
                word.set(w.toLowerCase());
                context.write(word, one);
            }

            // ✓ Idempotent: Same input → Same output
            // ✓ No side effects
            // ✓ No external state modification
            // ✓ Safe to retry
        }
    }

    // ============================================
    // GOOD: Idempotent Reducer
    // ============================================

    public static class IdempotentReducer
            extends Reducer<Text, IntWritable, Text, IntWritable> {

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values,
                Context context) throws IOException, InterruptedException {

            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }

            context.write(key, new IntWritable(sum));

            // ✓ Idempotent: Same input values → Same sum
            // ✓ No external dependencies
            // ✓ Safe to retry
        }
    }

    // ============================================
    // BAD: Non-Idempotent Mapper (Database writes)
    // ============================================

    public static class NonIdempotentMapper
            extends Mapper<LongWritable, Text, Text, IntWritable> {

        private java.sql.Connection dbConnection;

        @Override
        protected void setup(Context context) throws IOException {
            // Connect to database
            try {
                dbConnection = java.sql.DriverManager.getConnection(
                    "jdbc:mysql://db.example.com/analytics",
                    "user", "password"
                );
            } catch (Exception e) {
                throw new IOException("DB connection failed", e);
            }
        }

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            try {
                // ✗ BAD: Writing to database
                // If task retries, duplicate records created!
                String sql = "INSERT INTO events (data) VALUES (?)";
                java.sql.PreparedStatement stmt =
                    dbConnection.prepareStatement(sql);
                stmt.setString(1, value.toString());
                stmt.executeUpdate();

                // ✗ Non-idempotent operation
                // ✗ Side effects
                // ✗ Task retry creates duplicates

            } catch (Exception e) {
                throw new IOException("DB write failed", e);
            }
        }
    }

    // ============================================
    // FIXED: Make Database Writes Idempotent
    // ============================================

    public static class IdempotentDatabaseMapper
            extends Mapper<LongWritable, Text, Text, IntWritable> {

        private java.sql.Connection dbConnection;

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            try {
                // ✓ Use unique key from input
                String recordId = key.toString(); // File offset is unique
                String data = value.toString();

                // ✓ Use REPLACE or INSERT ... ON DUPLICATE KEY UPDATE
                // This makes operation idempotent
                String sql =
                    "REPLACE INTO events (record_id, data) VALUES (?, ?)";

                java.sql.PreparedStatement stmt =
                    dbConnection.prepareStatement(sql);
                stmt.setString(1, recordId);
                stmt.setString(2, data);
                stmt.executeUpdate();

                // ✓ Idempotent: Retry produces same result
                // ✓ No duplicates on retry

            } catch (Exception e) {
                throw new IOException("DB write failed", e);
            }
        }
    }

    // ============================================
    // BAD: Non-Idempotent Counter Update
    // ============================================

    public static class NonIdempotentCounterMapper
            extends Mapper<LongWritable, Text, Text, IntWritable> {

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            // ✗ BAD: Increment counter in external system
            // If task retries, counter incremented multiple times!
            incrementExternalCounter("records_processed");

            // Process record...
            context.write(new Text("result"), new IntWritable(1));
        }

        private void incrementExternalCounter(String name) {
            // External counter increment (Redis, Memcached, etc.)
            // ✗ Non-idempotent
        }
    }

    // ============================================
    // GOOD: Use Hadoop Counters (Idempotent)
    // ============================================

    public static class IdempotentCounterMapper
            extends Mapper<LongWritable, Text, Text, IntWritable> {

        enum RecordCounters {
            PROCESSED,
            SKIPPED,
            ERRORS
        }

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            try {
                // Process record
                String result = processRecord(value.toString());

                // ✓ Use Hadoop counters
                // Framework handles idempotency during task retries
                context.getCounter(RecordCounters.PROCESSED).increment(1);

                context.write(new Text(result), new IntWritable(1));

            } catch (Exception e) {
                context.getCounter(RecordCounters.ERRORS).increment(1);
            }

            // ✓ Counters are automatically deduplicated on task retry
            // ✓ Idempotent behavior
        }

        private String processRecord(String record) {
            return record.toUpperCase();
        }
    }
}