Kafka على أكتر من zone و region: الـ replication والـ ISR والـ acks والـ MirrorMaker والـ failover

أهلًا بيك. ده آخر درس في القسم ده. عندنا الـ payments كلها ماشية على Kafka، وكان كله تمام. لحد يوم الـ zone اللي فيها الـ cluster وقعت. بص على kafka dot naive dot js. الستة brokers في zone واحدة. الـ zone وقعت، الـ cluster كله وقع، ومش مهم عندك كام نسخة. الـ replicationFactor اتنين، يعني broker واحد بس يقع وانت على نسخة واحدة. الـ producer بـ acks واحد: الـ leader بيقول تمام قبل ما حد ياخد نسخة. الـ leader وقع، والرسالة اللي الـ user شاف إنها اتدفعت راحت. min.insync.replicas بواحد، فـ acks all نفسها مش هتحميك. الـ unclean leader election شغّال، فـ broker متأخر ممكن يبقى leader ويمسح رسايل. والـ region التانية؟ backup كل يوم. ولو عملنا failover، الـ consumers بيبدأوا من الأول، وكل payment يتعالج تاني.

نبدأ من الأساس. الـ topic متقسّم partitions، وكل partition هو log: رسايل ورا بعض، وكل رسالة ليها offset. والـ partition ليه نسخ على أكتر من broker، وده الـ replicationFactor. واحد منهم الـ leader، وكل الـ writes والـ reads بتروحله. والباقيين followers، بيفضلوا يسحبوا من الـ leader. الـ ISR، يعني in sync replicas: الـ followers اللي ماشيين ورا الـ leader ومش متأخرين. الـ follower لو ما سحبش لمدة replica dot lag dot time dot max dot ms، اللي الـ default بتاعها تلاتين تانية، بيطلع برّه الـ ISR. ولما يلحق، بيرجع. والـ ISR ده هو اللي بيتختار منه الـ leader الجديد لما الـ leader يقع. علشان كده هو أهم حاجة في الدرس.

طب الـ producer بيعرف إمتى إن الرسالة اتحفظت؟ ده الـ acks. acks صفر: بيبعت وخلاص. ولا بيعرف وصلت ولا لأ. ينفع للـ metrics، مش للـ payments. acks واحد: الـ leader كتبها عنده ورد. الـ followers لسه ما خدوهاش. لو الـ leader وقع في الوقت ده، الـ leader الجديد مش عنده الرسالة، والرسالة ضاعت مع إنها أخدت ack. acks all: الـ leader بيستنى كل الـ ISR ياخدوا نسخة، وبعدين يرد. وهنا الـ high watermark، أو الـ hw: آخر offset كل الـ ISR عندهم. والـ consumer بيقرا لحد الـ hw بس، مش لآخر الـ log. ليه؟ علشان ما يقراش رسالة ممكن تختفي لو الـ leader وقع. يعني مع acks واحد، الرسالة مكتوبة عند الـ leader، بس الـ consumer مش شايفها لحد ما الـ followers ياخدوها.

بس acks all لوحدها فيها مشكلة. لو الـ followers الاتنين وقعوا، الـ ISR بقى فيه الـ leader بس. و acks all معناها كل الـ ISR، يعني الـ leader لوحده. فرجعنا لـ acks واحد من غير ما نحس. الحل min dot insync dot replicas. لو اتنين، الـ produce بـ acks all بيترفض بـ NOT_ENOUGH_REPLICAS لو الـ ISR أقل من اتنين. يعني بنفضّل نرفض الـ write ونقول للـ producer جرّب تاني، على إننا نقول تمام والرسالة على broker واحد. والتركيبة اللي بتشوفها في أي setup جد: replicationFactor تلاتة، و min insync اتنين، و acks all. كده broker واحد يقع وانت لسه بتكتب عادي، واتنين يقعوا تقف الـ writes بس ما تخسرش حاجة. ومعاهم enable dot idempotence بـ true، وده الـ default في الـ versions الجديدة. الـ producer بيبعت مع كل رسالة producer id و sequence number، فلو عمل retry بعد timeout، الـ broker يعرف إنها نفس الرسالة وما يكتبهاش مرتين.

