Log-ভিত্তিক streaming — partition, offset, consumer group আর retention

১ · Log — সবচেয়ে সরল data structure, সবচেয়ে বড় ভিত্তি

একটি কাল্পনিক ride-sharing অ্যাপে একটি trip শুরু হলে খবর দরকার ভাড়ার হিসাব, fraud ধরার সার্ভিস, notification, analytics আর search-এর। Trip সার্ভিস প্রত্যেককে আলাদা করে ডাকলে নতুন অংশ যোগ করতেই তার কোড বদলাতে হয়, আর একটি অংশ ধীর হলে পুরো trip আটকে যায়। পর্ব ৩১ — Message queue-এ দেখেছেন, মাঝে একটি সারি বসালে প্রেরক আর গ্রাহক আলাদা হয়। এই পর্বের প্রশ্ন একটু ভিন্ন: একই ঘটনা যদি অনেক দল নিজের গতিতে পড়তে চায়, আর দরকার হলে গতকালের ঘটনাও আবার?

System design-এ log মানে একটি শুধু-যোগ-করা (append-only), সম্পূর্ণ ক্রমে সাজানো record-এর সারি। এখানে log বলতে অ্যাপের debug message (“error: …”) বোঝানো হচ্ছে না; এটি ডেটার একটি কাঠামো। তিনটি গুণ একে আলাদা করে:

এই ধারণা নতুন নয়। প্রতিটি database ভেতরে একটি log রাখে — write-ahead log, যা crash-এর পরে ডেটা ফেরায়; বিস্তারিত পর্ব ১৯ — Storage engine-এ। Replication-এ leader সেই log follower-দের পাঠায়, আর follower একই ক্রমে প্রয়োগ করে একই অবস্থায় পৌঁছায় — পর্ব ২২ — Replication-এ। পেছনের নীতিটি সরল: একই শুরুর অবস্থা আর একই ক্রমের একই পরিবর্তন মানে একই শেষ অবস্থা। Log-ভিত্তিক streaming এই ভেতরের log-কে বাইরে এনে সবার জন্য খুলে দেয়।

উদাহরণগুলো মূলত Kafka-ধাঁচের, কারণ এর নামগুলো সবচেয়ে প্রচলিত; ধারণাগুলো সাধারণ (অংশ ৮-এ অন্য সিস্টেম)। Setting-এর নাম, ডিফল্ট আর সীমা সংস্করণভেদে আলাদা — এখানে লেখার সময়ের documentation অনুযায়ী; নকশার আগে নিজের সংস্করণেরটি মিলিয়ে নিন।

২ · Topic আর partition — ক্রম শুধু partition-এর ভেতরে

একটি log একটি মেশিনের ডিস্ক আর নেটওয়ার্কের সীমায় আটকে থাকে। তাই log-কে ভাগ করা হয়। Topic হলো একটি নামওয়ালা stream — যেমন ride-events বা payments. প্রতিটি topic কয়েকটি partition-এ ভাগ, আর প্রতিটি partition নিজেই একটি আলাদা append-only log, নিজের offset ০ থেকে শুরু। Partition-গুলো কয়েকটি সার্ভারে (broker) ছড়ানো, আর প্রতিটির কয়েকটি কপি থাকে (অংশ ৬)। নিচের ছবিতে পুরো চিত্রটি।

Key ঠিক করে partition, আর partition ঠিক করে ক্রম

Producer প্রতিটি record-এর সাথে একটি ঐচ্ছিক key দিতে পারে। Key থাকলে partition ঠিক হয় hash(key) mod N দিয়ে (N = partition-এর সংখ্যা; Kafka-র Java client-এর ডিফল্ট partitioner key-এর murmur2 hash ব্যবহার করে)। ফলে একই key-এর সব record একই partition-এ, লেখার ক্রমেই। Key না থাকলে record-গুলো partition-এ ছড়িয়ে দেওয়া হয় — পুরোনো সংস্করণে পালা করে, নতুনগুলোতে একটি batch ভরা পর্যন্ত একটি partition-এ (“sticky”). তখন সমান ভার মেলে, কিন্তু কোনো ক্রম নেই।

Key বাছাই আসলে পর্ব ২৩ — Partitioning আর sharding-এর shard key বাছাইয়েরই আরেক রূপ, আর একই ফাঁদগুলো এখানেও আছে:

