บทที่ 08 — Apache Kafka (คาฟ-ก้า) Deep (เข้มข้นสุดของหนังสือ)

Kafka (คาฟ-ก้า) = distributed event log ของ LinkedIn open source 2011

← บทที่ 07: RabbitMQ Deep | สารบัญ | บทที่ 09: Other Brokers →

🆕 ถ้านี่คือครั้งแรกที่เรียน Kafka: ข้ามไป Part 1.1 ได้เลย — ส่วน TL;DR และ Update 2026 ด้านล่างเป็น "what's new ใน Kafka 4.0" สำหรับคนที่รู้จัก Kafka เวอร์ชันเก่าอยู่แล้ว มือใหม่อ่านครั้งแรกอาจตาลายได้ — เริ่ม Part 1 แล้วค่อยกลับมาอ่านส่วนนี้ทีหลัง

TL;DR: บทนี้ตอบ "ใช้ Kafka 4.0 ระดับ production ยังไง" ครอบคลุม 3 กลุ่มหลัก:
Architecture + Reliability:
Kafka 4.0 + KRaft (เครเฟต — โหมดใหม่ ไม่ใช้ ZooKeeper)
partition/replica/ISR (สำเนาที่ตามทัน)/offset
idempotent producer (ผลิตซ้ำได้ผลเดียว — broker dedup)
transactional producer (commit หลาย topic แบบ atomic)
EOS (Exactly-Once = ส่งครั้งเดียวเป๊ะ)
Ecosystem: Schema Registry + Avro/Protobuf + Kafka Connect + MirrorMaker 2 + Tiered Storage (เก็บ cold data ใน S3)
New in 4.0 (Early Access): KIP-848 (โปรโตคอล rebalance ใหม่) + KIP-932 (Share Groups = queue-style consumer) — ยังไม่ใช่ default, ใช้ production ต้องรอ GA
ข้ามได้ถ้า: ไม่ใช้ Kafka — แต่ถ้าทำ event-driven microservices ปี 2026, Kafka เป็น default
📖 คำย่อทั้งหมด (KRaft / KIP-848 / KIP-932 / ISR / EOS / KIP-405) มี gloss เต็มใน callout ด้านล่าง ก่อนเข้า Part 1

บทนี้คือ บทใหญ่สุด ของหนังสือทั้งเล่ม — Kafka ลึกถึงระดับ "ตั้ง cluster + tune production"

📦 Update 2026 (สำหรับคนที่ใช้ Kafka มาก่อน / migration guide — ข้ามได้ถ้าเพิ่งเริ่มเรียน):

อัพเดต 2026 — Apache Kafka 4.0 (มี.ค. 2025) มีของใหม่หลักๆ ดังนี้:

ลบ ZooKeeper (ซู-คี-เปอร์ = service กลางจัดการ metadata ของ cluster แบบเดิม) ออกถาวร ใช้ KRaft แทน
KIP-848 (consumer rebalance protocol ใหม่) — ยังเป็น Early Access ใน 4.0 ไม่ใช่ default production ใช้ classic protocol ไปก่อน คาด GA ใน 4.1/4.2
KIP-932 Queues for Kafka (share groups) — ยังเป็น Early Access ใน 4.0 ไม่ใช่ default เช่นกัน (ตรวจ Kafka release notes ก่อนใช้ production)
Tiered Storage GA — พร้อมใช้ production แล้ว

ครอบคลุม:

Architecture: broker, topic, partition, replica, leader, ISR
KRaft only (ZooKeeper ถูกลบใน 4.0)
Producer: idempotence, batching, compression, partitioner, transactions
Consumer: group, new consumer protocol (KIP-848), rebalance, offset, EOS
Share Groups (KIP-932) — queue semantics บน Kafka
Schema Registry + Avro/Protobuf
Kafka Connect, MirrorMaker 2
Kafka Streams overview (ลึกที่บท 11)
Tiered Storage (KIP-405 Early Access 3.6 → GA/production-ready 3.9) — cold data ใน S3
Cluster ops + tuning
Monitoring + lag
Production war stories + cheat sheet

ใช้เวลา 6-8 ชั่วโมง (บทที่ใหญ่ที่สุด)

🗺️ บทนี้ยาว 1500+ บรรทัด — ใช้ TOC นี้กระโดดไปจุดที่ต้องการ:
Part 1: Kafka คืออะไร — overview + เมื่อไหร่ใช้ + KRaft setup
Part 2: Architecture — topic/partition/replica/ISR/offset
Part 3: Producer — idempotence, batching, partitioner
Part 4: Consumer + KIP-848 — group, new rebalance protocol
Part 5: Exactly-Once Semantics (EOS)
Part 6: Schema Registry — Avro/Protobuf
Part 7: Kafka Connect + MirrorMaker 2
Part 8: Kafka Streams overview
Part 9: Security
Part 10: Cluster Ops + Tuning + Tiered Storage + Share Groups
Part 11: Monitoring + Lag
Part 12: Production War Stories
Part 13: Cheat Sheet
เพิ่งเริ่ม: Part 1-4 ก่อนเข้าใจ producer/consumer แล้วลองสร้าง topic. migrate จาก RabbitMQ: Part 1.2 + Part 10 ดูว่าต่างจาก queue ยังไง. production tune: Part 10 + 11 + 12.

📖 กางศัพท์หลัก 4 ตัวที่ต้องรู้ก่อน (ที่เหลือเนื้อบทจะอธิบายเมื่อเจอ):
partition = การแบ่ง topic เป็นช่องย่อย ๆ เพื่อขนาน + เรียงลำดับ
replica / leader = สำเนาข้อมูล / ตัวหลักที่รับเขียน-อ่าน
ISR (In-Sync Replicas) = "ชุดสำเนาที่ตามทัน" — replica ที่ sync กับ leader ทันเวลา ใช้ตัดสินว่า message ปลอดภัยพอจะ commit หรือยัง (ดู analogy "หนังสือพิมพ์รายวัน" ใน Part 2.1)
KRaft (เครเฟต — Kafka Raft) = โหมดจัดการ cluster ใหม่ใช้ algorithm Raft แทน ZooKeeper
ศัพท์ที่เหลือ (EOS, KIP, lag, offset, rebalance, compaction, head-of-line blocking ฯลฯ) — มี glossary ท้ายบท + เนื้อบทจะ gloss ตอนเจอ:
lag (แลก = ความล่าช้า) — consumer ยังอ่านไม่ทัน producer; วัดเป็นจำนวน message ที่ยังไม่ได้อ่าน หรือเวลา
rebalance (รี-บา-แลนซ์ = การจัดสรรใหม่) — เมื่อ consumer เพิ่ม/ลด, Kafka จัด partition ใหม่ให้แต่ละ consumer; rebalance นาน = ทุก consumer หยุด consume = พีระมิด!
compaction (คอม-แพคชั่น = การบีบอัด) — Kafka เก็บแค่ message ล่าสุดของแต่ละ key ลบของเก่า; ใช้ทำ "snapshot ของ state"
head-of-line blocking (ตัวหน้าบล็อกตัวหลัง) — message แรกใน partition ประมวลไม่ได้ → message หลังก็ต้องรอ
EOS (Exactly-Once Semantics) = "ส่งครั้งเดียวเป๊ะ ไม่หายไม่ซ้ำ"
KIP (Kafka Improvement Proposal) = เอกสารข้อเสนอปรับปรุง Kafka แต่ละข้อมีเลขกำกับ

Part 1: Kafka คืออะไร

1.1 ภาพรวม

Kafka = distributed event log (บันทึก event แบบกระจาย — "log" ในที่นี้คือสมุดบันทึก append-only เรียงตามเวลา ไม่ใช่ application log):

ตั้งใจเป็น "log ที่ scale ได้" — append, replay
ไม่ใช่ queue traditional
Pull-based — consumer ดึงเอง
Optimized for throughput (1M+ msg/s ต่อ broker)
Designed at LinkedIn (2010), open source 2011

1.2 เมื่อไหร่ใช้ Kafka

Kafka เก่งเรื่อง throughput สูงและการเก็บ event ไว้ให้อ่านซ้ำ/หลาย consumer — เหมาะกับ event sourcing, stream processing, CDC, analytics แต่ไม่เหมาะกับ RPC latency ต่ำ, routing ซับซ้อน หรืองานเล็ก ๆ (overhead สูง) ดูรายการด้านล่างเทียบความเหมาะ:

✅ Event sourcing, audit log ✅ Stream processing (Flink, Kafka Streams, ksqlDB) ✅ High throughput (> 100K msg/sec — ใหญ่กว่า web app ทั่วไป; ถ้าระบบ < 1K req/sec ใช้ Kafka ได้แต่ overkill) ✅ Multiple consumer ของ event เดียวกัน (analytics + audit + processing) ✅ Replay events ✅ CDC (Change Data Capture = จับการเปลี่ยนแปลงข้อมูลจาก DB) — Debezium ✅ ETL pipeline

❌ Low-latency RPC ❌ Per-message acknowledgment (ใช้ offset แทน) ❌ Complex routing (ใช้ RabbitMQ) ❌ Small workload (overhead สูง)

1.3 ติดตั้ง (KRaft mode — Kafka 4.0+ ไม่มี ZooKeeper อีกแล้ว)

Kafka รุ่นใหม่ (4.0+) ใช้ KRaft mode จัดการ metadata เองด้วย Raft ไม่ต้องพึ่ง ZooKeeper (ซู-คี-เปอร์ = service กลางจัดการ metadata ของ cluster แบบเดิม) อีกต่อไป ทำให้ setup ง่ายขึ้น — สำหรับ dev รัน controller+broker รวมในตัวเดียวได้ แต่ production ต้องแยก node และมีอย่างน้อย 3 controller + 3 broker

💡 เริ่มง่ายสุด: ถ้าต้องการแค่ลอง Kafka ภายใน 1 นาที ใช้ image confluentinc/cp-kafka หรือ bitnami/kafka พร้อม default config ก็พอ. config KRaft ด้านล่างเป็น "manual setup" สำหรับเข้าใจ internals — production จริงควรอ้างจาก official image documentation

yaml

# docker-compose.yml — Kafka 4.0 (2025+)
# ⚠️ นี่คือ config snippet ไม่ใช่ runnable compose ครบ — ดู runnable compose เต็มใน lab repo
services:
  kafka:
    image: apache/kafka:4.0.0
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: controller,broker        # combined mode สำหรับ dev
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT  # ⚠️ PLAINTEXT = dev only; production ต้อง SSL หรือ SASL_SSL
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093  # static voter; Kafka 4.0+ มี dynamic quorum (KIP-853) ใช้ KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS แทนได้
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'      # always false ใน production
      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'           # ⚠️ นี่แค่ตัวอย่าง placeholder — ต้อง generate UUID ใหม่ของตัวเอง (คำสั่งด้านล่าง) แล้วเอาค่านั้นมาแทนตรงนี้ ห้าม copy ค่านี้ไปใช้จริง
      KAFKA_LOG_DIRS: /var/kafka/kafka-logs          # ต้องตั้งให้ตรงกับ path ที่ mount volume ด้านล่าง — ไม่งั้น broker จะเขียน log ไปที่ default path อื่นแทน แล้ว volume ที่ mount ไว้จะไม่ได้เก็บอะไรจริง
    volumes:
      - kafka-data:/var/kafka/kafka-logs   # ⚠️ ถ้าไม่ mount volume (หรือ path ไม่ตรงกับ KAFKA_LOG_DIRS) → CLUSTER_ID รีเซ็ตทุกครั้งที่ container restart

volumes:
  kafka-data:

🚨 PLAINTEXT = dev only — ใน production ต้องใช้ SSL หรือ SASL_SSL เพื่อเข้ารหัส traffic และยืนยันตัวตน (ดู Part 13)

⚠️ ตัวอย่างนี้เป็น config snippet ไม่ใช่ runnable compose — KRaft ต้อง kafka-storage.sh format ก่อน start broker (ครั้งแรก) มิฉะนั้น broker จะไม่ boot. ภาพ apache/kafka:4.0.0 จะ auto-format ตอน start ถ้าไม่มี meta.properties แต่ถ้า volume เก่ามีอยู่จะ error. ดู runnable compose เต็มใน lab repo / official Kafka docker doc

🚨 Production: แยก controller node และ broker node — ห้ามใช้ combined mode. Minimum 3 controllers (Raft quorum) + 3+ broker nodes.

