NameNode high availability (QJM and ZooKeeper)

The day the NameNode went down

NameNode high availability

Words you'll meet in this lesson

Settings and numbers from our lab (Hadoop 3.5.0)

HA at three levels

Ways to protect the NameNode

Failover modes and admin commands

Building and breaking a real HA cluster

Our lab: ZooKeeper, 3 JournalNodes, NameNodes nn1 and nn2 (each with a ZKFC) and 3 DataNodes, all apache/hadoop:3.5.0 containers. The full files are in “Run it on your PC”.

1. The HA configuration

   CORE-SITE.XML_fs.defaultFS=hdfs://canteen
   CORE-SITE.XML_ha.zookeeper.quorum=zk:2181
   HDFS-SITE.XML_dfs.nameservices=canteen
   HDFS-SITE.XML_dfs.ha.namenodes.canteen=nn1,nn2
   HDFS-SITE.XML_dfs.namenode.rpc-address.canteen.nn1=nn1:8020
   HDFS-SITE.XML_dfs.namenode.shared.edits.dir=qjournal://jn1:8485;jn2:8485;jn3:8485/canteen
   HDFS-SITE.XML_dfs.client.failover.proxy.provider.canteen=org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider
   HDFS-SITE.XML_dfs.ha.automatic-failover.enabled=true
   HDFS-SITE.XML_dfs.ha.fencing.methods=shell(/bin/true)

   The client sees one name, canteen, with two possible addresses. shell(/bin/true) is a “fencer that always succeeds”, acceptable only in a lab; we'll see why QJM keeps us safe anyway.

2. First start: format, bootstrap, elect

   # on nn1, once:
   hdfs namenode -format -nonInteractive -clusterId canteen-cluster
   hdfs zkfc -formatZK -nonInteractive
   # on nn2, once:
   hdfs namenode -bootstrapStandby -nonInteractive
   # on both:
   hdfs --daemon start zkfc; hdfs namenode
   
   hdfs haadmin -getAllServiceState

   nn1:8020 active nn2:8020 standby
   nn2 copies nn1's fsimage; both ZKFCs race for the lock znode in ZooKeeper, and nn1 won. ZooKeeper now holds /hadoop-ha/canteen/ActiveStandbyElectorLock and ActiveBreadCrumb (we listed them with the ZooKeeper client).

3. Crash the active while a client writes

   docker kill lab7-nn1-1

   15:44:15 before: nn1:8020=active nn2:8020=standby 15:44:25 >>> docker kill lab7-nn1-1 (the active NameNode crashes) 15:44:27 nn2 is standby ... 15:44:34 nn2 is standby 15:44:36 nn2 is active 15:44:24 write 3 ok (took 3 s) 15:44:46 write 4 ok (took 22 s) 15:44:48 write 5 ok (took 2 s) ... 15:45:03 write 12 ok (took 2 s) 15:45:04 files written: 12
   About 10 s to fail over: the ZooKeeper session timeout. The client's write 4 was in progress: it tried nn1 (20 s connect timeout to the dead address), then nn2, and succeeded. 12 of 12 writes succeeded; the application only saw one slow write.

4. What the standby's ZKFC did

   grep … /var/log/hadoop/hadoop-hadoop-zkfc-nn2.out

   15:44:34 ActiveStandbyElector - Checking for any old active which needs to be fenced... 15:44:35 NodeFencer - ====== Beginning Service Fencing Process... ====== 15:44:35 NodeFencer - ====== Fencing successful by method ShellCommandFencer(/bin/true) ====== 15:44:35 ZKFailoverController - Trying to make NameNode at nn2/172.20.0.7:8020 active... 15:44:35 ZKFailoverController - Successfully transitioned NameNode at nn2/172.20.0.7:8020 to active state
   Lock taken → old active fenced → standby promoted, all within about a second. Before becoming active, nn2 reads any remaining edits from the JournalNodes, so no change is lost.