طب لو كل الـ ISR وقعوا، وفيه broker تاني شغّال بس متأخر؟ ده الـ unclean leader election. لو مقفول، وده الـ default: الـ partition يبقى offline. مفيش reads ولا writes، لحد ما واحد من الـ ISR يرجع. availability أقل، بس ولا رسالة ضاعت. لو مفتوح: الـ broker المتأخر يبقى leader على طول. والرسايل اللي ما كانتش عنده ضاعت، وحتى لو الـ leader القديم رجع، هيمسحها علشان يمشي على الـ leader الجديد. للـ payments: مقفول دايمًا. ممكن تفتحه لـ topic زي الـ logs أو الـ clicks، اللي فيه الـ availability أهم من رسايل قليلة. وحاجة كمان: لما broker يرجع بعد ما وقع، بيعمل truncate للـ log بتاعه لآخر مكان متأكد منه، لأن ممكن يكون عنده رسايل ما اتأكدتش، والـ leader الجديد كتب مكانها رسايل تانية. وبعدين يسحب من الـ leader ويرجع للـ ISR. Kafka بيعمل ده بحاجة اسمها leader epoch.

نرجع للمشكلة الأولى: الستة brokers في zone واحدة. الـ rack awareness: كل broker ليه broker dot rack، وفي الـ cloud ده بيبقى الـ availability zone. و Kafka لما يوزّع النسخ، بيحط نسخ كل partition في racks مختلفة. تلات نسخ في تلات zones، zone كاملة تقع والـ partition لسه عنده نسختين، يعني لسه فوق الـ min insync. وفيه كمان client dot rack للـ consumers: الـ consumer يقرا من الـ follower اللي في نفس الـ zone بدل الـ leader، وده بيوفّر فلوس الـ traffic بين الـ zones، اللي في الـ cloud بتتحسب. والـ controllers: في الـ versions الجديدة Kafka بقى بـ KRaft من غير ZooKeeper، والـ controllers بيعملوا quorum بالأغلبية. تلاتة، واحد في كل zone. zone تقع، والاتنين الباقيين أغلبية، والـ cluster شغّال. ولو حطيت اتنين في zone واحدة وهي وقعت، مفيش أغلبية.

ده الـ zones. طب لو الـ region كلها وقعت؟ فيه طريقتين. stretch cluster: cluster واحد brokers بتوعه في أكتر من region. بسيط من برّه، بس acks all معناها كل write بيستنى رحلة بين الـ regions، يعني عشرات الـ ms بدل أقل من واحد. ده ينفع لو الـ regions قريبة أوي. التانية والأشهر: cluster في كل region، و MirrorMaker اتنين بينسخ بينهم. الـ replication هنا async: الـ producer في الـ east بياخد ack من الـ east، والنسخ للـ west بييجي بعدين. وده معناه إن فيه lag، والـ lag ده هو اللي هيضيع لو الـ east وقعت فجأة. MirrorMaker بيسمّي الـ topic في الـ west باسم الـ region زي east dot payments. ليه؟ علشان لو الاتنين active وبينسخوا لبعض، ما يحصلش loop، وتعرف كل رسالة جاية منين. والـ offsets مش هي هي. الـ retention في الـ east ممكن يكون مسح أول رسايل، فالـ offset أربعة في الـ east هو صفر في الـ west. علشان كده MirrorMaker بيحفظ checkpoints، ويترجم الـ consumer offsets من region للتانية.