bash

# 1. generate cluster id (ใช้ตอนเริ่ม cluster ใหม่) — คำสั่งนี้จะพิมพ์ UUID ออกมา 1 บรรทัด เช่น "d3b07384-d113-4cad-8c1a-93c4b5e2f001"
docker run --rm apache/kafka:4.0.0 /opt/kafka/bin/kafka-storage.sh random-uuid

# 2. เอา UUID จริงที่ได้จากขั้นตอน 1 (ไม่ใช่ค่า CLUSTER_ID ตัวอย่างในไฟล์ compose ด้านบน — ค่านั้นเป็นแค่ placeholder)
#    ไปใส่แทน <UUID> ด้านล่าง แล้วเอาค่าเดียวกันไปใส่ CLUSTER_ID ในไฟล์ compose ด้วย
# 3. format storage ครั้งแรก (ถ้ารัน manual ไม่ใช่ผ่าน image entrypoint)
# kafka-storage.sh format -t <UUID> -c /opt/kafka/config/kraft/server.properties

Part 2: Architecture

2.1 Concepts

L = Leader, F = Follower

💡 Analogy ใกล้ตัว: หนังสือพิมพ์รายวัน (ภาพในใจของ ISR/Leader/Follower ที่กล่าวถึงใน TL;DR)
Topic = ชื่อหนังสือพิมพ์ ("Bangkok Post", "Manager")
Partition = section ของหนังสือพิมพ์ (กีฬา/บันเทิง/การเมือง) — แต่ละ section พิมพ์เรียงตามลำดับของตัวเอง แต่ทั้งสอง section แยกกัน
Replica = ก็อบหนังสือพิมพ์ที่กระจายไปขาย 3 แผง (ป้องกันแผงเดียวไฟไหม้)
Leader = แผงหลักที่รับขาย (write/read)
Follower = แผงสำรอง — copy จากแผงหลัก
ISR = แผงสำรองที่ "อัพเดตทัน" เสมอ
Offset = หมายเลขหน้า — บอกว่าอ่านถึงไหน

Topic — "หัวข้อ stream ของ event"

คือ: ชื่อกลุ่มของ event ที่ producer ส่งเข้า + consumer มา subscribe (เหมือน "table" ใน DB แต่เป็น append-only log)

text

topic = "order-events"
topic = "payment-events"
topic = "user-signups"

กฎ naming ที่แนะนำ (2026 standard):

ใช้ kebab-case หรือ dot.case — เลือก 1 ใช้ทั่วบริษัท
รูปแบบ: <domain>.<event-name> เช่น order.placed, payment.captured
versioning: ใส่ใน topic name ถ้าจำเป็น (order.placed.v2)
environment prefix: prod.order.placed, staging.order.placed

Partition — "หัวใจของ Kafka scaling"

คือ: แต่ละ topic แบ่งเป็น N partitions — เป็น sub-log (log ย่อย) อิสระต่อกัน

text

Topic: "order-events" (3 partitions)
┌──────────────────────────────────────────────┐
│ Partition 0:  [msg msg msg msg msg ...]      │  ← ordered ภายใน
│ Partition 1:  [msg msg msg msg msg ...]      │  ← ordered ภายใน
│ Partition 2:  [msg msg msg msg msg ...]      │  ← ordered ภายใน
└──────────────────────────────────────────────┘
   ↑ ordering ห้ามถือว่าตรงกันข้าม partition !

ทำไมต้องมี partition:

Parallelism — 3 partition = ประมวลขนานได้ 3 consumer พร้อมกัน. ถ้ามี 1 partition = 1 consumer เท่านั้น
Throughput scale — 1 broker จัดการได้ราว ~10MB/s ต่อ partition (ตัวเลข rough estimate ขึ้นกับ disk/network/payload จริง ต้อง benchmark เอง) ถ้าคิดง่ายๆ 100 partitions × 3 broker ≈ 300MB/s รวม — แต่ตัวเลขจริงจะต่ำกว่านี้เพราะยังไม่หัก replication traffic (RF=3 = เขียนซ้ำ 3 เท่า)
Storage scale — กระจาย disk ไป multi-broker

Producer เลือก partition ยังไง:

💡 เริ่มต้น: ถ้าไม่แน่ใจ ใช้วิธี 2 (ระบุ key) — รับประกัน ordering ของ entity เดียวกัน

java

// 1. ระบุ partition ตรง ๆ (rare)
new ProducerRecord<>("orders", 0, "key", "value");

// 2. ระบุ key → hash(key) % partition_count    ⭐ recommended
new ProducerRecord<>("orders", "order-42", "value");
//   "order-42" hash → ไปลง partition เดิมเสมอ
//   → message ของ order เดียวกัน ordered แน่นอน

// 3. ไม่ระบุ key → round-robin (sticky partitioner 2.4+)
new ProducerRecord<>("orders", "value");
//   กระจายโหลด — แต่ ordering ไม่รับประกัน

Key insight: เลือก partition key ผิด = scale ไม่ได้

order_id → ดี, กระจายดี
country (ไทย 80% ของ traffic) → partition skew (load ไม่กระจาย — partition หนึ่งหนัก ส่วนอื่นว่าง) — ⚠️ 1 partition โหลดหนัก, ที่เหลือว่าง

⚠️ Generalized lesson (เคยเห็นบ่อยตอน debug — ไม่ใช่ specific incident ของบริษัทใดบริษัทหนึ่ง): partition imbalance ทำให้ throughput รวมตกฮวบได้ อธิบายเป็นขั้นตอน:
มี key บาง key ที่ "hot" มากผิดปกติ (เช่น user หรือ tenant ที่ใหญ่ที่สุด)
key นั้นกินสัดส่วน traffic ของ topic ไปเยอะ ส่วน partition อื่นว่างเฉยๆ
ผลคือ broker ที่ถือ partition นั้นตัวเดียวกลายเป็นคอขวด (bottleneck) ของทั้งระบบ (ตัวเลข exact % ขึ้นกับ workload — ถ้าจะอ้างเป็นเคสจริงในเอกสารทางการ ให้หา postmortem สาธารณะที่ verify ได้ก่อน)

จำนวน partition ที่แนะนำ:

เริ่มต้น: 10-30 partition ต่อ topic
เพิ่มได้ทีหลังแต่ลดไม่ได้ — ตั้งเผื่ออนาคต
มากเกินไป (1000+) = controller overhead + slow rebalance
น้อยเกินไป = scale consumer ไม่ได้

Replica — "ก็อบสำรอง"

คือ: แต่ละ partition มี N copy กระจายไป N broker (default N=3 = "Replication Factor 3" / RF=3)

Partition 0 (RF=3)

Leader vs Follower:

	Leader	Follower
Write	✅ รับ write จาก producer	❌ pull จาก leader
Read	✅ รับ read จาก consumer	❌ (default)
ต่อ partition	มี 1 ตัว	N-1 ตัว
ตาย	trigger election ใหม่	broker หา replacement

💡 มี feature ชื่อ KIP-392 (Fetch from Follower) ที่ช่วยลดค่าใช้จ่าย cross-AZ traffic ได้ — รายละเอียด config ดูที่ Part 10.2 Rack Awareness (ตอน production tuning)

RF เลือกยังไง:

RF	กรณีใช้	Trade-off
1	dev/test เท่านั้น	1 broker ตาย = data หาย
3	production default	ทน broker fail ได้ 1 ตัว ขณะ min.insync.replicas=2
5	mission critical (banking)	ทน broker fail ได้ 2 ตัว + write latency สูง

ISR (In-Sync Replica) — "Replica ที่ตามทัน"

คือ: subset ของ replicas ที่ ตามทัน leader ภายใน threshold

text

ก่อนเวลา 10:00:00 — ทุก replica อัพเดทตรงกัน
Broker 1 [L]: offset 100
Broker 2 [F]: offset 100   ← ISR ✓
Broker 3 [F]: offset 100   ← ISR ✓
→ ISR = {1, 2, 3}

ตอนเวลา 10:00:05 — Broker 3 ช้า (network/disk issue)
Broker 1 [L]: offset 200
Broker 2 [F]: offset 199   ← ตามทัน — ISR ✓
Broker 3 [F]: offset 100   ← lag เกิน threshold — kicked out!
→ ISR = {1, 2}   (Broker 3 ออกจาก ISR)

ถ้า Broker 1 ตาย ตอนนี้:
→ เลือก leader ใหม่จาก ISR เท่านั้น = {2}
→ Broker 2 กลายเป็น leader (มี data ครบ)
→ Broker 3 ไม่ถูกเลือก (ข้อมูลขาด — เพราะ lag)

ทำไม ISR สำคัญ:

รับประกันว่า "ถ้า leader ตาย → leader ใหม่มี data ครบ" (ไม่มี data loss)
acks=all หมายถึง "รอ ทุก ISR ack" — ไม่ใช่ทุก replica
min.insync.replicas = ขั้นต่ำของ ISR ที่ต้อง online เพื่อ accept write

Config สำคัญ:

yaml

# topic level
replication.factor: 3            # RF=3
min.insync.replicas: 2           # ต้องมี ISR >= 2 ถึงจะ accept write
                                 # ถ้า ISR < 2 → producer ได้ NotEnoughReplicasException

# broker level
replica.lag.time.max.ms: 30000   # follower lag > 30s = kick ออกจาก ISR

กฎทอง 2026: RF=3, min.insync.replicas=2, acks=all → ทน broker fail 1 ตัวได้โดยไม่ data loss + ไม่หยุดให้บริการ

⚠️ Unclean Leader Election (default=false ตั้งแต่ Kafka 0.11 — ห้ามตั้ง true ใน production):

yaml

unclean.leader.election.enable: false  # default 0.11+ — ห้ามเปิด (config ยังรองรับ ไม่ได้ deprecate)

ถ้าเปิด = ยอมให้ replica ที่ lag (ไม่อยู่ใน ISR) เป็น leader → data loss แลก availability. ห้ามเปิดในระบบ business

Offset — "หมายเลขหน้าที่อ่านถึง"

คือ: ตำแหน่งของ message ใน partition (long, 0-indexed, monotonic increasing)

text

Partition 0:  [msg msg msg msg msg msg msg msg ...]
offset:        0   1   2   3   4   5   6   7   8

Consumer A (group=shipping):    committed offset = 4 → ถัดไปอ่าน 5
Consumer B (group=email):        committed offset = 7 → ถัดไปอ่าน 8
Consumer C (group=analytics):    committed offset = 0 → ถัดไปอ่าน 1

Key insight: แต่ละ consumer group track offset ของตัวเอง — อ่าน topic เดียวกันก็ได้ผลต่างกัน. นี่คือ superpower ของ Kafka เทียบกับ queue ทั่วไป (RabbitMQ ลบ msg หลัง consume — Kafka ไม่ลบ)

Offset เก็บที่ไหน (Kafka 0.9+):

ใน topic พิเศษ __consumer_offsets — Kafka เก็บ offset ของตัวเอง
replicated, durable
compact (เก็บแค่ latest offset ต่อ group+partition)

ทำไม offset สำคัญ:

Replay: reset offset = 0 → consumer อ่านซ้ำตั้งแต่ต้น (debug, rebuild state)
Time travel: seek(timestamp) → consumer ไปเริ่มที่จุดเวลาใด ๆ
Lag monitoring: producer offset - consumer offset = consumer ตามช้าแค่ไหน (Part 13 — lag monitoring)

Operations ที่ใช้บ่อย:

bash

# ดู consumer group lag
kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --describe --group shipping-service

# Reset offset เพื่อ replay
# ⚠️ ต้องหยุด consumer group ก่อน (ไม่มี active member) มิฉะนั้นจะ error
kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --group shipping-service --topic order-events \
  --reset-offsets --to-earliest --execute

🎯 สรุป Mental Model

2.2 Storage

เบื้องหลังความเร็วของ Kafka คือวิธีเก็บข้อมูลบน disk — แต่ละ partition เป็นไฟล์ log แบบ append-only (เขียนต่อท้ายเรื่อย ๆ) แบ่งเป็น segment พร้อม index หาตาม offset/timestamp ได้เร็ว การเขียนแบบ sequential + zero-copy ทำให้เร็วกว่า random write มาก:

ใน disk:

text