5. Freeze the active (the zombie test)

   docker pause lab7-nn1-1 ; sleep 25 ; docker unpause lab7-nn1-1

   15:40:49 before : nn1:8020=active nn2:8020=standby | epochs: jn1=7 jn2=7 jn3=7 15:41:16 during : nn2=active | epochs: jn1=8 jn2=8 jn3=8 15:41:19 client write to the new active: ok 15:41:19 >>> docker unpause lab7-nn1-1 (the old active wakes up, still thinking it is active) 172.20.0.5:8485: IPC's epoch 7 is less than the last promised epoch 8 ; journal id: canteen 172.20.0.4:8485: IPC's epoch 7 is less than the last promised epoch 8 ; journal id: canteen 172.20.0.2:8485: IPC's epoch 7 is less than the last promised epoch 8 ; journal id: canteen Exiting with status 1: Error: flush failed for required journal ... SHUTDOWN_MSG: Shutting down NameNode at nn1/172.20.0.6 15:41:39 after : nn1:8020=Failed nn2:8020=active
   A frozen process (like a long garbage-collection pause) is the dangerous case: it doesn't know it was replaced. When nn1 woke and tried to write with epoch 7, all three JournalNodes refused, and nn1 shut itself down. Split-brain prevented, even with a fencer that does nothing.

6. Lose the JournalNode majority

   docker stop lab7-jn3-1 ; … ; docker stop lab7-jn2-1

   15:45:56 nn1:8020=standby nn2:8020=active 15:46:00 write 1: ok 15:46:00 >>> stop jn3 (2 of 3 JournalNodes left) 15:46:02 write 2: ok 15:46:02 >>> stop jn2 (1 of 3 left: no majority) 15:47:32 write 3: FAILED (no answer) 15:47:42 nn1:8020=Failed nn2:8020=Failed nn1: java.io.IOException: Timed out waiting 20000ms for a quorum of nodes to respond. nn2: java.io.IOException: Timed out waiting 20000ms for a quorum of nodes to respond.
   With 2 of 3 JournalNodes, everything works. With 1 of 3, the active can't get a majority for 20 s and stops; the standby can't become active either. HDFS stops rather than risk losing edits. Starting the JournalNodes and NameNodes again (restore.sh) brought the cluster back with all files.

Crash, takeover, and the zombie that wakes up

Six steps from our lab. Left: what admins and clients see. Right: ZooKeeper, the ZKFCs and the JournalNodes.

Try it yourself: failover in the playground

This playground has two NameNodes. Upload a file, stop the active NameNode, and watch the standby take over.

HA at work

Simplified; organisations differ in tools and policies.

A quorum journal with epochs, in Python

Three JournalNodes each remember the highest epoch they have promised. A NameNode becomes the writer by getting a majority to promise its new epoch; an edit is accepted only by JNs that have not promised a higher one. Run it to replay our zombie test.

jns = {"jn1": 7, "jn2": 7, "jn3": 7}          # last promised epoch on each JournalNode
log = []

def become_active(name):
    epoch = max(jns.values()) + 1
    ok = [j for j in jns if jns[j] < epoch]
    if len(ok) >= 2:                          # majority of 3
        for j in ok:
            jns[j] = epoch
        print(f"{name} is ACTIVE with epoch {epoch}")
        return epoch
    print(f"{name} could not get a majority")

def write(name, epoch, edit):
    accepted = [j for j in jns if epoch >= jns[j]]
    if len(accepted) >= 2:
        log.append(edit)
        print(f"{name} wrote {edit!r} (accepted by {len(accepted)} JNs)")
    else:
        print(f"{name} REJECTED: epoch {epoch} is less than promised {max(jns.values())}")

e1 = 7                                        # nn1 was active with epoch 7, then froze
e2 = become_active("nn2")                     # nn2 takes over
write("nn2", e2, "mkdir /results")
write("nn1", e1, "delete /results")           # the zombie wakes up
print("edit log:", log)