• কম ভিন্ন মান: key = city দিলে কয়েকটি মাত্র partition কাজ পায়, আর বড় শহরের partition একাই ভারী — একটি hot partition.
• একটি key-এ বিশাল চাপ: একটি বড় ব্যবসায়ীর অ্যাকাউন্ট বা ভাইরাল একটি পোস্ট। পর্ব ২৩-এর salting এখানে দামি: key ভাঙলে সেই key-এর ক্রম আর নিশ্চিত থাকে না।
• Partition বাড়ানো: N বদলালে hash(key) mod N বদলায়, বেশিরভাগ key নতুন partition-এ যায় (পর্ব ২৩ আর ১৬)। পুরোনো record পুরোনো partition-এই থাকে, তাই একই key-এর পুরোনো আর নতুন record দুই partition-এ — ক্রম ভাঙতে পারে। ক্রম জরুরি হলে শুরুতেই যথেষ্ট partition নিন (অংশ ৭-এর হিসাব)।

৩ · Offset আর consumer group — কে কোন অংশ পড়বেন

Queue-তে broker মনে রাখে কোন বার্তা কে নিল। Log-এ কাজটি অনেক সরল: প্রতিটি consumer শুধু একটি সংখ্যা মনে রাখেন — প্রতিটি partition-এ পরের যে offset পড়তে হবে। Partition P0-তে offset ৪ মানে ০ থেকে ৩ পড়া আর প্রক্রিয়া শেষ। এই সংখ্যাটি broker-এর কাছে জমা রাখাকে বলে offset commit (অংশ ৪)। Kafka-র পুরোনো সংস্করণে এটি ZooKeeper-এ থাকত; এখন থাকে একটি ভেতরের, compacted topic-এ — log দিয়েই log-এর হিসাব।

Consumer group — কাজ ভাগ, কিন্তু partition ভাগ নয়

একজন consumer সেকেন্ডে যত record সামলাতে পারেন, তার চেয়ে বেশি এলে কয়েকজন লাগে। একই কাজের consumer-রা মিলে একটি consumer group, একটি নাম নিয়ে। নিয়মটি কঠোর: group-এর ভেতরে প্রতিটি partition ঠিক একজন consumer পড়েন; একজন কয়েকটি partition পড়তে পারেন, কিন্তু একটি partition দুজনে ভাগ করেন না। এতে দুটি ফল:

• ক্রম টেকে: একটি partition একজনই ক্রমে পড়েন, তাই একই key-এর record কখনো দুজন consumer একসাথে প্রক্রিয়া করেন না।
• সর্বোচ্চ parallelism = partition-এর সংখ্যা: ৬টি partition-এ ৮ জন consumer দিলে ২ জন অলস বসে থাকেন। Partition-এর সংখ্যা তাই ভবিষ্যৎ consumer-সংখ্যার ছাদ।

আর আলাদা group একে অপরকে চেনে না: “matching” আর “analytics” দুটি group একই topic পড়ে, প্রত্যেকে নিজের offset নিয়ে — একটি বার্তা প্রতিটি group-এ একবার করে পৌঁছায়। এটিই log-এর ওপর pub/sub আর fan-out; event-driven নকশায় এর ব্যবহার পর্ব ৩২ — Pub/sub আর event-driven architecture-এ।

Rebalancing — কে কখন partition ছাড়েন, কে নেন

Group-এ কেউ যোগ দিলে, কেউ স্বেচ্ছায় বা heartbeat বন্ধ হয়ে বিদায় নিলে, অথবা topic-এ partition বাড়লে, partition-এর মালিকানা নতুন করে ভাগ হয়। একটি broker এই group-এর coordinator; প্রচলিত (classic) protocol-এ group-এরই একজন consumer নতুন ভাগ হিসাব করে coordinator-কে দেন; নতুন সংস্করণের নতুন protocol-এ এই হিসাব broker নিজেই করে (ভাবনার প্রশ্ন ৩)। পুরোনো ধাঁচে (“eager”) rebalance-এর সময় সবাই নিজের সব partition ছেড়ে দেন, ফলে কিছুক্ষণ পুরো group থেমে থাকে। নতুন ধাঁচে (cooperative বা incremental) শুধু যেসব partition সত্যিই সরবে, সেগুলোই থামে। নিচের stepper-এ একটি group-এ তৃতীয় consumer আসা আর পরে একজনের বিকল হওয়া দেখুন।

Rebalance কমানোর কৌশল (ভাবনার প্রশ্ন ৩): consumer-কে স্থায়ী পরিচয় দেওয়া (“static membership”), একটি poll-এ বেশি কাজ না নেওয়া, আর cooperative assignment.

নিচের প্রোগ্রামে চারটি trip-এর তিনটি করে অবস্থান-হালনাগাদ (v1, v2, v3) key ধরে ৩টি partition-এ যায়; দুজনের group প্রতি partition থেকে সর্বোচ্চ ২টি পড়ে; তারপর তৃতীয়জন এসে rebalance, আর সবাই committed offset থেকে বাকিটা পড়েন। শেষে যাচাই — প্রতিটি key-এর হালনাগাদ ক্রমে এসেছে কি না। Hash FNV-1a 32-bit; ভাগের নিয়ম সরল round robin (partition p যায় সদস্য p mod n-এর কাছে)।