/var/kafka-logs/
└── topic-A-0/                    (partition 0)
    ├── 00000000000000000000.log  (segment file)
    ├── 00000000000000000000.index
    ├── 00000000000000000000.timeindex
    ├── 00000000000000045392.log (next segment — ชื่อ pad เป็น 20 หลักเสมอ)
    └── ...

Append-only segment files
Index by offset + timestamp
Old segment ลบตาม retention

→ Kafka เร็วเพราะ sequential write + zero-copy (sendfile syscall)

2.3 KRaft (ZooKeeper ตายแล้ว)

Timeline:

Kafka < 2.8: ZooKeeper only
Kafka 2.8-3.2: KRaft preview
Kafka 3.3 (Oct 2022): KRaft GA สำหรับ new cluster
Kafka 3.5: marked ZooKeeper deprecated
Kafka 4.0 (Mar 2025): ZooKeeper code ถูกลบทั้งหมด — KRaft only

KRaft ดีกว่ายังไง:

ไม่ต้อง deploy/operate 2 cluster (ZK + Kafka)
Metadata scale ได้ถึง ล้าน partition (ZK จำกัด ~200k)
Recovery จาก controller failover เร็วขึ้น 10x
Single binary, single security model
Cold start ของ cluster เร็วขึ้นเป็นนาที (vs ชั่วโมง)

Migration จาก ZK → KRaft (Kafka 3.x → 4.0):

bash

# Kafka 3.5-3.9 รองรับ "dual-write" mode ระหว่าง migration
# ขั้นตอน (เฉพาะ cluster ที่ยังใช้ ZK):
# 1. Upgrade ทุก node เป็น 3.9 ก่อน
# 2. ตั้ง KRaft controller quorum
# 3. ตั้ง zookeeper.metadata.migration.enable=true
# 4. Roll brokers ให้ point KRaft controller
# 5. Disable ZooKeeper

# Kafka 4.0 ไม่รับ migration อีกแล้ว — ต้อง migrate ก่อนใน 3.9

ใน 2026: ทุก cluster ควรเป็น KRaft only — ถ้ายังใช้ ZK = migration urgent

Part 3: Producer

3.1 Code พื้นฐาน

โครงสร้าง producer พื้นฐาน: ตั้ง config (bootstrap server, serializer ของ key/value, acks) แล้วสร้าง ProducerRecord ระบุ topic + key + value ส่งแบบ async พร้อม callback ตรวจผล key สำคัญเพราะใช้กำหนดว่า message ลง partition ไหน:

java

Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
// แนะนำใช้ StringSerializer.class.getName() แทน string literal (typo-safe, refactor-friendly)
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", StringSerializer.class.getName());
// ⚠️ raw JSON string เป็น anti-pattern ใน production — ใช้ JsonSerializer หรือ KafkaAvroSerializer + Schema Registry (Part 7)
props.put("acks", "all");
props.put("enable.idempotence", "true");
props.put("max.in.flight.requests.per.connection", "5");  // ต้อง ≤ 5 เมื่อ enable.idempotence=true (ถ้าเกิน producer จะ throw ConfigException ตอน startup — ไม่ใช่ silent fail)

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

ProducerRecord<String, String> record = new ProducerRecord<>(
    "orders",                       // topic
    "order-123",                    // key (used for partition)
    "{\"id\":\"order-123\",...}"    // value
);

producer.send(record, (metadata, exception) -> {
    if (exception != null) log.error("send failed", exception);
    else log.info("sent to partition {} at offset {}",
        metadata.partition(), metadata.offset());
});

producer.flush();
producer.close();

3.2 Acks

acks กำหนดว่า producer จะรอ broker ยืนยันแค่ไหนก่อนถือว่าส่งสำเร็จ — เป็น trade-off ระหว่างความเร็วกับความปลอดภัย: acks=0 เร็วสุดแต่เสี่ยงหาย, acks=all รอทุก replica ใน ISR ยืนยัน ปลอดภัยสุด (ใช้ใน production):

text

acks=0    — fire and forget (เร็วสุด, อันตราย)
acks=1    — wait leader confirm (ไม่รอ follower)
acks=all  — wait ทุก ISR confirm (ปลอดภัยสุด)

Production: acks=all

3.3 Idempotent Producer

ปัญหาคือเมื่อ producer ส่งแล้ว network ตอบช้า มันจะ retry ทำให้ broker ได้ message ซ้ำ — idempotent producer แก้โดยแนบ Producer ID + sequence number ให้ broker ตรวจจับและทิ้ง duplicate เอง (ในรุ่นใหม่เปิดให้ default แล้ว):

properties

enable.idempotence=true

แก้ปัญหา: retry → duplicate

Producer มี Producer ID (PID) + sequence number:

text

Producer sends:    msg(pid=1, seq=0), msg(pid=1, seq=1), msg(pid=1, seq=2)
Broker tracks:     last seq for pid=1 = 2
Producer retries:  msg(pid=1, seq=1) ← duplicate!
Broker:            dedup → ignore

→ no duplicate ภายใน 1 producer session

Default ใน Kafka 3.0+ = enable.idempotence=true + acks=all (ทั้งคู่เป็น default แล้ว ไม่ต้องตั้งเอง). แต่ถ้าใครไปตั้ง acks=0 หรือ acks=1 เองขณะที่ enable.idempotence=true — producer จะ throw ConfigException ตอน startup ทันที (ไม่ใช่แค่เงียบๆ override ค่าให้)

3.4 Batching + Linger

แทนที่จะส่ง message ทีละอันให้สิ้นเปลือง Kafka รวมหลาย message เป็น batch เดียวก่อนส่ง — linger.ms คือเวลารอสะสม, batch.size คือขนาด batch ยิ่งรอนานยิ่งได้ throughput สูง แต่แลกกับ latency เพิ่มเล็กน้อย:

properties

linger.ms=10            — รอสะสม batch ก่อนส่ง (default 0 = ส่งทันที)
batch.size=16384        — bytes ต่อ batch
buffer.memory=33554432  — buffer producer-side

text

Producer.send(record);    ─→ buffer
Producer.send(record);    ─→ buffer
   (linger.ms passed or batch full)
              │
              ▼
      ส่ง batch ไป broker

→ throughput ↑, latency ↑ เล็กน้อย

Recommend: linger.ms=10-100, batch.size=32KB-128KB

3.5 Compression

Kafka บีบอัด batch ที่ producer ก่อนส่ง ช่วยลดทั้ง network และ disk ที่ broker (เก็บแบบบีบอัดเลย) — เลือก algorithm ตาม trade-off ระหว่าง ratio กับ CPU โดย zstd ให้สมดุลดีที่สุดและเป็นตัวแนะนำในปัจจุบัน:

properties

compression.type=snappy   # gzip, lz4, snappy, zstd, none

gzip: ratio สูง, CPU เยอะ
snappy: ratio กลาง, CPU น้อย
lz4: balance
zstd: ratio + speed ดี (recommend ใน 2026) — ⚠️ CPU cost สูงกว่า lz4 ~1.5-2x; ถ้า CPU bound ใช้ lz4 แทน

→ Compression ที่ producer → broker → consumer (decompress) = ลด network + disk

3.6 Partitioner

ตัดสินใจว่า message ไป partition ไหน:

Default:

ถ้ามี key → hash(key) % numPartitions
ถ้าไม่มี key → sticky partitioner (เปลี่ยน partition เป็นช่วง ๆ)

Custom:

java

public class GeoPartitioner implements Partitioner {
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        Order order = (Order) value;
        return order.region().hashCode() % cluster.partitionCountForTopic(topic);
    }
}

properties

props.put("partitioner.class", "com.example.GeoPartitioner");

3.7 Transactional Producer (EOS — Exactly-Once)

ถ้าต้องการเขียนหลาย record (และ commit offset) แบบ "ทั้งหมดหรือไม่เลย" Kafka มี transactional producer — กำหนด transactional.id แล้วครอบงานด้วย begin/commit/abort ถ้าพังกลางทางจะ rollback ทั้งชุด เป็นพื้นฐานของ exactly-once ใน Kafka Streams:

java

// ⚠️ producer ใน 3.1 ใช้ StringSerializer — domain object (Order, Audit) ต้อง serialize เป็น String ก่อน
// หรือเปลี่ยน value.serializer เป็น JsonSerializer (Spring Kafka) / KafkaAvroSerializer (Schema Registry)
props.put("transactional.id", "tx-1");
props.put("enable.idempotence", "true");                    // (default 3.0+)
props.put("acks", "all");                                    // (default 3.0+ เมื่อ idempotence=true)
props.put("value.serializer", JsonSerializer.class.getName()); // หรือ Avro/Proto

KafkaProducer<String, Object> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", order));   // ต้องตรงกับ value.serializer
    producer.send(new ProducerRecord<>("audit", audit));
    // ถ้า flow นี้เป็น "consume → process → produce" (มี consumer อ่านมาก่อน)
    // ต้องเรียก producer.sendOffsetsToTransaction(offsetMap, consumer.groupMetadata()) ด้วย
    // เพื่อ commit offset ของ consumer พร้อม transaction เดียวกัน — ตัวอย่างเต็มดู Part 6.2
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

→ atomic write หลาย record + offset commit

ใช้กับ Kafka Streams สำหรับ EOS

3.8 Producer Settings Reference

รวม config ของ producer ที่ใช้บ่อยไว้เป็นจุดอ้างอิงเดียว — แบ่งเป็นกลุ่ม reliability (acks, idempotence, retries) และ performance (linger, batch, compression) ใช้เป็นจุดตั้งต้นแล้วปรับตามผล benchmark จริง:

yaml

bootstrap.servers: kafka1:9092,kafka2:9092,kafka3:9092
acks: all
enable.idempotence: true
max.in.flight.requests.per.connection: 5
retries: 2147483647       # max — แต่ delivery.timeout.ms (ด้านล่าง) จะเป็นตัว cap จริง ไม่ใช่ retry forever
delivery.timeout.ms: 120000   # 2 นาที — ตัวคุมจริงว่า producer รอ ack นานแค่ไหนก่อน fail
request.timeout.ms: 30000

# performance
linger.ms: 10
batch.size: 32768
compression.type: zstd
buffer.memory: 67108864

# transactional
transactional.id: my-app-tx   # ถ้าใช้ EOS

Part 4: Consumer

4.1 Basic

Consumer ทำงานเป็น loop: poll() ดึง batch ของ record มา process แล้ว commit offset (จุดที่อ่านถึง) — group.id กำหนด consumer group, auto.offset.reset บอกว่าเริ่มอ่านจากไหนเมื่อไม่มี offset เดิม ปิด auto-commit เพื่อคุมเองว่าจะ commit เมื่อ process จริงเสร็จ:

java

Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("group.id", "billing-service");
props.put("key.deserializer", "...StringDeserializer");
props.put("value.deserializer", "...StringDeserializer");
props.put("auto.offset.reset", "earliest");   // earliest / latest / none
props.put("enable.auto.commit", "false");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> record : records) {
        try {
            process(record.value());
        } catch (Exception e) {
            // handle
        }
    }
    consumer.commitSync();
}

4.2 Consumer Group

text

Topic "orders" (6 partitions: P0-P5)

Group "billing":
  Consumer C1 → P0, P1
  Consumer C2 → P2, P3
  Consumer C3 → P4, P5

Group "audit":  (อ่านได้พร้อมกัน — เป็น independent group)
  Consumer A1 → P0, P1, P2, P3, P4, P5

Rules:

1 partition → 1 consumer ภายใน group
ถ้า consumer > partition → consumer บางตัวว่าง
ถ้า partition > consumer → consumer 1 ตัวรับหลาย partition

→ Max parallelism = partition count

4.3 Rebalance

ตอน consumer join/leave group:

text

[C1, C2, C3] → C3 ตาย → rebalance → [C1, C2]
   C1 ← P0,P1                          C1 ← P0,P1,P2
   C2 ← P2,P3                          C2 ← P3,P4,P5
   C3 ← P4,P5

STW ของ group ระหว่าง rebalance → ทุก consumer หยุด process ชั่วครู่

Rebalance Protocols (Classic, ผ่าน client-side coordinator)

RangeAssignor — assign by range (default เก่า)
RoundRobinAssignor — round robin
StickyAssignor — minimize re-assignment
CooperativeStickyAssignor — incremental rebalance (ไม่ STW ทั้ง group)

yaml

partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor

🆕 KIP-848 — New Consumer Group Protocol (Kafka 4.0 Early Access)