• epoch = max(jns.values()) + 1A new writer always asks for a higher epoch than any JournalNode has seen. Our lab epochs went 7 → 8.
• accepted = [j for j in jns if epoch >= jns[j]]A JournalNode refuses any write whose epoch is lower than what it promised: “IPC's epoch 7 is less than the last promised epoch 8”.
• if len(accepted) >= 2:A write needs a majority (2 of 3). That is why losing two JournalNodes stopped the whole cluster.

Run it on your PC: build and break an HA cluster

Create a folder hdfs-ha-lab with four files from the lesson's lab: config (the HA settings shown in Part 4), docker-compose.yaml, nn.sh and zk.sh (ZooKeeper runs from the jars inside the Hadoop image, so no extra image is needed). Start it with docker compose -p lab7 up -d --scale datanode=3, wait a minute, then run the three experiments. It needs about 4 GB of free RAM.

x-hadoop: &hadoop
  image: apache/hadoop:3.5.0
  env_file: [./config]
  volumes: [./nn.sh:/opt/nn.sh:ro, ./zk.sh:/opt/zk.sh:ro]
services:
  zk:  { <<: *hadoop, hostname: zk,  command: ["bash", "/opt/zk.sh"] }
  jn1: { <<: *hadoop, hostname: jn1, command: ["hdfs", "journalnode"] }
  jn2: { <<: *hadoop, hostname: jn2, command: ["hdfs", "journalnode"] }
  jn3: { <<: *hadoop, hostname: jn3, command: ["hdfs", "journalnode"] }
  nn1: { <<: *hadoop, hostname: nn1, command: ["bash", "/opt/nn.sh"], depends_on: [zk, jn1, jn2, jn3] }
  nn2: { <<: *hadoop, hostname: nn2, command: ["bash", "/opt/nn.sh"], depends_on: [nn1] }
  datanode: { <<: *hadoop, command: ["hdfs", "datanode"], depends_on: [nn1, nn2] }

# ---- nn.sh ----
#!/bin/bash
# Start a NameNode plus its ZKFailoverController in one container.
# nn1 formats the namespace; nn2 copies it from nn1 (bootstrapStandby).
NAME=/tmp/hadoop-hadoop/dfs/name
if [ ! -d $NAME/current ]; then
  if [ "$(hostname)" = "nn1" ]; then
    hdfs namenode -format -nonInteractive -clusterId canteen-cluster
    hdfs zkfc -formatZK -nonInteractive
  else
    until hdfs namenode -bootstrapStandby -nonInteractive; do sleep 5; done
  fi
fi
hdfs --daemon start zkfc
exec hdfs namenode

# ---- zk.sh ----
#!/bin/bash
# Run a single ZooKeeper server using the ZooKeeper jars shipped inside the Hadoop image.
mkdir -p /tmp/zk && printf 'tickTime=2000\ndataDir=/tmp/zk\nclientPort=2181\nadmin.enableServer=false\n4lw.commands.whitelist=*\n' > /tmp/zk/zoo.cfg
exec java -cp "/opt/hadoop/share/hadoop/common/lib/*" org.apache.zookeeper.server.quorum.QuorumPeerMain /tmp/zk/zoo.cfg

# Start:  docker compose -p lab7 up -d --scale datanode=3

#!/usr/bin/env bash
# ha_failover.sh: crash the active NameNode while a client keeps writing, and watch HDFS fail over.
# Run from the folder with docker-compose.yaml (project name lab7).
P=lab7
state() { docker exec $P-nn2-1 hdfs haadmin -getAllServiceState 2>/dev/null | awk '/^nn[12]:8020/{printf "%s=%s  ", $1, $2}'; }
echo "$(date +%T) before: $(state)"