// A topic with 3 partitions: keyed messages, then a consumer group of 2 -> 3 consumers.
function fnv1a(s) {               // FNV-1a 32-bit hash
  let h = 0x811c9dc5;
  for (let i = 0; i < s.length; i++) { h ^= s.charCodeAt(i); h = Math.imul(h, 0x01000193); }
  return h >>> 0;
}
const P = 3, log = [[], [], []], keys = ["ride-10", "ride-11", "ride-13", "ride-17"];
let rr = 0; const noKey = [];
for (let v = 1; v <= 3; v++)      // every ride sends 3 location updates, in order v1, v2, v3
  for (const k of keys) {
    log[fnv1a(k) % P].push(k + ":v" + v);
    if (k === "ride-10") noKey.push("v" + v + "->P" + (rr % P)); rr++;   // what a key-less round robin would do
  }
log.forEach((m, p) => console.log("P" + p + "  " + m.map((x, o) => o + "=" + x).join("  ")));
console.log("ride-10 without a key: " + noKey.join(" ") + "  (spread out, no order guarantee)");
const committed = [0, 0, 0], seen = {};
function run(members, limit) {    // round-robin assignment: partition p -> member p % n
  for (let i = 0; i < members.length; i++) {
    const mine = [0, 1, 2].filter((p) => p % members.length === i), done = [];
    for (const p of mine) {
      const end = Math.min(log[p].length, committed[p] + limit);
      for (let o = committed[p]; o < end; o++) {
        const [k, v] = log[p][o].split(":"); (seen[k] = seen[k] || []).push(v); done.push("P" + p + "@" + o);
      }
      committed[p] = end;         // commit after processing
    }
    console.log("  " + members[i] + " owns " + mine.map((p) => "P" + p).join(",").padEnd(6) + " read " + (done.join(" ") || "-"));
  }
}
console.log("group of 2 (at most 2 messages per partition):"); run(["c1", "c2"], 2);
console.log("c3 joins -> rebalance; resume from committed offsets P0=" + committed[0] + " P1=" + committed[1] + " P2=" + committed[2]);
run(["c1", "c2", "c3"], 99);
for (const k of keys) console.log(k + " seen " + seen[k].join(" ") + (seen[k].join() === "v1,v2,v3" ? "  in order" : "  OUT OF ORDER"));

P0  0=ride-11:v1  1=ride-11:v2  2=ride-11:v3
P1  0=ride-13:v1  1=ride-13:v2  2=ride-13:v3
P2  0=ride-10:v1  1=ride-17:v1  2=ride-10:v2  3=ride-17:v2  4=ride-10:v3  5=ride-17:v3
ride-10 without a key: v1->P0 v2->P1 v3->P2  (spread out, no order guarantee)
group of 2 (at most 2 messages per partition):
  c1 owns P0,P2  read P0@0 P0@1 P2@0 P2@1
  c2 owns P1     read P1@0 P1@1
c3 joins -> rebalance; resume from committed offsets P0=2 P1=2 P2=2
  c1 owns P0     read P0@2
  c2 owns P1     read P1@2
  c3 owns P2     read P2@2 P2@3 P2@4 P2@5
ride-10 seen v1 v2 v3  in order
ride-11 seen v1 v2 v3  in order
ride-13 seen v1 v2 v3  in order
ride-17 seen v1 v2 v3  in order

৪ · Offset commit — কখন “পড়েছি” বলবেন

Consumer-এর জীবন একটি চক্র: poll (কয়েকটি record আনা) → প্রক্রিয়া (database-এ লেখা, SMS পাঠানো) → commit (broker-কে জানানো “এই offset পর্যন্ত শেষ”)। Crash হলে নতুন মালিক শুরু করেন শেষ committed offset থেকে। তাই প্রক্রিয়া আর commit-এর মাঝে যা ঘটে, সেটিই নির্ধারণ করে বার্তা হারাবে, না দুবার আসবে।

Auto commit-এর দুটি জানালা: commit হয়ে গেছে কিন্তু প্রক্রিয়া বাকি — crash-এ record হারায় (যেমন record অন্য thread-এ পাঠিয়ে পরের poll ডাকলে; Java client-এর documentation অনুযায়ী প্রতিটি poll-এর সব record পরের poll-এর আগে শেষ করলে auto commit at-least-once দেয়); প্রক্রিয়া শেষ কিন্তু commit বাকি — crash-এ শেষ কয়েক সেকেন্ডের record আবার আসে। নিচের প্রোগ্রাম দুটি নীতি পাশাপাশি চালায়: offset ০ থেকে ৯, প্রতিবার ৪টি poll, আর offset ৫-এ একবার crash.