ปัญหาเดิม: rebalance ทำที่ client → consumer ทุกตัวต้องตกลงกัน, ถ้า client buggy = stuck

KIP-848 ย้าย logic ไปอยู่ที่ broker (group coordinator):

✅ Incremental rebalance ทั้งหมด — ไม่มี STW
✅ Assignment ถูก compute ที่ broker → client ทำตามอย่างเดียว
✅ Heartbeat แยกจาก poll → no more "consumer kicked because process ช้า"
✅ Rolling upgrade ของ client library ไม่ trigger rebalance ทั้ง group

yaml

# enable per consumer (Kafka client 3.7+)
group.protocol: consumer       # ค่าใหม่ (vs default "classic")
# ⚠️ ต้องตั้ง broker config ก่อน: group.coordinator.rebalance.protocols=classic,consumer
# ⚠️ ใน Kafka 4.0 EA อาจต้องเปิด unstable.api.versions.enable=true ที่ broker เพิ่มด้วย (ตรวจ release notes)

⚠️ KIP-848 = Early Access ใน Kafka 4.0 (Mar 2025) — ยังไม่ใช่ stable default. Release notes ระบุว่า "new consumer protocol" อยู่ใน early access + อาจเปลี่ยนได้. ใช้ใน production ตอนนี้ยังเร็วเกินไป — ตรวจ stability ใน release notes ก่อนเปิด. คาด GA ใน Kafka 4.1/4.2

text

Classic:                          KIP-848:
  Client computes assignment         Broker computes assignment
  All clients must agree             Clients receive + apply
  Coordinator only stores            Coordinator authoritative
  Rebalance = STW                    Rebalance = incremental always

💡 ใน Kafka 4.0+ ถ้าตั้ง group.protocol: consumer ไม่ต้องตั้ง partition.assignment.strategy แล้ว — broker จัดให้

4.4 Offset Management

Auto Commit (ไม่แนะนำ)

properties

enable.auto.commit=true
auto.commit.interval.ms=5000

→ auto-commit จะ commit offset ของ records ที่ได้มาจาก poll ก่อนหน้า ตอนเรียก poll() ครั้งถัดไป (โดยอิงว่าผ่าน auto.commit.interval.ms แล้วหรือยัง — ไม่ใช่ commit ทุก poll call)

⚠️ ผลที่ตามมา (at-least-once gap): records ที่กำลัง process อยู่ใน batch ปัจจุบัน ยังไม่ถูก commit — ถ้า consumer crash ระหว่าง process batch นี้ (ก่อน poll() ครั้งถัดไป) → records ทั้ง batch จะถูก redeliver และ process ซ้ำ. ห้ามใช้ auto-commit กับงานที่ side effect ไม่ idempotent (ทำซ้ำแล้วผลเปลี่ยน)

Manual Commit (แนะนำ)

Sync:

java

consumer.commitSync();    // block จน commit ok

Async:

java

consumer.commitAsync((offsets, exception) -> {
    if (exception != null) log.error("commit fail", exception);
});

Per-record:

java

consumer.commitSync(Map.of(
    new TopicPartition("orders", 0),
    new OffsetAndMetadata(offset + 1)
));

Offset Reset

properties

auto.offset.reset=earliest    # อ่านจาก offset 0 (สำหรับ group ใหม่)
auto.offset.reset=latest      # อ่านจากปลาย (skip ของเก่า)
auto.offset.reset=none        # throw error ถ้าไม่มี committed offset

4.5 Poll Loop ที่ถูก

java

while (running) {
    ConsumerRecords<K, V> records = consumer.poll(Duration.ofSeconds(1));
    for (var r : records) {
        process(r);
    }
    consumer.commitSync();
}

⚠️ อย่า process ช้าใน poll loop — ถ้า process > max.poll.interval.ms (default 5 min) → consumer ถูก kick ออกจาก group

แก้:

Process ใน thread pool แยก
เพิ่ม max.poll.interval.ms
ลด max.poll.records

yaml

# Classic protocol config (ก่อน KIP-848)
max.poll.records: 100
max.poll.interval.ms: 300000
session.timeout.ms: 45000
heartbeat.interval.ms: 3000
# 💡 ถ้าใช้ group.protocol=consumer (KIP-848) ไม่ต้องตั้ง heartbeat.interval —
#    broker-side coordinator จัด heartbeat ผ่าน separate channel

4.6 Spring Kafka Consumer

ในงานจริงไม่ต้องเขียน poll loop เอง — Spring Kafka ให้ใช้ @KafkaListener แล้ว framework จัดการ loop/thread/commit ให้ ตั้ง concurrency เพิ่ม consumer thread, ใช้ manual ack เพื่อ acknowledge หลัง process เสร็จ และ throw exception ให้ ErrorHandler จัดการ retry/DLT:

java

@KafkaListener(
    topics = "orders",
    groupId = "billing-service",
    concurrency = "3"                           // 3 consumer thread per app
)
public void onOrder(
    @Payload OrderEvent event,
    @Header(KafkaHeaders.RECEIVED_KEY) String key,
    @Header(KafkaHeaders.OFFSET) long offset,
    @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
    Acknowledgment ack) {

    try {
        process(event);
        ack.acknowledge();
    } catch (Exception e) {
        log.error("error at offset {}", offset, e);
        throw e;  // ให้ ErrorHandler จัดการ
    }
}

yaml

spring.kafka:
  consumer:
    group-id: billing-service
    auto-offset-reset: earliest
    enable-auto-commit: false
  listener:
    ack-mode: manual
    concurrency: 3

Part 5: Topic Configuration

5.1 สร้าง Topic

สร้าง topic ระบุ 3 อย่างหลัก: จำนวน partition (กำหนด parallelism), replication factor (กำหนด HA), และ config เช่น retention/compression ทำได้ทั้งผ่าน CLI (kafka-topics.sh) หรือประกาศเป็น @Bean ใน Spring Boot:

bash

kafka-topics.sh --bootstrap-server kafka:9092 --create \
  --topic orders --partitions 6 --replication-factor 3 \
  --config retention.ms=604800000 \
  --config min.insync.replicas=2 \
  --config compression.type=zstd
# ⚠️ broker-level compression.type ต้อง Kafka 2.1+ (broker recompress = CPU เปลือง)
#    แนะนำตั้ง compression ที่ producer-side แทน

หรือ Spring Boot:

java

@Bean
public NewTopic orders() {
    return TopicBuilder.name("orders")
        .partitions(6)
        .replicas(3)
        .config(TopicConfig.RETENTION_MS_CONFIG, "604800000")
        .config(TopicConfig.MIN_INSYNC_REPLICAS_CONFIG, "2")
        .build();
}

5.2 Partition Count — สำคัญที่สุดในการตัดสินใจ

กฎ:

จำนวน partition ≥ จำนวน consumer สูงสุดที่จะมี
เผื่อ scale (10-50 partition เริ่มต้น OK)
เปลี่ยนเพิ่มได้ แต่ ลดไม่ได้ — เลือกให้ดี
เพิ่ม partition → key hash เปลี่ยน → ordering พัง!

ตัวอย่าง: เริ่ม 12 partition + 3 broker = 4 partition/broker

5.3 Replication Factor

replication factor คือจำนวนสำเนาของแต่ละ partition ที่กระจายข้าม broker — ยิ่งมากยิ่งทนต่อ broker ล่ม ค่ามาตรฐานคือ RF=3 (ทน 1 broker ล่มได้และยังมี majority) ใช้คู่กับ min.insync.replicas เพื่อกัน data loss:

RF=1 — ไม่มี HA — broker ตาย = หาย
RF=2 — survive 1 broker down
RF=3 — recommended (survive 1 down + still have majority)
RF=5 — สำหรับ topic mission critical ที่ต้องการทน broker fail 2 ตัวภายใน DC เดียว (ไม่ใช่ multi-DC! cross-DC replication ใช้ MirrorMaker 2 หรือ Cluster Linking + RF=3 ภายในแต่ละ DC แทน — RF ข้าม DC = latency ระเบิด)

min.insync.replicas:

text

acks=all + min.insync.replicas=2 + RF=3
  → producer ต้องรอ 2/3 replica ack
  → broker ตาย 1 — ยัง publish ได้
  → broker ตาย 2 — block publish (กัน data loss)

5.4 Retention

properties

retention.ms = 604800000    # 7 days
retention.bytes = -1         # no size limit

หลัง retention → segment ลบ

Log compaction — เก็บ "ค่าล่าสุดของแต่ละ key":

properties

cleanup.policy=compact

ใช้กับ "current state" topic (เช่น user profile updates):

text

user-1: {name: Alice, age: 25}
user-2: {name: Bob, age: 30}
user-1: {name: Alice, age: 26}   ← จะแทน entry แรก

หลัง compact → topic เก็บแค่ latest per key

⚠️ Compaction ไม่ลบทันที — มีจังหวะของมัน:
Tombstone — ลบ key ใช้ส่ง message ค่า null (Kafka จะลบ key นี้หลัง delete.retention.ms)
segment.ms — segment ต้อง roll over ก่อน compactor ถึงจะแตะ
min.cleanable.dirty.ratio (default 0.5) — compactor วิ่งเมื่อ "ratio ของ dirty data" เกิน threshold
สรุป: เปิด cleanup.policy=compact แล้วลบ key ทันทีไม่ได้ ต้องรอ compactor วิ่ง

5.5 Important Topic Configs

รวม config ระดับ topic ที่ควรตั้งให้ถูกตั้งแต่แรก — retention/segment คุมว่าเก็บข้อมูลนานแค่ไหน, min.insync.replicas คุม durability, unclean.leader.election=false กันการเลือก leader จาก replica ที่ตามไม่ทัน (ซึ่งทำให้ข้อมูลหาย):

yaml

retention.ms: 604800000              # 7 days
retention.bytes: -1                   # unlimited size
segment.ms: 604800000                 # rollover segment ทุก 7 days
segment.bytes: 1073741824             # หรือ 1GB
compression.type: zstd
cleanup.policy: delete                # delete | compact
min.insync.replicas: 2
unclean.leader.election.enable: false # ไม่เลือก leader จาก non-ISR

Part 6: Exactly-Once Semantics (EOS)

6.1 ภายใน Kafka

EOS ทำงานได้เมื่อ:

Idempotent producer — ป้องกัน duplicate retry
Transactional producer — atomic write หลาย topic + offset
Isolation level ของ consumer = read_committed (ไม่อ่าน uncommitted)

java

// Producer
props.put("transactional.id", "tx-1");
props.put("enable.idempotence", "true");                    // default 3.0+
props.put("acks", "all");                                    // default 3.0+ เมื่อ idempotence=true

// Consumer
props.put("isolation.level", "read_committed");              // ข้าม uncommitted/aborted
props.put("enable.auto.commit", "false");                    // จำเป็นใน EOS — offset commit ต้องผ่าน transaction

💡 Kafka Streams: ใช้ processing.guarantee=exactly_once_v2 (ไม่ใช่ legacy exactly_once ที่ deprecate ตั้งแต่ 3.0) — Streams จัดการ tx ให้อัตโนมัติ ไม่ต้อง wire เอง

6.2 Stream Process EOS

รูปแบบที่ EOS เหมาะที่สุดคือ "consume → process → produce" ภายใน Kafka ล้วน — Spring จัดการให้การ commit offset ของ input กับการ send output เกิดใน transaction เดียวกัน (atomic) ทำให้ไม่มีทั้ง message หายและซ้ำในวง Kafka:

⚠️ EOS ไม่ได้แค่ใส่ @Transactional บน @KafkaListener — ต้อง wire KafkaTransactionManager เข้า ConcurrentKafkaListenerContainerFactory ผ่าน setTransactionManager() + ตั้ง consumer isolation.level=read_committed + enable.auto.commit=false + producer acks=all + enable.idempotence=true (default 3.0+)

java

// Listener container factory (ตัวคุมจริงว่า consume→process→produce atomic)
@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderIn> kafkaListenerContainerFactory(
        ConsumerFactory<String, OrderIn> consumerFactory,
        KafkaTransactionManager<String, OrderOut> txManager) {
    var factory = new ConcurrentKafkaListenerContainerFactory<String, OrderIn>();
    factory.setConsumerFactory(consumerFactory);              // consumer: isolation.level=read_committed, enable.auto.commit=false
    factory.getContainerProperties().setTransactionManager(txManager);  // 👈 จุดสำคัญ — ไม่มีอันนี้ไม่ atomic
    return factory;
}