# a client on datanode-1 writes one small file every loop, and logs success or failure
docker exec -d $P-datanode-1 bash -c 'for i in $(seq 1 12); do
  start=$(date +%s)
  if echo "order $i" | hdfs dfs -put -f - /orders_$i.txt >/tmp/err_$i 2>&1; then r=ok; else r=FAILED; fi
  echo "$(date +%T) write $i $r  (took $(( $(date +%s) - start )) s)"
done > /tmp/writer.log 2>&1'
sleep 8
echo "$(date +%T) >>> docker kill $P-nn1-1   (the active NameNode crashes)"
docker kill $P-nn1-1 > /dev/null
for t in $(seq 1 30); do
  s=$(docker exec $P-nn2-1 hdfs haadmin -getServiceState nn2 2>/dev/null)
  echo "$(date +%T) nn2 is $s"
  [ "$s" = "active" ] && break
done
until [ "$(docker exec $P-datanode-1 bash -c 'wc -l < /tmp/writer.log')" -ge 12 ]; do sleep 2; done
echo "--- client log (datanode-1)"
docker exec $P-datanode-1 cat /tmp/writer.log
echo "--- ZKFC log on nn2"
docker exec $P-nn2-1 bash -c 'grep -hE "Checking for any old active|Beginning Service Fencing|Fencing successful|Trying to make|Successfully transitioned" /var/log/hadoop/*zkfc*.out | cut -c1-150 | tail -5'
echo "$(date +%T) files written: $(docker exec $P-nn2-1 hdfs dfs -ls / 2>/dev/null | grep -c orders_)"

# Run:  bash ha_failover.sh   (then: docker start lab7-nn1-1)

#!/usr/bin/env bash
# zombie.sh: freeze the active NameNode (it is alive but silent), let the other take over,
# then wake the old one up and watch the JournalNodes refuse its writes.
P=lab7
state() { docker exec $P-nn2-1 hdfs haadmin -getAllServiceState 2>/dev/null | awk '/^nn[12]:8020/{printf "%s=%s  ", $1, $2}'; }
epoch() { for j in 1 2 3; do printf "jn$j=%s " "$(docker exec $P-jn$j-1 cat /tmp/hadoop-hadoop/dfs/journal/canteen/current/last-promised-epoch)"; done; }
echo "$(date +%T) before : $(state)| epochs: $(epoch)"
echo "$(date +%T) >>> docker pause $P-nn1-1   (active NameNode frozen, e.g. a 60 s GC pause)"
docker pause $P-nn1-1 > /dev/null
sleep 25
echo "$(date +%T) during : nn2=$(docker exec $P-nn2-1 hdfs haadmin -getServiceState nn2 2>/dev/null) | epochs: $(epoch)"
echo "order after takeover" | docker exec -i $P-datanode-1 hdfs dfs -put -f - /after_takeover.txt 2>/dev/null && echo "$(date +%T) client write to the new active: ok"
echo "$(date +%T) >>> docker unpause $P-nn1-1   (the old active wakes up, still thinking it is active)"
docker unpause $P-nn1-1 > /dev/null
sleep 20
echo "--- old active nn1, last log lines"
docker logs $P-nn1-1 2>&1 | grep -E "epoch|IPC's epoch|FATAL|Error: flush|shutting down|SHUTDOWN_MSG|Transitioning to standby|standby" | cut -c1-190 | tail -6
echo "$(date +%T) after  : $(state)"

# Run:  bash zombie.sh   (then: docker start lab7-nn1-1)