// When to commit the offset? One partition, offsets 0..9, polls of 4, one crash mid-batch.
const LOG = 10, BATCH = 4, CRASH_AT = 5;     // the consumer dies while handling offset 5

function consume(mode) {
  let committed = 0, crashed = false; const effects = [];
  while (committed < LOG) {
    const from = committed, to = Math.min(LOG, from + BATCH);   // poll: next batch from the committed offset
    if (mode === "before") committed = to;                     // commit first, then process
    let died = false;
    for (let o = from; o < to; o++) {
      if (!crashed && o === CRASH_AT) { crashed = died = true; break; }   // process dies, memory is lost
      effects.push(o);                                         // the side effect, e.g. send an SMS
    }
    if (died) { console.log("  crash at offset " + CRASH_AT + "; restart from committed offset " + committed); continue; }
    if (mode === "after") committed = to;                      // commit only after the whole batch is done
  }
  const count = {}; effects.forEach((o) => (count[o] = (count[o] || 0) + 1));
  const lost = [], dup = [];
  for (let o = 0; o < LOG; o++) { if (!count[o]) lost.push(o); else if (count[o] > 1) dup.push(o); }
  console.log("  effects: " + effects.join(" "));
  console.log("  lost: [" + lost.join(" ") + "]  duplicated: [" + dup.join(" ") + "]");
}
console.log("commit BEFORE processing (at-most-once):"); consume("before");
console.log("commit AFTER processing (at-least-once):"); consume("after");

commit BEFORE processing (at-most-once):
  crash at offset 5; restart from committed offset 8
  effects: 0 1 2 3 4 8 9
  lost: [5 6 7]  duplicated: []
commit AFTER processing (at-least-once):
  crash at offset 5; restart from committed offset 4
  effects: 0 1 2 3 4 4 5 6 7 8 9
  lost: []  duplicated: [4]

একটি বিষাক্ত record: log-এ ক্রম আছে, তাই offset ৫ বারবার ব্যর্থ হলে consumer সেখানেই আটকে থাকেন, পুরো partition পিছিয়ে পড়ে। প্রচলিত সমাধান: কয়েকবার চেষ্টার পরে record-টি আলাদা একটি retry বা dead letter topic-এ লিখে offset এগিয়ে নেওয়া (পর্ব ৩১-এর dead letter queue-এর log-রূপ)। দাম: সেই key-এর পরের record আগে প্রক্রিয়া হবে — ক্রম জরুরি হলে সেগুলোও আটকে রাখতে হয়।

৫ · Retention আর log compaction — খাতা কতদিন রাখবেন

Log শুধু বাড়তেই থাকে, তাই পুরোনো record মোছার নিয়ম লাগে। Partition-এর log ডিস্কে কয়েকটি ফাইলে ভাগ থাকে — প্রতিটি একটি segment; নতুন লেখা যায় শুধু শেষের active segment-এ। দুটি নীতি প্রচলিত:

Retention নকশার সিদ্ধান্ত। একটি consumer group retention-এর চেয়ে বেশি সময় বন্ধ থাকলে তার committed offset-এর record আর নেই — তখন setting অনুযায়ী সে সবচেয়ে পুরোনো বা সবচেয়ে নতুন record থেকে শুরু করে, আর মাঝের ডেটা হারায়। তাই retention হওয়া উচিত consumer-এর সম্ভাব্য দীর্ঘতম বিরতি আর replay-এর প্রয়োজন — দুটির বড়টির চেয়ে বেশি; দাম ডিস্কে (অনুশীলন ৪)।

Log compaction — প্রতিটি key-এর শেষ মান

ধরুন একটি topic-এ প্রতিটি অ্যাকাউন্টের balance বদলালেই একটি record আসে: key = অ্যাকাউন্ট, value = নতুন balance. এক মাস পরে নতুন একটি সার্ভিসের দরকার সব অ্যাকাউন্টের বর্তমান balance — মাঝের লাখো পুরোনো মান নয়। Compaction ঠিক এই কাজ করে: পেছনের একটি cleaner closed segment-গুলো পড়ে প্রতিটি key-এর সর্বশেষ offset খুঁজে নেয়, তারপর শুধু সেগুলো রেখে নতুন segment লেখে। মুছতে চাইলে value ফাঁকা (null) রেখে একটি record পাঠাতে হয় — একটি tombstone; নির্দিষ্ট সময় পরে সেটিও সরে। পর্ব ১৯-এর LSM-tree compaction-এর সাথে মিলটি স্পষ্ট — দুটোই “শেষ মান রাখা, পুরোনো ফেলা”, তবে এখানে offset বদলায় না।