// KafkaTemplate ต้องมี transactionIdPrefix (Spring จะ generate unique tx.id ต่อ instance)
@Bean
public KafkaTemplate<String, OrderOut> kafkaTemplate(ProducerFactory<String, OrderOut> pf) {
    pf.setTransactionIdPrefix("orders-tx-");                  // stable prefix + ordinal/shard
    return new KafkaTemplate<>(pf);
}

@KafkaListener(topics = "orders-in", containerFactory = "kafkaListenerContainerFactory")
public void process(ConsumerRecord<String, OrderIn> in) {
    OrderOut out = transform(in.value());
    kafkaTemplate.send("orders-processed", out);
    // container จัดการ commit offset ผ่าน sendOffsetsToTransaction() + producer.commitTransaction() = atomic
    // ❌ อย่าใส่ @Transactional("kafkaTransactionManager") ที่นี่ — ซ้อน tx ไม่ทำให้ atomic ขึ้น
}

→ "Consume → Process → Produce" atomic — รูปแบบเดียวกับที่ Kafka Streams ใช้ภายใน (processing.guarantee=exactly_once_v2)

6.3 ⚠️ EOS ครอบคลุมแค่ Kafka

ข้อจำกัดสำคัญที่ต้องเข้าใจ: EOS ของ Kafka การันตี atomic เฉพาะภายใน Kafka เท่านั้น — พอมี side effect ภายนอก (เขียน DB, เรียก HTTP, เขียนไฟล์) transaction ของ Kafka ครอบไปไม่ถึง ต้องใช้ Outbox pattern เสริมเพื่อให้ DB กับ event สอดคล้องกัน:

ถ้ามี external side effect (DB, HTTP, file):

text

Consume order → write DB → publish processed-order
                  │
              ถ้า DB fail หลัง consume ack → reprocess (DB อาจซ้ำ)
              ถ้า publish fail → DB rollback ไม่ได้

→ ต้อง Outbox Pattern (บท 12)

Part 7: Schema Registry

7.1 ปัญหา

เมื่อ producer กับ consumer คนละทีม/คนละรอบ deploy การที่ producer เปลี่ยน schema ของ message โดยไม่มีการควบคุมจะทำให้ consumer พัง — เพิ่ม field ยัง ignore ได้ แต่เปลี่ยน type หรือลบ field คือหายนะ จึงต้องมีกลไกบังคับความเข้ากันได้ของ schema:

text

Producer schema v1: {orderId, total}
Producer push: {orderId: 1, total: 99.5, currency: "USD"}    ← เพิ่ม field
Consumer v1: ignore field — OK

แต่ถ้า:

text

Producer change: total → string ("99.50")
Consumer v1: parse fail → crash

→ schema เปลี่ยนแบบไม่ control = pain

7.2 Schema Registry

Schema Registry คือบริการกลางเก็บ schema ทุกเวอร์ชัน — producer ลงทะเบียน schema แล้วแนบแค่ schema id ไปกับ message ส่วน consumer ดึง schema ตาม id มา parse ที่สำคัญคือมันบังคับ "compatibility rule" ทำให้ schema ใหม่ที่ทำ consumer พังถูกปฏิเสธตั้งแต่ตอน register:

text

Producer → register schema → Schema Registry
                                    │
                                    ↓
Producer write message + schema id (in message header)
                                    │
                                    ↓
Consumer ← read schema by id ← Schema Registry
Consumer parse message

Confluent Schema Registry (most popular)

⚠️ License 2026: Confluent Schema Registry เปลี่ยนเป็น Confluent Community License v2 (CCLv2) ตั้งแต่ Aug 2024 — ห้ามใช้แข่งกับ Confluent commercially (เช่น ขาย managed Schema Registry service). Self-hosted ภายในองค์กรยังใช้ได้ฟรี. ไม่ใช่ BSL มาตรฐาน (BSL มีวันหมดอายุแล้วเปลี่ยนเป็น open source, CCLv2 ไม่มี). ทางเลือก Apache-2 license:
Apicurio Registry (Red Hat, Apache-2) — รองรับ Avro / Protobuf / JSON Schema
Karapace (Aiven, Apache-2) — wire-compatible กับ Confluent Schema Registry API
AWS Glue Schema Registry (managed, ใช้กับ MSK)

Compatibility modes:

BACKWARD — new schema reads old data
FORWARD — old schema reads new data
FULL — both
NONE — anything

Registry ปฏิเสธ schema ที่ break compatibility

7.3 Avro Example

Avro เป็น format binary ยอดนิยมที่ใช้คู่ Schema Registry — นิยาม schema เป็นไฟล์ .avsc (มี field พร้อม type และ default) แล้วใช้ KafkaAvroSerializer ตอนส่ง ได้ payload เล็ก, ตรวจ schema, และ evolve ได้ปลอดภัย (field ใหม่ใส่ default ไว้):

order.avsc:

json

{
  "type": "record",
  "name": "Order",
  "namespace": "com.example",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "total", "type": "double"},
    {"name": "currency", "type": "string", "default": "USD"}
  ]
}

Producer:

yaml

spring.kafka:
  producer:
    value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
  properties:
    schema.registry.url: http://schema-registry:8081

java

kafka.send("orders", "key-1", Order.newBuilder()
    .setOrderId("1")
    .setCustomerId("c-1")
    .setTotal(99.5)
    .build());

→ binary, schema-checked, evolution-safe

7.4 Protobuf as alternative

yaml

spring.kafka.producer.value-serializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer

ใช้ proto file สำหรับ schema

Part 8: Kafka Connect

ETL framework สำหรับ Kafka:

text

[Source]                   [Kafka]                  [Sink]
PostgreSQL ──Source────────→ topic ──Sink────────→ Elasticsearch
S3         ──Source────────→ topic ──Sink────────→ BigQuery
MongoDB    ──Source────────→ topic ──Sink────────→ Snowflake

Pre-built connectors:

Debezium (CDC จาก PostgreSQL/MySQL/MongoDB/...)
JDBC source/sink
S3, GCS sink
Elasticsearch sink
HDFS, BigQuery, Snowflake, ฯลฯ

Config (REST):

bash

curl -X POST http://connect:8083/connectors -H 'Content-Type: application/json' -d '{
  "name": "users-pg-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.dbname": "shop",
    "database.user": "debezium",
    "plugin.name": "pgoutput",
    "table.include.list": "public.users,public.orders",
    "topic.prefix": "shop"
  }
}'
# 💡 plugin.name=pgoutput — Postgres 10+ ใช้ logical decoding built-in ไม่ต้องลง extension
#    (default ของ Debezium คือ decoderbufs ที่ต้อง install extension เพิ่ม → fail ตอน start)

→ ทุก row change ใน Postgres → Kafka topic

Part 9: MirrorMaker 2 + Multi-DC

text

DC-A Kafka ─── MirrorMaker 2 ───→ DC-B Kafka
              (active-passive)

หรือ

DC-A Kafka ⇄── MirrorMaker 2 ──⇄ DC-B Kafka
              (active-active)

ใช้ทำ:

Disaster recovery
Geographic locality
Aggregate ระหว่าง DC

Part 10: Cluster Operations

10.1 Sizing

การวางขนาด broker ของ Kafka มีจุดที่ต่างจากระบบอื่น — RAM ส่วนใหญ่ไม่ควรไปกองที่ JVM heap (หน่วยความจำที่ Java program ใช้เก็บ object ตอนรัน) แต่ควรปล่อยให้ OS page cache (พื้นที่ที่ระบบปฏิบัติการเก็บสำเนาไฟล์จาก disk ไว้ใน RAM เพื่ออ่านซ้ำได้เร็วโดยไม่ต้องแตะ disk จริง) ใช้แทน เพราะ Kafka อ่าน/เขียนผ่าน page cache นี้เป็นหลัก ดังนั้น heap ตั้งแค่ 6-8 GB ก็พอแล้ว ที่เหลือปล่อยให้ OS จัดการ ส่วน disk ควรเป็น SSD เร็ว ๆ และเครือข่ายแบนด์วิดท์สูง:

text

ขนาด broker
- CPU: 8-16 cores
- RAM: 32-64 GB (mostly for page cache, not heap!)
- Disk: NVMe SSD, RAID 10, ขนาดตาม retention × throughput
- Network: 10 Gbps+

Heap size: 6-8 GB (ใช้ G1 GC)
ที่เหลือใช้ page cache

10.2 Rack Awareness

ถ้าตั้ง broker.rack บอก Kafka ว่า broker ไหนอยู่ rack/AZ ใด Kafka จะกระจาย replica ของ partition ให้อยู่คนละ rack เสมอ — ผลคือถ้า AZ ทั้งโซนล่ม ก็ยังมี replica อีก AZ ให้บริการต่อได้:

yaml

broker.rack: us-east-1a

→ Kafka assign replica ข้าม rack → survive AZ failure

💡 KIP-392 (Fetch from Follower) GA ใน Kafka 2.4 (ธ.ค. 2019) — ตั้งแต่ Kafka 2.4 ขึ้นไป consumer อ่าน data จาก follower ใน same AZ ได้ (ลด cross-AZ traffic cost). config ฝั่ง consumer: client.rack=us-east-1a; ฝั่ง broker ต้องตั้ง broker.rack (ตั้งไว้แล้วด้านบน) + replica.selector.class=org.apache.kafka.common.replica.RackAwareReplicaSelector

10.3 Reassign Partition

เมื่อเพิ่ม/ลด broker หรือ load ไม่สมดุล ต้องย้าย partition ระหว่าง broker — ใช้ kafka-reassign-partitions.sh สร้างแผน (generate) แล้วสั่งทำ (execute) หรือใช้ Cruise Control ช่วย rebalance อัตโนมัติเพื่อลดงาน manual:

bash

# generate plan
kafka-reassign-partitions.sh --bootstrap-server kafka:9092 \
  --topics-to-move-json-file topics.json \
  --broker-list "0,1,2,3" \
  --generate

# execute
kafka-reassign-partitions.sh --bootstrap-server kafka:9092 \
  --reassignment-json-file plan.json \
  --execute

หรือใช้ Cruise Control — auto rebalance

10.4 Throttling Reassign

การย้าย partition ก้อนใหญ่กิน network/disk มาก อาจกระทบ traffic production ที่กำลังวิ่งอยู่ — จึงควรจำกัดความเร็ว (throttle) ของการ replicate ระหว่าง reassign เพื่อให้งานจริงไม่สะดุด:

ใหญ่ ๆ → throttle เพื่อไม่ impact production:

text

--throttle 50000000   # 50MB/s

10.5 Tiered Storage (KIP-405 — GA ใน Kafka 3.6, production-ready ใน 4.0)

KIP-405 ประกาศ GA ใน Kafka 3.6 (ต.ค. 2023) → matured ใน 3.9 (ต.ค. 2024) → production-ready ใน Kafka 4.0

แนวคิด:

Config:

bash

# broker level (Apache Kafka public API names — ตรวจกับ official docs ก่อน production)
remote.log.storage.system.enable=true
remote.log.storage.manager.class.name=<fully.qualified.PluginClass>
remote.log.storage.manager.impl.prefix=rsm.config.            # prefix สำหรับ config ที่ plugin อ่าน (เช่น rsm.config.s3.bucket.name)
remote.log.metadata.manager.class.name=org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager
# 💡 Plugin JAR ใส่ผ่าน broker classpath (เช่น /opt/kafka/libs/) — ไม่มี config property สำหรับ JAR path
# 💡 S3 plugin ไม่ได้มากับ Apache Kafka ต้อง download จาก community:
#    - Aiven (open source, Apache-2): https://github.com/aiven/tiered-storage-for-apache-kafka
#    - Confluent (commercial)
#    - AWS MSK (managed, Tiered Storage built-in)

# per topic
kafka-configs.sh --alter --topic orders \
  --add-config "remote.storage.enable=true,local.retention.ms=86400000,retention.ms=31536000000"
# local 1 day (hot tier), total 365 days
# ⚠️ consumer ที่ reset offset > 1 day จะไปอ่าน cold tier — latency 100-500x ของ hot

ประโยชน์:

ลด storage cost 80%+ (S3 cheap, broker SSD แพง)
Retention เป็นปี (ทำได้แต่จะแพงถ้าเก็บใน broker disk)
Broker reboot เร็วขึ้น (น้อย segment ต้อง re-validate)
Replay historical data ทำได้แม้เกิน hot window