والـ failover نفسه، ده قرار مش زرار. الـ RPO: قد إيه data مستعد تخسرها. والـ RTO: قد إيه وقت مستعد تبقى واقف. لو الـ east وقعت والـ lag صفر، اعمل failover. لو الـ lag كبير، وكل رسالة فيه payment، فـ failover معناه إن الـ payments دي مش هتتعالج، أو هتتعالج بعدين مرتين لما الـ east يرجع. هنا ممكن تستنى شوية، أو حد مسؤول يقرر إن الخسارة مقبولة، ويبقى عندك خطة للـ reconciliation بعدين. وبعد القرار: الـ consumers يكمّلوا في الـ west من الـ offset المترجم، مش من earliest ولا latest. earliest يعني تعالج كل حاجة تاني، و latest يعني تفوّت. والـ producers يتنقلوا للـ west، عن طريق الـ DNS أو الـ config. والـ failback، الرجوع للـ east، أصعب: الـ east فيه رسايل ما اتنسختش، والـ west فيه رسايل جديدة. محتاج mirror من الـ west للـ east الأول، وتراجع الفرق. والأهم: كل ده تتدرّب عليه. game day كل كام شهر، تقفل region وتشوف.

التحدي. عندك kafka dot naive dot js، و lab dot js بيجرّب الـ rack awareness، و acks واحد قصاد acks all، و min insync، والـ unclean election، و region تانية بـ mirror و failover. والمطلوب cluster dot js: simulator لـ Kafka. كل التفاصيل في الكومنت: createCluster بالتوزيع على الـ racks، و produce بالـ acks والـ errors، والـ hw، و replicate، و failBroker بالـ clean والـ unclean election، و recoverBroker بالـ truncate، و fetch لحد الـ hw، و describe، والـ commits. وبعدين createMirror بالـ remote topic والـ checkpoints والـ translate والـ lag، و failoverPlan. اقرا الكومنت كويس، ونشغّل test dot js لحد ما تبقى أربعتاشر من أربعتاشر.

الحل. assign: بنجمّع الـ brokers بالـ rack، ونعمل list بالدور، وكل partition بياخد نسخ ورا بعض منها، فبتيجي في racks مختلفة. produce: الـ checks الأول، وبعدين الكتابة، ومع acks all الـ ISR بياخدوا نسخة والـ hw يتحرك. failBroker: بنحفظ الـ hw، ونشيل الـ broker من الـ ISR، ولو كان leader نختار من الـ ISR، ولو مفيش يا unclean بخسارة متسجّلة يا offline. recoverBroker: truncate للـ hw المحفوظ. والـ mirror: fetch لحد الـ hw، و produce في الـ remote، و checkpoint لكل رسالة. نشغّل lab dot js.

// cluster.js: simulator لـ Kafka: replication و ISR و acks و high watermark و leader election و rack awareness، و mirroring بين regions
export class KafkaError extends Error {
  constructor(code, message = code) {
    super(message);
    this.code = code;
  }
}