নিচের প্রোগ্রাম একই কাজ করে আর শেষে যাচাই করে: পুরো log আর compacted log — দুটি থেকে শুরু থেকে পড়ে table বানালে ফল এক কি না।

// Log compaction: keep only the latest value per key; offsets never change.
// A compacted topic of account balances (key = account, value = balance; null = deleted).
const log = [["acc-A", 500], ["acc-B", 200], ["acc-A", 450], ["acc-C", 900], ["acc-B", 260],
             ["acc-A", 700], ["acc-C", null], ["acc-D", 50], ["acc-B", 300], ["acc-A", 650]]
  .map(([key, value], offset) => ({ offset, key, value }));

const show = (title, recs) => console.log(title.padEnd(8) + recs.map((r) =>
  r.offset + ":" + r.key.slice(4) + "=" + (r.value === null ? "DEL" : r.value)).join(" "));
// activeFrom: offsets from here are in the active segment, which the cleaner never touches
function compact(recs, activeFrom, dropTombstones) {
  const latest = {};                                  // key -> latest offset in the closed segments
  for (const r of recs) if (r.offset < activeFrom) latest[r.key] = r.offset;
  return recs.filter((r) => r.offset >= activeFrom ||
    (r.offset === latest[r.key] && !(dropTombstones && r.value === null)));
}
function state(recs) {                                // replay from the start into a table
  const t = {};
  for (const r of recs) { if (r.value === null) delete t[r.key]; else t[r.key] = r.value; }
  return Object.keys(t).sort().map((k) => k + "=" + t[k]).join(" ");
}
show("before", log);
const c1 = compact(log, 8, false);    // first pass: offsets 8-9 still in the active segment
const c2 = compact(c1, 10, true);     // later: segment rolled, tombstone retention has passed
show("pass 1", c1);
show("pass 2", c2);
console.log("records: " + log.length + " -> " + c1.length + " -> " + c2.length +
  "; offsets kept: " + c2.map((r) => r.offset).join(","));
console.log("replay full log : " + state(log));
console.log("replay compacted: " + state(c2));

before  0:A=500 1:B=200 2:A=450 3:C=900 4:B=260 5:A=700 6:C=DEL 7:D=50 8:B=300 9:A=650
pass 1  4:B=260 5:A=700 6:C=DEL 7:D=50 8:B=300 9:A=650
pass 2  7:D=50 8:B=300 9:A=650
records: 10 -> 6 -> 3; offsets kept: 7,8,9
replay full log : acc-A=650 acc-B=300 acc-D=50
replay compacted: acc-A=650 acc-B=300 acc-D=50

৬ · Partition-এর replication — leader, follower, ISR আর acks

একটি broker-এর disk নষ্ট হলে তার partition-এর ডেটা শেষ। তাই প্রতিটি partition-এর কয়েকটি কপি (replication factor, সাধারণত ৩), আলাদা broker-এ, সম্ভব হলে আলাদা zone-এ। একটি কপি leader — সব লেখা আর সাধারণত সব পড়া তার কাছে; বাকিরা follower, leader থেকে নতুন record টেনে আনে — পর্ব ২২-এর leader-follower replication, partition ধরে।

দুটি শব্দ জরুরি। ISR (in-sync replicas): যেসব কপি leader-এর সাথে তাল রাখছে — নির্দিষ্ট সময়ের বেশি পিছিয়ে পড়লে একটি follower ISR থেকে বাদ পড়ে, আবার ধরে ফেললে ফেরে। High watermark: যে offset পর্যন্ত record ISR-এর সব কপিতে পৌঁছেছে; consumer শুধু এর আগ পর্যন্ত পড়তে পান। ফলে leader বদলালেও consumer এমন কিছু দেখেননি যা পরে উধাও হতে পারে। নিচের stepper-এ একটি record-এর পুরো যাত্রা — producer থেকে consumer পর্যন্ত।

শেষ সারির ফাঁদটি গুরুত্বপূর্ণ: ISR-এ শুধু leader থাকলে acks=all মানে “শুধু leader লিখেছে”. তাই একটি সঙ্গী setting লাগে — min.insync.replicas: ISR এর চেয়ে ছোট হলে broker লেখা প্রত্যাখ্যান করে। প্রচলিত টেকসই সমন্বয়: replication factor ৩, min.insync.replicas=2, acks=all — একটি broker বিকল হলেও লেখা চলে, দুটি বিকল হলে লেখা থামে কিন্তু নিশ্চিত হওয়া ডেটা হারায় না। এটি একটি সচেতন বাছাই: availability-র বদলে durability; CAP-এর ভাষায় এই আলোচনা পর্ব ৪০-এ। Kafka-র লেখার সময়ের documentation অনুযায়ী ৩.০ সংস্করণ থেকে producer-এর ডিফল্ট acks=all, আগে ছিল ১ — পুরোনো client-এ নিজে যাচাই করুন।