ข้อจำกัด + Production trade-off:

📖 gloss คำที่ใช้บ่อยในหัวข้อนี้: hot tier = ข้อมูลล่าสุดเก็บใน disk ของ broker (เร็ว), cold tier = ข้อมูลเก่าย้ายไป S3 (ช้ากว่าแต่ถูกกว่า) · egress = ค่าธรรมเนียม data ที่ "ออกจาก" cloud provider (เช่น ดึงจาก S3 ข้าม region) · RPO (Recovery Point Objective — ข้อมูลย้อนหลังสูงสุดที่ยอมเสียได้ตอนกู้ระบบ)

Cold-tier latency ช้ากว่า hot 100-500 เท่า: อ่าน hot < 1ms vs cold 10-500ms (ขึ้นกับ S3 endpoint + segment size) — ถ้า consumer reset offset ไปก่อน hot window จะเจอ latency spike ทันที
Lag spike เมื่อ replay เก่า: consumer ที่ rebuild state จาก beginning จะเห็น lag พุ่งจาก 0 → ล้านระหว่าง replay ต้องตั้ง monitor/alert ให้รู้ว่านี่คือ "intentional replay" ไม่ใช่ outage
Compacted topic ยังไม่ support (KIP-1018 work in progress — ยังไม่ GA ใน Kafka 4.0; ตรวจ KIP index ก่อนใช้) — __consumer_offsets + key-based state store ยัง tier ไม่ได้
S3 cost trap (ค่าใช้จ่ายแอบแฝง):
- GET request fee — ทุก fetch จาก cold = 1 S3 GET ($0.0004/1000 reqs สะสมเร็วถ้า partition เยอะ)
- Egress fee ข้าม region — S3 ใน us-east-1, broker ใน us-west-2 = $0.02/GB
- List request — broker periodic list cold segments → ตั้ง remote.log.metadata.manager ให้ cache
ตัวเลือก storage backend: S3 (most popular), GCS, Azure Blob, HDFS, MinIO — แต่ละตัว plugin ของชุมชนแยกกัน (Confluent, Aiven, AWS MSK) — ตรวจว่า plugin ที่ใช้รองรับ remote.log.copier.thread.pool.size tunable ไหม
RPO เสี่ยงหาก S3 ล่ม: ถ้า region S3 ล่ม consumer ยังอ่าน hot ได้ปกติ แต่ historical query (CDC replay, audit) จะ fail จนกว่า S3 ฟื้น — ต้องรวมเคสนี้ใน DR plan
ไม่เปิดทุก topic: เปิดเฉพาะ topic ที่ retention > 7 วันและ access pattern เป็น "write hot, read rarely" (audit log, event sourcing). Topic latency-critical (real-time alerting) ปิดไว้

When to enable:

Retention > 30 วัน + storage cost เริ่มเป็นค่าใช้จ่ายหลัก
Compliance ต้องเก็บ ≥ 7 ปี (GDPR audit, SOX)
Multi-region replay จาก single source of truth

When NOT to enable:

Latency-sensitive consumer ที่ reset offset บ่อย
Topic ขนาดเล็ก (< 1 TB total) — overhead ไม่คุ้ม
Cluster ที่ไม่มี platform team ตรวจ S3 cost dashboard เป็นนิสัย

ปัญหาเดิมของ Kafka: ordering ผูกกับ partition → ถ้าอยาก queue semantics (อ่านได้พร้อมกันมากกว่า partition count) ต้องไป RabbitMQ

Share Groups = queue บน Kafka จริง ๆ:

text

Topic "tasks" 3 partitions:
  Classic consumer group:        Share group:
  - max 3 consumers              - unlimited consumers
  - 1 partition → 1 consumer     - any consumer pulls any message
  - ordering per partition       - no ordering guarantee
  - offset-based                 - per-message ack

Use case:

⚠️ นี่คือ consumer API คนละแบบ ไม่ใช่ส่วนขยายของ KafkaConsumer ใน Part 4 — Share Group ใช้ class ใหม่ KafkaShareConsumer ที่ไม่มี partition assignment และ ack เป็นรายข้อความ (AcknowledgeType.ACCEPT/RELEASE) แทนการ commit offset แบบเดิม อย่าสับสนกับ consumer group ที่เพิ่งเรียนมา

java

// Kafka 4.0 — KIP-932 Early Access
// ⚠️ Share groups ไม่ได้ใช้ group.protocol=share (group.protocol รับเฉพาะ classic/consumer สำหรับ KIP-848)
// ⚠️ ใช้ class แยก: KafkaShareConsumer (API ใหม่ ไม่ใช่ KafkaConsumer เดิม)
// ⚠️ Property names ใน EA อาจเปลี่ยนก่อน GA — ตรวจ Apache Kafka 4.0 client docs + KIP-932 ก่อนนำขึ้น production
Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("group.id", "image-processors");
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());

KafkaShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props);  // 👈 class ใหม่
consumer.subscribe(List.of("image-tasks"));

while (true) {
    var records = consumer.poll(Duration.ofSeconds(1));
    for (var r : records) {
        try {
            processImage(r.value());
            consumer.acknowledge(r, AcknowledgeType.ACCEPT);   // 👈 per-message ack (EA API)
        } catch (Exception e) {
            consumer.acknowledge(r, AcknowledgeType.RELEASE);  // give back to share pool
        }
    }
    consumer.commitSync();
}

→ Scale image processor เป็น 100 instance ได้แม้ topic มีแค่ 3 partition

เมื่อใช้:

Task queue (image processing, video encode)
Email / SMS dispatcher
Webhook delivery
Work where ordering ไม่สำคัญ + ต้องการ horizontal scale สูง

ตอนนี้ยัง Early Access ใน Kafka 4.0 (ไม่ใช่ default) — production ใช้ classic consumer group หรือ RabbitMQ ไปก่อน, monitor KIP-932 ที่จะ GA ใน Kafka 4.1/4.2

Part 11: Performance Tuning

11.1 Producer

จูน producer เพื่อเพิ่ม throughput — เพิ่ม linger/batch ให้รวม message เป็นก้อนใหญ่ขึ้น, เปิด compression (zstd), และตั้ง buffer ให้พอรองรับ burst ค่าด้านล่างเป็นจุดตั้งต้นที่ใช้ได้จริง ปรับตาม workload:

properties

linger.ms: 20-100
batch.size: 64KB-128KB
compression.type: zstd
buffer.memory: 64MB-256MB
max.in.flight.requests.per.connection: 5

11.2 Consumer

ฝั่ง consumer จูนเรื่องการดึงข้อมูล — fetch.min.bytes/fetch.max.wait.ms ทำให้รอสะสมก่อนดึง (ลดจำนวน round-trip), max.poll.records คุมจำนวน record ต่อรอบให้ process ทันใน interval:

properties

fetch.min.bytes: 1KB-1MB
fetch.max.wait.ms: 500
max.partition.fetch.bytes: 1MB-10MB
max.poll.records: 500-1000

11.3 Broker

ระดับ broker จูนจำนวน thread (network/io) และขนาด socket buffer ให้รองรับ throughput ที่ต้องการ รวมถึง num.replica.fetchers ที่กำหนดความเร็วในการ replicate ระหว่าง broker:

properties

num.network.threads: 8-16
num.io.threads: 16-32
socket.send.buffer.bytes: 100KB-1MB
socket.receive.buffer.bytes: 100KB-1MB
num.replica.fetchers: 4-8
log.flush.interval.messages: rely on replication

11.4 OS

Kafka อาศัย OS page cache อย่างหนัก ดังนั้นต้องตั้ง OS ให้เหมาะ — ปิด swap (กันไม่ให้ memory ที่ควรเป็น cache ถูกสลับลง disk), เพิ่ม limit ของ open files, และจูน parameter ของ page cache ให้ Kafka ใช้ RAM เต็มประสิทธิภาพ:

bash

# Disable swap
swapoff -a

# Increase open files
ulimit -n 100000

# Page cache tuning
sysctl vm.swappiness=1
sysctl vm.dirty_ratio=10                 # ห้ามตั้งสูง (เช่น 80) — เสี่ยง fsync stall ใหญ่
sysctl vm.dirty_background_ratio=5       # flush ทำงานเร็วขึ้น ลด spike

11.5 JVM

heap ของ Kafka ไม่ต้องใหญ่ (6-8 GB) เพราะข้อมูลจริงอยู่ที่ page cache — สิ่งที่สำคัญกว่าคือ GC ที่ pause สั้น ใช้ G1GC ตั้ง MaxGCPauseMillis ต่ำ ๆ เพื่อไม่ให้ broker หยุดนานจนกระทบ latency:

bash

-Xms6g -Xmx6g
-XX:+UseG1GC
-XX:MaxGCPauseMillis=50            # 50-100ms เหมาะสำหรับ Kafka heap 6GB
                                    # ⚠️ ตั้งต่ำเกินไป (เช่น 20) → G1 ทำ young GC ถี่ → GC overhead สูง / thrash
# -XX:+ExplicitGCInvokesConcurrent  # legacy CMS-era flag — ส่วนใหญ่ no-op กับ G1, ตัดออกได้

# JDK 17/21 unified logging (แทน -XX:+PrintGCDetails เดิม)
-Xlog:gc*:file=/var/log/kafka/gc.log:time,uptime:filecount=10,filesize=100M

Part 12: Monitoring

12.1 JMX Metrics

Kafka expose metric ผ่าน JMX จำนวนมาก แต่มีไม่กี่ตัวที่ "ต้องเฝ้าเสมอ" — UnderReplicatedPartitions และ OfflinePartitionsCount ควรเป็น 0 ตลอด (ถ้าไม่ใช่ = มี replica ตามไม่ทันหรือ partition ใช้งานไม่ได้ = สัญญาณอันตราย):

ของสำคัญ:

kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions ← 0 always
kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec
kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec
kafka.controller:type=KafkaController,name=OfflinePartitionsCount ← 0 always
kafka.controller:type=KafkaController,name=ActiveControllerCount ← = 1 ต่อ cluster เสมอ (ผลรวมข้าม broker; ≠1 = split brain / no controller)
kafka.server:type=KafkaRequestHandlerPool,name=RequestHandlerAvgIdlePercent ← > 0.3 healthy (ใกล้ 0 = broker saturate)
kafka.server:type=ReplicaFetcherManager,name=MaxLag ← <1000

12.2 Consumer Lag

consumer lag คือ metric ที่บอกสุขภาพ consumer ได้ดีที่สุด — วัดว่า consumer อ่านตามหลัง producer อยู่กี่ message ถ้า lag โตเรื่อย ๆ แปลว่า process ไม่ทัน ต้อง scale consumer หรือหาคอขวด:

Lag = (latest offset) - (committed offset)

bash

kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --describe --group billing-service

GROUP    TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
billing  orders  0          12345           12400           55
billing  orders  1          11200           11250           50

📖 CURRENT-OFFSET = ตำแหน่งที่ consumer อ่านถึงล่าสุด, LOG-END-OFFSET = message ล่าสุดที่มีอยู่จริงใน partition — ส่วนต่างระหว่างสองค่านี้คือ lag

→ alert ที่ lag > 10000 ค้างเกิน 5 นาที

12.3 Tools

มีเครื่องมือสำเร็จรูปช่วย monitor และ operate Kafka — Burrow เฝ้า lag, Cruise Control ช่วย rebalance, UI อย่าง Kafka UI/Conduktor ดูสถานะแบบ visual, และ Prometheus + JMX exporter + Grafana สำหรับ metric ระยะยาว:

Burrow — consumer lag monitoring (legacy — minimal maintenance ตั้งแต่ 2021)
kminion (Redpanda, modern open source) — lag + cluster metrics สำหรับ Prometheus
kafka_exporter — Prometheus exporter ที่ใช้กันแพร่หลาย
Cruise Control — auto rebalance + capacity
Confluent Control Center — full UI (paid)
Kafka UI / Conduktor — open source UI
Prometheus + JMX exporter + Grafana

Part 13: Security

13.1 SSL/TLS

ชั้นแรกของ security คือเข้ารหัสการรับส่งข้อมูลด้วย SSL/TLS — กัน traffic ระหว่าง client กับ broker ถูกดักอ่าน ตั้งค่าโดยชี้ truststore (เชื่อ cert ของ broker) และ keystore (cert ของ client ถ้าใช้ mutual TLS):