#!/usr/bin/env bash
# jn_quorum.sh: the active NameNode needs a majority (2 of 3) of JournalNodes to write edits.
P=lab7
put() { if echo "x" | timeout 90 docker exec -i $P-datanode-1 hdfs dfs -put -f - /q_$1.txt > /dev/null 2>&1; then echo "$(date +%T) write $1: ok"; else echo "$(date +%T) write $1: FAILED (no answer)"; fi; }
state() { docker exec $P-datanode-1 hdfs haadmin -getAllServiceState 2>/dev/null | awk '/^nn[12]:8020/{printf "%s=%s  ", $1, $2}'; }
echo "$(date +%T) $(state)"
put 1
echo "$(date +%T) >>> stop jn3 (2 of 3 JournalNodes left)"; docker stop $P-jn3-1 > /dev/null
put 2
echo "$(date +%T) >>> stop jn2 (1 of 3 left: no majority)"; docker stop $P-jn2-1 > /dev/null
put 3
sleep 10
echo "$(date +%T) $(state)"
for n in nn1 nn2; do docker logs $P-$n-1 2>&1 | grep -E "Timed out waiting|SHUTDOWN_MSG: Shutting" | tail -2 | cut -c1-120 | sed "s/^/$n: /"; done

# Run:  bash jn_quorum.sh   then restore with:  bash restore.sh

An election in ZooKeeper

Two ZKFCs try to create the same ephemeral lock znode. The first wins; when its session expires the znode vanishes and the other wins.

Worked examples: real problems, solved and traced in memory

Three HA calculations: failover time, JournalNode sizing, and what an hour of NameNode downtime costs. Each runs live and is traced in memory.

15:44:25 · 15:44:34 · 15:44:35

detection: 9 s   takeover: 1 s   total: 10 s

p(down) = 0.01

3 JNs: majority lost 0.029800% of the time
5 JNs: majority lost 0.000985% of the time

55 min · 300 × ₹900/h · 40 × ₹15,000 · ₹6 lakh

one outage costs about Rs 847,500
HA pays for itself after 1 outage(s)

Check yourself

Quick revision

Interview and viva questions

Practice problems

HA exercises using the real timestamps, epochs and logs from the lab.

3, 5

3 JNs tolerate 1 failure
5 JNs tolerate 2 failures

for n in [3, 5]:
    t = n - (n // 2 + 1)
    print(n, "JNs tolerate", t, "failure" if t == 1 else "failures")

two lines

['nn2:8020']

out = """nn1:8020                                           standby
nn2:8020                                           active"""
print([l.split()[0] for l in out.splitlines() if l.split()[1] == "active"])

2 → 8

6 new active writers

print(8 - 2, "new active writers")

write lines

15:44:46 write 4 ok  (took 22 s)

log = ["15:44:19 write 1 ok  (took 2 s)", "15:44:24 write 3 ok  (took 3 s)", "15:44:46 write 4 ok  (took 22 s)", "15:44:48 write 5 ok  (took 2 s)"]
slow = max(log, key=lambda l: int(l.split("took ")[1].split()[0]))
print(slow)

timestamps and states

state change at 15:44:40 standby -> active

samples = [("15:44:25", "standby"), ("15:44:30", "standby"), ("15:44:35", "standby"), ("15:44:40", "active")]
prev = None
for t, s in samples:
    if prev and s != prev:
        print("state change at", t, prev, "->", s)
    prev = s

[True, False, True]

True False

def can_write(up):
    return sum(up) > len(up) // 2
print(can_write([True, False, True]), can_write([True, False, False]))

12 × 20 min × 300 × ₹900/h

Rs 1,080,000 per year

print(f"Rs {12 * 20 / 60 * 300 * 900:,.0f} per year")

3 JNs at epoch 4

nn1: 3 promises -> ACTIVE
nn2: 0 promises -> refused

jns = [4, 4, 4]
def propose(name, epoch):
    yes = 0
    for i, e in enumerate(jns):
        if epoch > e:
            jns[i] = epoch; yes += 1
    return f"{name}: {yes} promises -> {'ACTIVE' if yes >= 2 else 'refused'}"
print(propose("nn1", 5))
print(propose("nn2", 5))

Real-world scenario questions (asked in company interviews)

HA is a favourite topic for Hadoop administrator and senior data engineer interviews.