Leader মারা গেলে: cluster-এর controller ISR-এর একটি follower-কে নতুন leader করে। ISR-এর বাইরের, পিছিয়ে থাকা কপিকে leader বানানো (unclean leader election) partition দ্রুত চালু করে, কিন্তু নিশ্চিত record-ও হারাতে পারে — সাধারণত এটি বন্ধ থাকে (ভাবনার প্রশ্ন ৪)। Controller আর metadata-র জন্য Kafka আগে ZooKeeper ব্যবহার করত, এখন নিজের Raft-ভিত্তিক KRaft; লেখার সময়ের সর্বশেষ বড় সংস্করণে ZooKeeper আর নেই — consensus পর্ব ৪২-এ।

৭ · Throughput আর consumer lag — গতি কোথা থেকে, আর কী মাপবেন

Log-ভিত্তিক broker প্রায়ই একটি সাধারণ মেশিনেও সেকেন্ডে কয়েকশো MB পর্যন্ত লেখা নিতে পারে — হার্ডওয়্যার আর setting-ভেদে অনেক আলাদা। গোপন রহস্য নেই; কয়েকটি সরল নকশার সিদ্ধান্ত:

• শুধু শেষে লেখা: log-এ লেখা মানে sequential I/O — random লেখার চেয়ে বহু গুণ দ্রুত, index হালনাগাদের ঝামেলাও নেই (পর্ব ১৯)।
• OS-এর page cache: সদ্য লেখা ডেটা operating system-এর memory-তে থাকে, তাই সময়মতো চলা consumer প্রায়ই disk না ছুঁয়েই পড়েন।
• Zero-copy: Linux-এর sendfile-এর মতো system call দিয়ে ডেটা application-এর memory-তে না এনেই socket-এ পাঠানো যায়; TLS চালু থাকলে encryption-এর জন্য আনতেই হয়, তাই সুবিধা কমে।
• Batching আর compression: producer একই partition-এর record কয়েক millisecond জমিয়ে পুরো batch একসাথে compress করে পাঠায় (gzip, snappy, lz4, zstd); broker batch প্রায় অপরিবর্তিত রাখে, consumer-ও batch ধরে পড়েন। প্রতি record-এর স্থির খরচ ভাগ হয়; দাম কয়েক millisecond বাড়তি latency.
• Partition-এ parallelism: প্রতিটি partition আলাদা broker-এ, আলাদা consumer-এর হাতে — throughput মোটামুটি partition-এর সংখ্যার সাথে বাড়ে।
• Pull মডেল: consumer নিজের গতিতে টানেন; log নিজেই ভিড় শুষে নেয় — পর্ব ১৮ — Overload সামলানো-এর backpressure-এর একটি স্বাভাবিক রূপ।

Consumer lag — সবচেয়ে জরুরি একটি সংখ্যা

প্রতিটি partition-এ lag = সর্বশেষ offset (log end) − group-এর committed offset; group-এর মোট lag সব partition-এর যোগফল। ছবি ৩৩.১-এ ধরুন সব partition-এ LEO ৬: group matching-এর lag (৬ − ৪) + (৬ − ৩) + (৬ − ৫) = ৬, আর analytics-এর (৬ − ১) + (৬ − ০) + (৬ − ২) = ১৫. Record-এর সংখ্যার চেয়েও অর্থবহ হলো সময়ে lag: সবচেয়ে নতুন record প্রক্রিয়া হতে কত সেকেন্ড বাকি — মোটামুটি lag ÷ consumer-এর প্রক্রিয়ার হার; পর্ব ৪ — Latency, throughput আর percentile-এর Little-এর সূত্রের আত্মীয়।

কেন CPU-র বদলে lag? Consumer-এর CPU ৩০% হলেও lag বাড়তে পারে — হয়তো তাঁরা একটি ধীর database-এর অপেক্ষায়। Lag সরাসরি ব্যবহারকারীর অভিজ্ঞতা মাপে: “চালকের অবস্থান কত পুরোনো”. সতর্কসংকেত দিন lag একটি সীমা ছাড়ালে, আর কয়েক মিনিট ধরে ক্রমাগত বাড়তে থাকলে (observability পর্ব ৫০-এ)। নিচের প্রোগ্রাম একটি কাল্পনিক notification topic-এ ৫ মিনিটের ভিড় নিয়ে হিসাব করে।