export function createCluster({ name = "kafka", brokers }) {
  const nodes = new Map(brokers.map((b) => [b.id, { ...b, up: true }]));
  const topics = new Map();
  const groups = new Map();

  function assign(partitions, rf) {
    // rack aware: لف على الـ racks بالترتيب، ومن كل rack broker، علشان نسخ الـ partition ما تبقاش في rack واحد
    const byRack = new Map();
    for (const b of [...nodes.values()].sort((a, b) => a.id - b.id)) {
      if (!byRack.has(b.rack)) byRack.set(b.rack, []);
      byRack.get(b.rack).push(b.id);
    }
    const racks = [...byRack.keys()].sort();
    const ordered = [];
    for (let i = 0; ordered.length < nodes.size; i++) for (const r of racks) if (byRack.get(r)[i] != null) ordered.push(byRack.get(r)[i]);
    return Array.from({ length: partitions }, (_, p) => Array.from({ length: rf }, (_, k) => ordered[(p + k) % ordered.length]));
  }

  function createTopic(topic, { partitions = 1, replicationFactor = 3, minInsync = 2, uncleanLeaderElection = false } = {}) {
    if (replicationFactor > nodes.size) throw new KafkaError("INVALID_REPLICATION_FACTOR");
    const parts = assign(partitions, replicationFactor).map((replicas) => ({
      replicas,
      leader: replicas[0],
      isr: [...replicas],
      logs: Object.fromEntries(replicas.map((id) => [id, []])),
      hw: 0,
      hwAtFail: {},
    }));
    topics.set(topic, { minInsync, uncleanLeaderElection, parts });
    return parts.map((p) => [...p.replicas]);
  }

  const part = (topic, p) => {
    const t = topics.get(topic);
    if (!t || !t.parts[p]) throw new KafkaError("UNKNOWN_TOPIC_OR_PARTITION");
    return [t, t.parts[p]];
  };
  const updateHw = (pt) => {
    const leo = Math.min(...pt.isr.map((id) => pt.logs[id].length));
    pt.hw = Math.max(pt.hw, Math.min(leo, pt.logs[pt.leader]?.length ?? 0));
  };

  function produce(topic, p, value, { acks = "all" } = {}) {
    const [t, pt] = part(topic, p);
    if (pt.leader == null) throw new KafkaError("LEADER_NOT_AVAILABLE");
    if (acks === "all" && pt.isr.length < t.minInsync) throw new KafkaError("NOT_ENOUGH_REPLICAS");
    const log = pt.logs[pt.leader];
    const offset = log.length;
    log.push(value);
    if (acks === "all") {
      for (const id of pt.isr) pt.logs[id] = [...log]; // الـ followers في الـ ISR خدوها قبل الـ ack
      updateHw(pt);
    }
    if (pt.isr.length === 1) updateHw(pt);
    return { offset, acked: acks !== 0 };
  }

  function replicate() {
    for (const t of topics.values()) for (const pt of t.parts) {
      if (pt.leader == null) continue;
      const leaderLog = pt.logs[pt.leader];
      for (const id of pt.replicas) {
        if (id === pt.leader || !nodes.get(id).up) continue;
        pt.logs[id] = [...leaderLog]
…

$ node lab.js
— 1) rack awareness —
  p0: brokers 1,2,3 في east-a,east-b,east-c
  p1: brokers 2,3,4 في east-b,east-c,east-a
  p2: brokers 3,4,5 في east-c,east-a,east-b

— 2) acks 1 قصاد acks all، والـ leader وقع —
  acks 1، بعد ما الـ leader وقع: [] ← الرسالة أخدت ack وضاعت
  acks all، بعد ما الـ leader وقع: [ 'pay-1' ]

— 3) min.insync.replicas —
  p0 leader=1 isr=[1] hw=0 | p1 leader=4 isr=[4] hw=0 | p2 leader=4 isr=[4,5] hw=0
  produce على p1 بـ acks all: NOT_ENOUGH_REPLICAS
  بعد ما broker اتنين رجع: 0 ← p0 leader=1 isr=[1,2] hw=0 | p1 leader=4 isr=[2,4] hw=1 | p2 leader=4 isr=[4,5] hw=0

— 4) unclean leader election —
  unclean=false: p0 leader=null isr=[] hw=5 OFFLINE ← مستنيين broker واحد يرجع
  unclean=true: p0 leader=2 isr=[2] hw=3 ← ضاع 2 رسايل

— 5) region تانية و failover —
  synced: 10 رسايل في east.payments
  الـ lag على p0: 1
  failoverPlan rpoAccepted=false: { action: 'wait', reason: 'LAG_TOO_HIGH' }
  failoverPlan rpoAccepted=true: { action: 'failover', reason: 'DATA_LOSS_ACCEPTED' }
  billing كان واصل لأتنين في east، يكمّل من 2 في west: [ 'pay-6' ]