yaml

spring.kafka:
  properties:
    security.protocol: SSL
    ssl.truststore.location: /etc/kafka/truststore.jks
    ssl.truststore.password: ${TRUSTSTORE_PASSWORD}
    ssl.keystore.location: /etc/kafka/keystore.jks
    ssl.keystore.password: ${KEYSTORE_PASSWORD}

13.2 SASL Authentication

SSL เข้ารหัสช่องทาง แต่ SASL คือการ "พิสูจน์ตัวตน" ว่า client เป็นใคร — รองรับหลาย mechanism (SCRAM, Kerberos, OAuth) ใช้คู่กับ SSL (SASL_SSL) เพื่อให้ทั้งเข้ารหัสและยืนยันตัวตนพร้อมกัน:

yaml

spring.kafka:
  properties:
    security.protocol: SASL_SSL
    sasl.mechanism: SCRAM-SHA-512
    sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="${USER}" password="${PASSWORD}";

Mechanisms: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI (Kerberos), OAUTHBEARER (with KIP-768 JWT validation ตั้งแต่ Kafka 3.1 = modern cloud-native standard, ใช้กับ AWS MSK IAM / Confluent Cloud SSO / Azure AD)

13.3 ACL

หลังรู้ว่า client เป็นใคร (authentication) ขั้นต่อไปคือคุมว่าใครทำอะไรได้ (authorization) ผ่าน ACL — กำหนดสิทธิ์รายคน เช่น billing-service อ่าน topic orders ได้แต่เขียนไม่ได้ ยึดหลัก least privilege ให้สิทธิ์เท่าที่จำเป็น:

bash

kafka-acls.sh --bootstrap-server kafka:9092 \
  --add --allow-principal User:billing-service \
  --operation Read --topic orders

text

Operations: Read, Write, Create, Delete, Alter, Describe, ClusterAction
Resources: Topic, Group, Cluster, TransactionalId

Part 14: Pitfalls + Anti-Patterns

14.1 Too Few Partitions

ตั้ง partition น้อยเกินไปจำกัด parallelism — เพราะ 1 partition อ่านได้แค่ 1 consumer ต่อ group ดังนั้นถ้ามี 3 partition ก็ scale consumer ได้สูงสุด 3 ตัว เพิ่มกว่านั้นก็นั่งว่าง กลายเป็นคอขวด:

= ไม่ scale consumer → bottleneck

14.2 Too Many Partitions

ในทางกลับกัน partition เยอะเกินก็มีต้นทุน — กิน metadata, file handle และทำ rebalance ช้าลง อย่าตั้งเผื่อแบบสุดโต่ง ให้อยู่ในช่วง sweet spot ตามจำนวน broker และ throughput จริง:

Metadata overhead
File handle pressure
Rebalance ช้า

→ 100-1000 partition per broker เป็น sweet spot

14.3 No Compression

ไม่เปิด compression ทำให้เปลือง network และ disk หลายเท่าโดยไม่จำเป็น — โดยเฉพาะ payload ที่เป็น text/JSON ที่บีบอัดได้ดีมาก เปิด zstd/lz4 แทบไม่กระทบ CPU แต่ลดต้นทุนได้มหาศาล:

→ network + disk × 5-10

14.4 Auto-commit + Long process

ถ้า process แต่ละ message นานเกิน max.poll.interval.ms Kafka จะคิดว่า consumer ตายแล้ว kick ออกจาก group → rebalance → message ถูก process ซ้ำ ทางแก้คือย้ายงานหนักไป thread pool แยกแล้วใช้ manual ack ไม่ block poll loop:

java

@KafkaListener
public void process(Event e) {
    sleep(60_000);             // process > poll interval
    // → consumer kicked, partition rebalanced, message reprocessed
}

→ ใช้ manual ack + thread pool

14.5 No Replication / RF=1

ตั้ง RF=1 หมายถึงไม่มีสำเนาเลย — broker ตัวที่เก็บ partition นั้นล่มเมื่อไหร่ ข้อมูลหายทันทีและ partition ใช้งานไม่ได้ production ต้องตั้ง RF อย่างน้อย 3 เสมอ:

→ broker ตาย = data หาย

14.6 No min.insync.replicas

กับดักที่เข้าใจผิดบ่อย: ตั้ง acks=all แต่ไม่ตั้ง min.insync.replicas — ถ้าตอนนั้นเหลือแค่ leader ตัวเดียวใน ISR "all" ก็แปลว่า ack แค่ leader ทำให้ยังเสี่ยงข้อมูลหาย ต้องตั้ง min.insync.replicas=2 ควบคู่กันเสมอ:

→ acks=all = ack แค่ leader = data loss possible

14.7 Single consumer group สำหรับทุกอย่าง

ใช้ consumer group เดียวรวมงานทุกชนิดทำให้ scale แยกกันไม่ได้ — งานที่ต้องการ throughput ต่างกันถูกผูกติดกัน และ offset ปะปนกันจน reasoning ยาก ควรแยก group ตามหน้าที่ (billing, audit, analytics) เป็นอิสระต่อกัน:

→ ไม่สามารถ scale independently, ปะปนกัน

14.8 Hot Partition

เลือก partition key ไม่ดีทำให้ traffic กระจุกตัวที่ partition เดียว (hot partition) — เช่นใช้ country code แล้ว US ส่งเยอะสุดก็ถล่มลง P0 ตัวเดียว ส่วนที่เหลือว่าง แก้โดยใช้ composite key ที่กระจายตัวดีกว่า (country + userId):

text

Partition key = country code
Most traffic from US → P0 hot

→ ใช้ composite key (country + userId) สำหรับ distribution

14.9 ไม่ใช้ Schema Registry

ไม่ใช้ Schema Registry เท่ากับฝากความหวังว่า producer จะไม่เปลี่ยน schema แบบ breaking — ซึ่งไม่ยั่งยืน วันที่มีคนเปลี่ยน type หรือลบ field consumer จะพังพร้อมกันกลายเป็น incident จริง ควรบังคับ compatibility ตั้งแต่ต้น:

→ ตอน schema break = production incident

Part 15: Lab

Lab 1: Single broker + first message

เริ่มจากวงจรพื้นฐานสุด — รัน Kafka 1 broker ด้วย docker compose แล้วลองส่ง message จาก producer ให้ consumer พิมพ์ออกมา เพื่อเห็นการไหลของ message ก่อนเข้าเรื่อง partition/group:

docker compose → producer send → consumer print

Lab 2: Partition + consumer group

ฝึกเรื่องการ scale — สร้าง topic 6 partition แล้วรัน 3 consumer ในกลุ่มเดียวกัน สังเกตว่า Kafka แบ่ง partition ให้แต่ละ consumer อย่างไร และเกิดอะไรขึ้นเมื่อ consumer เข้า/ออก (rebalance):

create topic 6 partition → 3 consumer ใน group → ดู assignment

Lab 3: Idempotent + EOS

ทดลอง exactly-once จริง — ตั้ง producer ให้มี transactional.id และ consumer อ่านแบบ read_committed แล้วลองจำลอง failure ดูว่าระบบไม่เกิด duplicate และไม่อ่าน message ที่ยังไม่ commit:

Producer with transactional.id + consumer with read_committed

Lab 4: Schema Registry + Avro

พิสูจน์ว่า Schema Registry ป้องกัน incident จริง — ตั้ง registry, ส่ง message แบบ Avro, ให้ consumer parse ได้ปกติ แล้วลองจงใจเปลี่ยน schema แบบ breaking เพื่อดูว่า registry ปฏิเสธตั้งแต่ตอน register:

setup registry → producer Avro → consumer parse → break compatibility → fail

Lab 5: Debezium CDC

ลอง Change Data Capture — ให้ Debezium จับทุกการเปลี่ยนแปลงในตาราง Postgres แล้วส่งเข้า Kafka topic อัตโนมัติ จากนั้น consume ดู เพื่อเข้าใจว่า CDC ทำให้ระบบอื่นรับรู้การเปลี่ยนแปลงของ DB ได้แบบ real-time โดยไม่ต้องแก้ app เดิม:

Postgres → Debezium → Kafka topic → consume

Lab 6: Reassign partition

ทดสอบความทนทานและการ rebalance — รัน 3 broker, kill ไป 1 ตัวแล้วดู ISR หดตัว (replica ที่ตามไม่ทันหลุดออก) จากนั้น restart broker ดูมัน catch up กลับเข้า ISR เพื่อเข้าใจ failover ของ Kafka:

3 broker → kill 1 → ดู ISR shrink → restart → catch up

Part 15.5: Production War Stories

⚠️ หมายเหตุ: เรื่องเล่าด้านล่างผสมระหว่าง pattern ทั่วไปที่เคยเห็นจริงตอน debug + เรื่องที่มีการพูดถึงในวงการ — ไม่ใช่ทุกเคสมี public postmortem โดยตรงจากบริษัทที่ระบุชื่อ. ใช้เป็น learning lesson ได้ แต่ถ้าจะอ้างต่อในเอกสารทางการ ให้หา source ทางการก่อน

🔥 Pattern เคส 1 (อ้างชื่อ Cloudflare R2, 2023): partition imbalance ทำ throughput ตก 60%

Producer ใช้ null key → sticky partitioner เลือก 1 partition นาน → broker leader นั้น overload, ที่เหลือ idle. ISR shrink → block publish

Lesson:

ตรวจ partition skew ด้วย kafka-topics --describe + monitor BytesInPerSec per partition
ใช้ key ที่ distribute ดี (composite: country:userId)
ตรวจว่า partitioner.adaptive.partitioning.enable=true (default ตั้งแต่ 3.3 — แค่ verify ว่าไม่ถูก disable) — ปรับตาม latency อัตโนมัติ

🔥 Pattern เคส 2 (รวบรวมจาก e-commerce หลายเจ้า ช่วง Black Friday): consumer lag ระดับล้าน message

ตั้ง max.poll.records=500 + กระบวนการเขียน DB 100ms/record → 1 poll ใช้ 50 วินาที → max.poll.interval.ms 5 นาที = OK แต่ throughput แค่ 10 msg/s ต่อ thread

Lesson:

เพิ่ม partition (จาก 12 → 48) เพื่อให้ scale consumer ขึ้นไปได้
Process แบบ async batch (จัด batch insert DB)
Monitor lag + auto-scale consumer pod ตาม lag (KEDA + Kafka scaler)

🔥 Pattern เคส 3 (อ้างเป็นเหตุการณ์ Robinhood 2021): ZooKeeper outage ทำ trading หยุดนาน

ZK ensemble session expire ระหว่าง GC pause ใหญ่ → controller election storm → topic metadata ปั่นป่วน → producer ทุกตัว error

Lesson:

Migrate ไป KRaft (Kafka 3.3+) — ปัญหานี้ไม่เกิดอีก
ถ้ายังใช้ ZK: tune GC (G1, low pause), แยก disk จาก Kafka
Disaster runbook: producer ต้องมี local buffer + retry exponential

🔥 Pattern เคส 4 (รวบรวมจาก SaaS หลายเจ้า — generic pattern, ไม่ระบุชื่อบริษัทเพราะ public postmortem ไม่ชัด): exactly-once + outbox ผิด → duplicate publish

ใช้ transactional.id ที่ผูกกับ pod hostname → pod restart = id ใหม่ → producer ID เก่า fence ไม่ได้ → duplicate batch publish

Lesson:

transactional.id ต้อง stable per logical producer (ไม่ใช่ per pod) — ใช้ StatefulSet ordinal หรือ service+shard
เปิด enable.idempotence=true + acks=all ทุกครั้ง
Outbox pattern (บท 12) คุ้มค่ากว่า raw EOS สำหรับ business

Part 15.6: 📋 Cheat Sheet

Producer config (production baseline)

yaml

acks: all                          # default ตั้งแต่ 3.0 เมื่อ idempotence=true
enable.idempotence: true           # default ตั้งแต่ 3.0
max.in.flight.requests.per.connection: 5   # safe กับ idempotence=true (เกินจะ ConfigException)
retries: 2147483647
delivery.timeout.ms: 120000
linger.ms: 20
batch.size: 65536
compression.type: zstd
buffer.memory: 67108864
partitioner.adaptive.partitioning.enable: true   # default ตั้งแต่ 3.3 — แค่ verify ไม่ถูก disable
# ถ้า EOS:
transactional.id: <stable-id-per-logical-producer>