// Consumer lag during a traffic spike, and how many partitions the peak needs.
// Assumed: each consumer handles 150 msg/s; a partition is read by one consumer of the group.
const PER_CONSUMER = 150, PARTITIONS = 6;
const rateAt = (min) => (min >= 2 && min <= 6 ? 1500 : 600);      // msg/s; spike in minutes 2-6

function simulate(consumers) {
  const active = Math.min(consumers, PARTITIONS);                  // extra consumers sit idle
  const cap = active * PER_CONSUMER;
  let lag = 0, peak = 0, drainedAt = null; const rows = [];
  for (let min = 0; min < 20; min++) {
    lag = Math.max(0, lag + (rateAt(min) - cap) * 60);              // messages behind after this minute
    peak = Math.max(peak, lag);
    if (min > 6 && lag === 0 && drainedAt === null) drainedAt = min + 1;
    rows.push([min, rateAt(min), lag, Math.round(lag / cap)]);
  }
  return { active, cap, peak, drainedAt, rows };
}
const a = simulate(6);
console.log("minute  in/s     lag  behind(s)    capacity " + a.cap + " msg/s");
for (const [m, r, lag, s] of a.rows)
  if (m <= 8 || m % 4 === 0 || m === a.drainedAt - 1)
    console.log(String(m).padStart(6) + String(r).padStart(6) + String(lag).padStart(8) + String(s).padStart(11));
console.log("peak lag " + a.peak + " messages; back to 0 at the end of minute " + (a.drainedAt - 1));
const b = simulate(10);
console.log("10 consumers on " + PARTITIONS + " partitions: active " + b.active + ", capacity " + b.cap +
  " msg/s, peak lag " + b.peak);
const need = Math.ceil(1500 / PER_CONSUMER);
console.log("no lag at the peak needs ceil(1500 / " + PER_CONSUMER + ") = " + need + " consumers -> at least " + need + " partitions");
console.log("with 2x growth headroom: " + Math.ceil(2 * 1500 / PER_CONSUMER) + " partitions");

minute  in/s     lag  behind(s)    capacity 900 msg/s
     0   600       0          0
     1   600       0          0
     2  1500   36000         40
     3  1500   72000         80
     4  1500  108000        120
     5  1500  144000        160
     6  1500  180000        200
     7   600  162000        180
     8   600  144000        160
    12   600   72000         80
    16   600       0          0
peak lag 180000 messages; back to 0 at the end of minute 16
10 consumers on 6 partitions: active 6, capacity 900 msg/s, peak lag 180000
no lag at the peak needs ceil(1500 / 150) = 10 consumers -> at least 10 partitions
with 2x growth headroom: 20 partitions

এটি প্রচলিত একটি মোটা হিসাব (Confluent-এর engineering blog-এ এমন নিয়মের আলোচনা আছে); দুটি ক্ষমতাই নিজের হার্ডওয়্যার আর কোডে মেপে নিতে হয়। বেশি partition-এরও দাম আছে — ভাবনার প্রশ্ন ২।

৮ · Replay, queue বনাম log, CDC আর অন্য সিস্টেম

Replay — অতীত আবার চালানো

Log-এর সবচেয়ে শক্তিশালী গুণ: consumer-এর offset পেছনে সরিয়ে (সময় দিয়েও — “গতকাল সকাল ১০টা থেকে”) একই record আবার প্রক্রিয়া করা। কাজে লাগে বাগ ঠিক করায় (ভুল ভাড়ার হিসাব কোড ঠিক করে আবার), নতুন সার্ভিসে (নতুন fraud model পুরোনো ঘটনা থেকে অবস্থা বানায়) আর নতুন view-তে — একই ঘটনা থেকে ভিন্ন আকারের একটি materialized view তৈরি করায়। শর্ত: retention যথেষ্ট লম্বা, effect idempotent (নইলে দ্বিতীয়বার SMS), আর replay-এর গতিতে নিচের database যেন না ডোবে।

Kreps ২০১৪ সালের একটি প্রবন্ধে (“Questioning the Lambda Architecture”) প্রস্তাব দেন: batch আর stream-এর আলাদা দুটি pipeline-এর বদলে একটি stream pipeline, আর আবার হিসাব লাগলে log থেকে replay — পরে যার নাম হয় “Kappa architecture”. পুরো তুলনা পর্ব ৩৬ — Batch আর stream processing-এ।

Queue বনাম log — কখন কোনটি

পর্ব ৩১-এর message queue আর এই পর্বের log বাইরে থেকে একই রকম দেখায় — দুটোতেই producer লেখে, consumer পড়েন। ভেতরের মডেল আলাদা:

সীমারেখা ঝাপসা হয়ে আসছে, তবু প্রশ্নটি স্থির: বার্তাটি কি একটি কাজ (একজন করলেই শেষ), নাকি একটি ঘটনা (অনেকে জানবেন, ক্রমে)?