الأول: كل partition نسخه في تلات zones. التاني: مع acks واحد الرسالة ضاعت، ومع acks all لسه موجودة. التالت: الـ ISR بقى broker واحد، فالـ produce اترفض بـ NOT_ENOUGH_REPLICAS، ولما broker رجع الـ write عدّى. الرابع: من غير unclean الـ partition offline ومستني، ومعاه leader جديد ورسالتين ضاعوا. الخامس: الـ mirror نسخ لـ east dot payments، وفيه رسالة لسه في الـ lag. الـ failoverPlan قال wait، ولما الخسارة اتقبلت قال failover، والـ billing كمّل في الـ west من الـ offset المترجم من غير ما يعيد اللي عالجه. ونشغّل test dot js.

$ node test.js
✅ rack awareness: كل partition نسخها التلاتة في تلات racks مختلفة، والـ leaders متوزعين على الستة [[1,2,3],[2,3,4],[3,4,5],[4,5,6],[5,6,1],[6,1,2]]
✅ replicationFactor أكبر من عدد الـ brokers ← INVALID_REPLICATION_FACTOR، و topic مش موجود ← UNKNOWN_TOPIC_OR_PARTITION 
✅ acks all: الـ offsets بالترتيب، والـ hw بقى اتنين على طول، والـ consumer شايفهم {"r1":{"offset":0,"acked":true},"r2":{"offset":1,"acked":true},"d":{"leader":1,"replicas":[1,2,3],"isr":[1,2,3],"hw":2,"underReplicated":false,"offline":false}}
✅ acks 1: الـ leader بس عنده الرسالة، فالـ hw ما اتحركش والـ consumer مش شايفها لحد الـ replicate. و acks صفر ← acked بـ false {"before":["a"],"after":["a","b"],"z":{"offset":2,"acked":false}}
✅ acks 1 والـ leader وقع قبل الـ replicate: leader جديد من الـ ISR، والرسالة اللي أخدت ack ضاعت {"leader":2,"replicas":[1,2,3],"isr":[2,3],"hw":1,"underReplicated":true,"offline":false}
✅ acks all والـ leader وقع: clean election، مفيش ولا رسالة ضاعت، والـ partition بقى underReplicated وشغّال {"leader":2,"replicas":[1,2,3],"isr":[2,3],"hw":3,"underReplicated":true,"offline":false}
✅ min.insync.replicas اتنين و ISR فيه واحد بس: acks all ← NOT_ENOUGH_REPLICAS، إنما acks 1 بيعدّي
…

كله نجح. وكده خلّصنا قسم الـ distributed systems والتكاملات. في القسم الجاي، التوسع والاختبار والتصميم. وأول درس: TypeScript. الـ types بتختفي وقت الـ runtime، فإزاي تتأكد إن الـ data اللي جاية من برّه شكلها صح؟ الـ runtime boundaries، و zod، و الـ types اللي بتطلع من الـ schema، والـ branded types.

// kafka.naive.js: الـ setup اللي كان شغّال "كويس" لحد ما region وقعت
export const topic = {
  name: "payments",
  partitions: 6,
  replicationFactor: 2, // نسختين، "كفاية"
  // min.insync.replicas: الـ default واحد
  // unclean.leader.election.enable: true ← "علشان ما يبقاش فيه downtime"
};

export const brokers = [
  // الستة في نفس الـ availability zone، ومن غير broker.rack
  { id: 1, host: "kafka-1.eu-west-1a" }, { id: 2, host: "kafka-2.eu-west-1a" }, { id: 3, host: "kafka-3.eu-west-1a" },
  { id: 4, host: "kafka-4.eu-west-1a" }, { id: 5, host: "kafka-5.eu-west-1a" }, { id: 6, host: "kafka-6.eu-west-1a" },
];

export const producer = {
  acks: 1, // "أسرع"، الـ leader بس بيأكد
  retries: 0,
};