Consumer config (production baseline)

yaml

group.id: <service>-<purpose>
group.protocol: classic           # 👈 ยังเป็น default. KIP-848 "consumer" ยัง Early Access ใน 4.0 — รอ 4.1/4.2 GA ก่อน opt-in production
partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor  # incremental rebalance ลด STW
enable.auto.commit: false
auto.offset.reset: earliest
isolation.level: read_committed   # 👈 default เมื่อ consume topic ที่ producer ใช้ transactional.id (EOS-aware stack)
                                  # ถ้าไม่มี transactional producer ในเส้นทาง = read_uncommitted (default) ก็พอ
                                  # (read_committed มี cost: consumer รอจน producer commit transaction)
group.instance.id: <stable-id>    # static membership ลด rebalance สำหรับ stateful consumer (Kafka 2.3+)
max.poll.records: 100
max.poll.interval.ms: 300000
fetch.min.bytes: 10240
fetch.max.wait.ms: 500

Topic config (production baseline)

yaml

partitions: 12                    # เผื่อ scale 4x
replication.factor: 3
min.insync.replicas: 2
retention.ms: 604800000           # 7 days
cleanup.policy: delete            # หรือ compact สำหรับ current-state
compression.type: zstd
unclean.leader.election.enable: false
# ถ้า cold data ยาว:
remote.storage.enable: true
local.retention.ms: 86400000

Decision matrix

ปัญหา	ใช้
Order matters, scale ปานกลาง	Classic consumer group + key partition
Need queue (no order), scale สูง	Share Groups (KIP-932) หรือ RabbitMQ
Atomic "consume → DB → produce"	Outbox (บท 12), อย่าใช้ raw EOS ข้ามระบบ
Many consumer ของ event เดียวกัน	Multiple consumer groups
Retention เป็นปี	Tiered Storage
Cross-DC replication	MirrorMaker 2 หรือ Cluster Linking (Confluent)
Audit / replay	Kafka เก็บนาน + new consumer group `earliest`
Schema evolution	Schema Registry + Avro/Proto เสมอ

Commands ใช้บ่อย

bash

# describe topic
kafka-topics.sh --bootstrap-server kafka:9092 --describe --topic orders

# describe consumer group + lag
kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group billing

# reset offset (only when consumer stopped!)
kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --group billing --topic orders --reset-offsets --to-datetime 2026-05-23T00:00:00 --execute

# alter config
kafka-configs.sh --bootstrap-server kafka:9092 --alter \
  --entity-type topics --entity-name orders --add-config retention.ms=2592000000

# consume from beginning (debug)
kafka-console-consumer.sh --bootstrap-server kafka:9092 \
  --topic orders --from-beginning --property print.key=true

Alert ที่ต้องมี

Metric	Threshold
`UnderReplicatedPartitions`	> 0 for 5 min
`OfflinePartitionsCount`	> 0 (any)
Consumer lag	> 10000 for 5 min
Controller count	≠ 1 (split brain)
`RequestHandlerAvgIdlePercent`	< 0.3 (busy broker)
Disk usage	> 80%
Network throughput	> 70% of NIC capacity

Part 16: Checkpoint

ทำไม Kafka pull-based ดีกว่า push สำหรับ throughput?
Partition คืออะไร? Replication factor 3 หมายความว่ายังไง?
ISR คืออะไร? min.insync.replicas ทำอะไร?
KRaft ต่าง ZooKeeper ยังไง? Kafka 4.0 เปลี่ยนอะไรสำคัญ?
acks=all + enable.idempotence=true ทำงานยังไงด้วยกัน?
Consumer Group กับ partition assignment?
Rebalance STW vs Cooperative Sticky vs KIP-848 (broker-side)?
Log compaction ใช้ตอนไหน?
EOS ครอบคลุม external side effect ไหม? ทำไม outbox จำเป็น?
Schema Registry แก้ปัญหาอะไร? BACKWARD vs FORWARD compat ต่างกันอย่างไร?
Tiered Storage ลด cost ยังไง? ข้อแลกเปลี่ยนคืออะไร?
Share Groups (KIP-932) ต่างจาก consumer group เดิมยังไง? ใช้กรณีไหน?
transactional.id ควรตั้งจากอะไร? ทำไมไม่ใช่ hostname?
Partition skew (hot partition) ตรวจ + แก้ยังไง?

Part 17: สรุปบทนี้

Kafka 4.0 (2025) = KRaft only — ZooKeeper code ลบหมดแล้ว
Kafka = distributed log — partition + replica + offset, sequential write + zero-copy
Producer: acks=all + idempotent (default 3.0+) + batching + compression (zstd)
Consumer: group + KIP-848 protocol (broker-side rebalance — Early Access ใน 4.0, คาด GA ใน 4.1/4.2) + manual commit
Topic config: partition count (เผื่อ scale), RF=3, min.insync=2
EOS ใน Kafka — ครอบคลุมแค่ Kafka. External side effect → ใช้ Outbox pattern (บท 12)
Schema Registry + Avro/Proto = bullet-proof evolution
🆕 Tiered Storage (KIP-405, GA 3.6, production-ready 4.0) — cold data ใน S3 → retention ยาว + cost ต่ำ
🆕 Share Groups (KIP-932 — Early Access ใน 4.0) — queue semantics บน Kafka, scale > partition
Kafka Connect = ETL — Debezium, S3, Elasticsearch
MirrorMaker 2 = multi-DC
Monitor: lag, under-replicated, offline partition, partition skew

บทถัดไป — Other Brokers (Redis Streams, NATS, Pulsar, SQS/SNS, GCP Pub/Sub)

← บทที่ 07 | สารบัญ | บทที่ 09: Other Brokers →

🔤 Glossary · 📋 Style guide · 📅 last_verified: 2026-06-03

บทที่ 08 — Apache Kafka (คาฟ-ก้า) Deep (เข้มข้นสุดของหนังสือ) ​

Part 1: Kafka คืออะไร ​

1.1 ภาพรวม ​

1.2 เมื่อไหร่ใช้ Kafka ​

1.3 ติดตั้ง (KRaft mode — Kafka 4.0+ ไม่มี ZooKeeper อีกแล้ว) ​

Part 2: Architecture ​

2.1 Concepts ​

Topic — "หัวข้อ stream ของ event" ​

Partition — "หัวใจของ Kafka scaling" ​

Replica — "ก็อบสำรอง" ​

ISR (In-Sync Replica) — "Replica ที่ตามทัน" ​

Offset — "หมายเลขหน้าที่อ่านถึง" ​

🎯 สรุป Mental Model ​

2.2 Storage ​

2.3 KRaft (ZooKeeper ตายแล้ว) ​

Part 3: Producer ​

3.1 Code พื้นฐาน ​

3.2 Acks ​

3.3 Idempotent Producer ​

3.4 Batching + Linger ​

3.5 Compression ​

3.6 Partitioner ​

3.7 Transactional Producer (EOS — Exactly-Once) ​

3.8 Producer Settings Reference ​

Part 4: Consumer ​

4.1 Basic ​

4.2 Consumer Group ​

4.3 Rebalance ​

Rebalance Protocols (Classic, ผ่าน client-side coordinator) ​

🆕 KIP-848 — New Consumer Group Protocol (Kafka 4.0 Early Access) ​

4.4 Offset Management ​

Auto Commit (ไม่แนะนำ) ​

Manual Commit (แนะนำ) ​

Offset Reset ​

4.5 Poll Loop ที่ถูก ​

4.6 Spring Kafka Consumer ​

Part 5: Topic Configuration ​

5.1 สร้าง Topic ​

5.2 Partition Count — สำคัญที่สุดในการตัดสินใจ ​

5.3 Replication Factor ​

5.4 Retention ​

5.5 Important Topic Configs ​

Part 6: Exactly-Once Semantics (EOS) ​

6.1 ภายใน Kafka ​

6.2 Stream Process EOS ​

6.3 ⚠️ EOS ครอบคลุมแค่ Kafka ​

Part 7: Schema Registry ​

7.1 ปัญหา ​

7.2 Schema Registry ​

7.3 Avro Example ​

7.4 Protobuf as alternative ​

Part 8: Kafka Connect ​

Part 9: MirrorMaker 2 + Multi-DC ​

Part 10: Cluster Operations ​

10.1 Sizing ​

10.2 Rack Awareness ​

10.3 Reassign Partition ​

10.4 Throttling Reassign ​

10.5 Tiered Storage (KIP-405 — GA ใน Kafka 3.6, production-ready ใน 4.0) ​

10.6 🆕 Share Groups / Queues for Kafka (KIP-932 — Early Access ใน Kafka 4.0) ​

Part 11: Performance Tuning ​

11.1 Producer ​

11.2 Consumer ​

11.3 Broker ​

11.4 OS ​

11.5 JVM ​

Part 12: Monitoring ​

12.1 JMX Metrics ​

12.2 Consumer Lag ​

12.3 Tools ​

Part 13: Security ​

13.1 SSL/TLS ​

13.2 SASL Authentication ​

13.3 ACL ​

Part 14: Pitfalls + Anti-Patterns ​

14.1 Too Few Partitions ​

14.2 Too Many Partitions ​

14.3 No Compression ​

14.4 Auto-commit + Long process ​

14.5 No Replication / RF=1 ​

บทที่ 08 — Apache Kafka (คาฟ-ก้า) Deep (เข้มข้นสุดของหนังสือ)

Part 1: Kafka คืออะไร

1.1 ภาพรวม

1.2 เมื่อไหร่ใช้ Kafka

1.3 ติดตั้ง (KRaft mode — Kafka 4.0+ ไม่มี ZooKeeper อีกแล้ว)

Part 2: Architecture

2.1 Concepts

Topic — "หัวข้อ stream ของ event"

Partition — "หัวใจของ Kafka scaling"

Replica — "ก็อบสำรอง"

ISR (In-Sync Replica) — "Replica ที่ตามทัน"

Offset — "หมายเลขหน้าที่อ่านถึง"

🎯 สรุป Mental Model

2.2 Storage

2.3 KRaft (ZooKeeper ตายแล้ว)

Part 3: Producer

3.1 Code พื้นฐาน

3.2 Acks

3.3 Idempotent Producer

3.4 Batching + Linger

3.5 Compression

3.6 Partitioner

3.7 Transactional Producer (EOS — Exactly-Once)

3.8 Producer Settings Reference

Part 4: Consumer

4.1 Basic

4.2 Consumer Group

4.3 Rebalance

Rebalance Protocols (Classic, ผ่าน client-side coordinator)

🆕 KIP-848 — New Consumer Group Protocol (Kafka 4.0 Early Access)

4.4 Offset Management

Auto Commit (ไม่แนะนำ)

Manual Commit (แนะนำ)

Offset Reset

4.5 Poll Loop ที่ถูก

4.6 Spring Kafka Consumer

Part 5: Topic Configuration

5.1 สร้าง Topic

5.2 Partition Count — สำคัญที่สุดในการตัดสินใจ

5.3 Replication Factor

5.4 Retention

5.5 Important Topic Configs

Part 6: Exactly-Once Semantics (EOS)

6.1 ภายใน Kafka

6.2 Stream Process EOS

6.3 ⚠️ EOS ครอบคลุมแค่ Kafka

Part 7: Schema Registry

7.1 ปัญหา

7.2 Schema Registry

7.3 Avro Example

7.4 Protobuf as alternative

Part 8: Kafka Connect

Part 9: MirrorMaker 2 + Multi-DC

Part 10: Cluster Operations

10.1 Sizing

10.2 Rack Awareness

10.3 Reassign Partition

10.4 Throttling Reassign

10.5 Tiered Storage (KIP-405 — GA ใน Kafka 3.6, production-ready ใน 4.0)

10.6 🆕 Share Groups / Queues for Kafka (KIP-932 — Early Access ใน Kafka 4.0)

Part 11: Performance Tuning

11.1 Producer

11.2 Consumer

11.3 Broker

11.4 OS

11.5 JVM

Part 12: Monitoring

12.1 JMX Metrics

12.2 Consumer Lag

12.3 Tools

Part 13: Security

13.1 SSL/TLS

13.2 SASL Authentication

13.3 ACL

Part 14: Pitfalls + Anti-Patterns

14.1 Too Few Partitions

14.2 Too Many Partitions

14.3 No Compression

14.4 Auto-commit + Long process

14.5 No Replication / RF=1