CDC — database-এর log থেকে stream

একটি অর্ডার সার্ভিস database-এ অর্ডার লেখে, আর search, cache ও analytics-কেও খবর দিতে চায়। সরল চিন্তা “dual write” — কোডে প্রথমে database, তারপর topic-এ লেখা। কিন্তু দুই লেখার মাঝে crash হলে একটিতে আছে, অন্যটিতে নেই; আর দুটি request একই সারি প্রায় একসাথে বদলালে দুই জায়গায় উল্টো ক্রমে পৌঁছাতে পারে — চিরস্থায়ী অমিল (ভাবনার প্রশ্ন ৫)।

Change data capture সমস্যাটি উল্টো দিক থেকে ধরে: database নিজেই প্রতিটি commit তার log-এ ক্রমে লেখে — PostgreSQL-এ WAL, MySQL-এ binlog. একটি CDC টুল সেই log replication-এর পথে পড়ে (পর্ব ২২-এর logical replication), আর প্রতিটি সারির পরিবর্তন একটি event হিসেবে topic-এ লেখে — সাধারণত প্রতিটি table-এর জন্য একটি topic, key = সারির primary key. Open-source Debezium এই ধাঁচের একটি পরিচিত টুল, যা Kafka Connect-এর ওপর চলে; সমর্থিত database আর খুঁটিনাটি সংস্করণভেদে আলাদা।

• শুরুতে snapshot, তারপর stream: নতুন CDC চালু হলে আগে পুরো table-এর একটি consistent snapshot, আর সেই মুহূর্তের log-এর অবস্থান থেকে পরিবর্তন — যাতে কিছু বাদ না পড়ে।
• Key = primary key: একটি সারির সব পরিবর্তন এক partition-এ, ক্রমে। Topic-টি compacted রাখলে নতুন যে কেউ পুরো table-এর বর্তমান অবস্থা পান।
• ব্যবহার: search index হালনাগাদ (পর্ব ২৭ — Search-এ CDC থেকে index); cache-এর key মুছে দেওয়া বা নতুন করা (পর্ব ২৯ — Eviction আর invalidation); data warehouse আর analytics-এ টানা ডেটা; আর পুরোনো সিস্টেম থেকে নতুনে সরানো (পর্ব ২৩-এর resharding).
• Schema: table-এর একটি column বদলালে event-এর গঠনও বদলায় — consumer ভাঙা ঠেকাতে schema registry আর compatibility-র নিয়ম, পর্ব ১২ — Data format আর serialization-এ।
• কাঁচা table না ঘটনা? CDC database-এর ভেতরের কাঠামো বাইরে ফাঁস করে — column-এর নাম বদলালে বাইরের দশটি সার্ভিস ভাঙে। ব্যবসার অর্থবহ event চাইলে outbox pattern: event-টি ব্যবসার সারির সাথে একই transaction-এ একটি outbox table-এ, আর CDC শুধু সেই table পড়ে — পর্ব ২৫ — Distributed transaction-এ।

অন্য log-ভিত্তিক সিস্টেম

৯ · AI-এর যুগে — streaming-এর নকশায় AI-কে যাচাই

“Microservice-গুলো কীভাবে যুক্ত করব?” — AI-এর প্রিয় উত্তর “মাঝে Kafka দিন”. কখনো সেটি ঠিক, কিন্তু প্রায়ই key, partition, commit আর retention নিয়ে একটি শব্দও থাকে না — অথচ সব ঝামেলা ঠিক সেখানেই। Streaming-এ AI-এর সাধারণ ভুলগুলো:

AI-এর streaming নকশা হাতে পেলে — যাচাইয়ের চেকলিস্ট

• Record-এর হার, MB/s আর ডিস্কের হিসাব নিজে আবার করেছি — streaming platform সত্যিই দরকার কি না দেখেছি
• কোন জিনিসের ক্রম জরুরি লিখেছি, আর key ঠিক সেটিই কি না মিলিয়েছি; hot key-এর ঝুঁকি দেখেছি
• Partition-এর সংখ্যা মাপা ক্ষমতা দিয়ে হিসাব করেছি; consumer-এর সংখ্যা তার বেশি নয়
• Commit-এর সময়, crash-এ হারানো বা ডুপ্লিকেট, আর idempotency-র পরিকল্পনা একটি simulation বা পরীক্ষায় যাচাই করেছি
• acks, ISR, min.insync.replicas আর retention-এর মান আমার সংস্করণের অফিসিয়াল documentation-এ মিলিয়েছি
• Broker, consumer আর নিচের database বিকল হলে কী হবে, আর lag-এর সতর্কসংকেত — একটি failure-mode পর্যালোচনা করেছি