export const dr = {
  // الـ region التاني؟ "عندنا backup كل يوم بالليل"
  // ولو حصل failover: الـ consumers هيبدأوا من earliest ← كل الـ payments تتعالج تاني
  consumerResetPolicy: "earliest",
};

// lab.js: cluster بستة brokers في تلات zones، وبعدين region تانية بـ mirror
import { createCluster, createMirror, failoverPlan } from "./cluster.js";

const zones = ["a", "b", "c"];
const region = (name, n) => createCluster({ name, brokers: Array.from({ length: n }, (_, i) => ({ id: i + 1, rack: `${name}-${zones[i % 3]}`, region: name })) });
const show = (k, t) => k.describe(t).map((p, i) => `p${i} leader=${p.leader} isr=[${p.isr}] hw=${p.hw}${p.offline ? " OFFLINE" : ""}`).join(" | ");
const tryIt = (fn) => { try { return fn(); } catch (e) { return e.code; } };

console.log("— 1) rack awareness —");
const east = region("east", 6);
const parts = east.createTopic("payments", { partitions: 3, replicationFactor: 3, minInsync: 2 });
parts.forEach((r, p) => console.log(`  p${p}: brokers ${r} في ${r.map(east.rackOf)}`));

console.log("\n— 2) acks 1 قصاد acks all، والـ leader وقع —");
const k1 = region("lab", 3);
k1.createTopic("t1"); k1.createTopic("t2");
k1.produce("t1", 0, "pay-1", { acks: 1 });
k1.produce("t2", 0, "pay-1", { acks: "all" });
k1.failBroker(1);
console.log("  acks 1، بعد ما الـ leader وقع:", k1.fetch("t1", 0).map((r) => r.value), "← الرسالة أخدت ack وضاعت");
console.log("  acks all، بعد ما الـ leader وقع:", k1.fetch("t2", 0).map((r) => r.value));

console.log("\n— 3) min.insync.replicas —");
east.failBroker(2); east.failBroker(3);
console.log(" ", show(east, "payments"));
console.log("  produce على p1 بـ acks all:", tryIt(() => east.produce("payments", 1, "x")));
east.recoverBroker(2); east.replicate();
console.log("  بعد ما broker اتنين رجع:", tryIt(() => east.produce("payments", 1, "x").offset), "←", show(east, "payments"));

console.log("\n— 4) unclean leader election —");
for (const unclean of [false, true]) {
  const k = region("u", 3);
  k.createTopic("t", { replicationFactor: 2, minInsync: 1, uncleanLeaderElection: unclean });
  ["a", "b", "c"].forEach((v) => k.produce("t", 0, v));
  k.failBroker(2);
  ["d", "e"].forEach((v) => k.produce("t", 0, v));
  k.recoverBroker(2);
  const lost = k.failBroker(1);
  console.log(`  unclean=${unclean}:`, show(k, "t"), lost.length ? `← ضاع ${lost[0].lost} رسايل` : "← مستنيين broker واحد يرجع");
}

console.log("\n— 5) region تانية و failover —");
const west = region("west", 3);
const m = createMirror({ source: east, target: west, topic: "payments", partitions: 3 });
for (let i = 0; i < 9; i++) east.produce("payments", i % 3, `pay-${i}`);
console.log("  synced:", m.sync(), "رسايل في", m.remote);
east.commit("billing", "payments", 0, 2);
east.produce("payments", 0, "pay-late");
console.log("  الـ lag على p0:", m.lag(0));
for (let b = 1; b <= 6; b++) east.failBroker(b);
for (const rpoAccepted of [false, true]) console.log(`  failoverPlan rpoAccepted=${rpoAccepted}:`, failoverPlan({ primaryHealthy: false, mirrorLag: m.lag(0), maxLag: 0, rpoAccepted }));
const from = m.translate(0, east.committed("billing", "payments", 0));
console.log(`  billing كان واصل لأتنين في east، يكمّل من ${from} في west:
…