পাঠ ১৮ · ২৯-এর মধ্যে · মডিউল ৩

Home / AI Courses / ডেটা ইঞ্জিনিয়ারিং / Producer, Consumer, Topic

Producer, Consumer, Topic

Kafka core concepts in depth

৭ মিনিট পড়া মাঝারি · Intermediate Hands-on Python

এই পাঠে যা শিখবেন

Producer config — acks, linger.ms, batch.size, idempotence-এর প্রভাব
Consumer group dynamics — assignment, rebalance, sticky strategies
Partition key strategy — bKash, Pathao-এর ordering guarantee
Offset management — auto vs manual commit, at-least-once vs exactly-once

১ · Producer — কীভাবে message Kafka-তে পাঠায়

ProducerProducerএকটি Kafka client library যা topic-এ messages publish করে। Internally batching, retry, partitioning handle করে — application code শুধু send() করে। একটি client library — application থেকে Kafka broker-এ events পাঠানোর responsibility। JVM-এ official client; Python-এ kafka-python, confluent-kafka-python।

Internal flow:

App producer.send(topic, key, value) call করে।
Producer key থেকে partition নির্ধারণ করে: partition = hash(key) % num_partitions।
Message একটি in-memory batch-এ যায় (নির্দিষ্ট partition-এর জন্য)।
Batch full হলে বা linger.ms অতিক্রান্ত হলে — broker-এ পাঠায়।
Broker write করে — ack ফেরত পাঠায়।
Producer next batch।

২ · Acks — durability vs latency

acks config — তিনটি মান, তিনটি trade-off:

acks=0 (fire-and-forget): Broker-এর ack চাইব না। Lowest latency, কিন্তু broker fail-এ message hারিয়ে যেতে পারে। Logging-এ ঠিক, transaction-এ NEVER।
acks=1 (leader-only): Leader broker write করলে ack। Follower replicate-এর আগে leader fail করলে data loss। Default পুরনো versions-এ; কিন্তু production-এ কম safe।
acks=all বা acks=-1: ISR-এর সব broker write করলে ack। Highest durability। সাথে min.insync.replicas=2 দিলে — ১ broker fail-এও safe।

bKash-এর Producer Config

Financial transaction-এর জন্য সঠিক config:
acks=all, enable.idempotence=true, retries=Integer.MAX_VALUE, max.in.flight.requests.per.connection=5।
Throughput কমে কিন্তু $0$ data loss + $0$ duplicate (single partition-এ)।

৩ · Batching — throughput-এর secret

Producer প্রতি message immediate পাঠালে — network round-trip-এ সময় নষ্ট, broker overwhelmed। তাই messages batch-এ পাঠায়। দু'টি knob:

batch.size (default ১৬ KB): একটি batch-এর max size। Larger = better compression।
linger.ms (default ০): batch full না হলে কতক্ষণ অপেক্ষা। ০ = immediate; ১০-১০০ = throughput অনেক বেশি।

Compression (compression.type=lz4 বা zstd) batch-এ apply হয় — ৪-১০× space saving। Network bandwidth ও storage দুটোতেই সাশ্রয়।

৪ · Idempotent producer — duplicate রোধ

Producer message পাঠাল, broker write করল কিন্তু ack network-এ harয়ে গেল। Producer retry — same message আবার লেখা। Result: duplicate।

সমাধান — enable.idempotence=true:

Producer-এ unique producer_id assign।
প্রতি message-এ monotonically increasing sequence_number।
Broker (producer_id, partition)-এর per-partition last sequence number store করে।
Duplicate sequence এলে — drop (silently)।

Cost minimal (small per-batch overhead), benefit বিশাল। ২০২২+ Kafka-তে এটি default; manually disable না করলে on।

৫ · Consumer ও consumer group

ConsumerConsumerKafka topic থেকে messages pull করে। Push model নয় — consumer নিজের গতিতে পড়ে। Slow consumer Kafka কে slow করে না; just lag বাড়ে। topic থেকে message pull করে। কিন্তু "pull" architecture — Kafka নিজে push করে না; consumer রিকোয়েস্ট পাঠায়। ফলে slow consumer Kafka-কে slow করে না।

Consumer group: একই group.id-এর সব consumer একটি group। Group-এর মধ্যে — Kafka partitions বণ্টন করে। প্রতি partition exactly একজন consumer পড়ে (group-এর মধ্যে)। ফলে parallelism partition count পর্যন্ত।

একাধিক group: different group.id-গুলো independent। প্রতিটি group একই messages পৃথকভাবে পড়ে — নিজের offset।

৬ · Rebalance — যখন group পরিবর্তিত হয়

Consumer যোগ/বিয়োগ হলে — Kafka partitions পুনঃ-বণ্টন করে। এটাই rebalanceRebalanceConsumer group-এর সদস্য পরিবর্তিত হলে — partitions নতুন করে assign। সাধারণত consumer ১-৫ সেকেন্ড "stop the world" — production-এ মূল pain point।। তিন strategy:

Range assignor: Topic-by-topic partition range বণ্টন। Skew হতে পারে।
Round-robin: সব topic mix করে round-robin বণ্টন। Better balance।
Sticky/Cooperative-sticky (recommended): পুরনো assignment যতটা সম্ভব রক্ষা — শুধু minimum partition move। ২০২০+ default।

Rebalance storm — common production problem। Consumer GC pause-এ heartbeat miss → group থেকে kick → rebalance → আরেক consumer slow → আরেক rebalance। সমাধান: session.timeout.ms বাড়ান, GC tune করুন, cooperative-sticky use করুন।

Producer key-based partition assignment, topic-এ ৪ partition, ২ consumer group স্বাধীনভাবে পড়ছে নিজের গতিতে।

৭ · Partition key — ordering guarantee

Kafka partition-এর মধ্যে message order রক্ষিত — কিন্তু partition-এর মধ্যে নয়। তাই key বাছাই critical:

bKash: key = user_phone। একই user-এর deposit, withdraw, send money — সব ordered।
Pathao: key = ride_id। একই ride-এর accept, started, completed events — sequential।
Daraz: key = order_id। একটি order-এর state machine একই partition-এ।

Key skew problem: একটি key extremely active হলে — সেই partition hot। যেমন bKash-এ একটি বড় agent-এর partition load অনেক বেশি। Solution: composite key (user + transaction_type) বা random suffix (যদি ordering শুধু group-level prefix-এ চাই)।

৮ · Offset commit — at-least-once vs exactly-once

Consumer offset commit কখন করবে — এটি delivery semantic-এর core:

Auto commit (default): প্রতি ৫ সেকেন্ডে background commit। সহজ, কিন্তু crash-এ duplicate বা loss।
Manual commit (recommended): Process করার পর consumer.commitSync()। At-least-once guaranteed। Output sink-এ idempotency দরকার।
Transactional (exactly-once): Kafka Streams বা Flink — read offsets + state + write — সব single transaction।

৯ · Hands-on — Python producer ও consumer

Realistic bKash transaction stream — proper config-এর সাথে:

Python · confluent-kafka producer

from confluent_kafka import Producer
import json, random, time

# Production-ready config
conf = {
    'bootstrap.servers': 'localhost:9092',
    'acks': 'all',
    'enable.idempotence': True,
    'compression.type': 'lz4',
    'linger.ms': 10,
    'batch.size': 32768,
    'retries': 2147483647,
    'max.in.flight.requests.per.connection': 5,
}

producer = Producer(conf)

def delivery_report(err, msg):
    if err:
        print(f"FAILED: {err}")
    else:
        print(f"OK: partition={msg.partition()} offset={msg.offset()}")

# Simulate bKash transactions
phones = ['01711000001', '01612000002', '01911000003']
for i in range(20):
    phone = random.choice(phones)
    tx = {
        'tx_id': f'TX{i:06d}',
        'user_phone': phone,
        'amount': random.choice([100, 500, 1000, 5000]),
        'type': random.choice(['send_money', 'cash_in', 'payment']),
        'event_time': int(time.time() * 1000),
    }
    # key = user_phone → একই user-এর tx একই partition-এ
    producer.produce(
        topic='bkash-transactions',
        key=phone,
        value=json.dumps(tx),
        callback=delivery_report,
    )
    producer.poll(0)  # trigger callbacks

producer.flush()  # সব pending messages broker-এ পাঠাও

callback — async confirmation, প্রতিটি message broker-এ পৌঁছালে print। flush() — সব in-flight শেষ পর্যন্ত wait। লক্ষ্য করুন একই phone-এর tx গুলো একই partition-এ যাচ্ছে।

Manual offset commit সহকারে fraud-detector consumer:

Python · confluent-kafka consumer

from confluent_kafka import Consumer, KafkaError
import json

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'fraud-detector',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,        # manual commit-এ control
    'partition.assignment.strategy': 'cooperative-sticky',
    'session.timeout.ms': 45000,
}

consumer = Consumer(conf)
consumer.subscribe(['bkash-transactions'])

def is_fraud(tx):
    # সরল rule — production-এ ML model
    return tx['amount'] > 50000 and tx['type'] == 'send_money'

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            print(f"ERROR: {msg.error()}")
            continue

        tx = json.loads(msg.value())
        decision = 'BLOCK' if is_fraud(tx) else 'ALLOW'
        print(f"[{msg.partition()}/{msg.offset()}] "
              f"{tx['user_phone']} ৳{tx['amount']} → {decision}")

        # Process সফল — তবে commit
        consumer.commit(message=msg, asynchronous=False)

except KeyboardInterrupt:
    pass
finally:
    consumer.close()

enable.auto.commit=False + commit(message) — at-least-once guarantee। Process-এর পর crash হলে সর্বশেষ committed offset থেকে restart, সম্ভাব্য duplicate। Idempotency downstream-এ।

Production tip: commit(message=msg)-এ প্রতিটি message commit slow। বরং প্রতি ১০০ message বা প্রতি ৫ সেকেন্ডে batch commit। Trade-off: কতটুকু duplicate acceptable।

ভাবনার প্রশ্ন

প্রতিটি প্রশ্ন নিজে কিছুক্ষণ ভাবুন — তারপর "→ উত্তর" চাপুন।

প্র ০১ bKash-এ একটি transaction-এর state machine: initiated → debited → credited → completed। এই sequence-এ ordering critical। আপনি partition key কী বাছবেন? Key skew হলে কী করবেন?

এটি real production challenge — Bangladesh-এর top fintech-এ এই সমস্যা সমাধান করা হয়।

Key choice options:

(১) transaction_id:

Pros: uniform distribution (UUID/snowflake), no skew।
Cons: একই user-এর different transactions ভিন্ন partition-এ — user-level analysis কঠিন।
Best for: per-transaction state machine।

(২) user_phone:

Pros: একই user-এর সব transactions ordered। Per-user fraud check natural।
Cons: Skew — VIP users (large agents) hot partition।
Best for: user-level analytics, fraud detection।

(৩) Composite key user_phone + transaction_id:

Pros: Distribution ভাল, কিন্তু same user-এর tx একসাথে নয়।
Cons: ordering guarantee হারায়।

Recommended approach:

Multi-topic strategy:

transaction-state-events: key = transaction_id। Per-tx state machine ordered।
user-activity-events: key = user_phone। User-level analysis।
Same source publishes-এ দু'টি topic — different consumers different needs।

Skew detection:

Kafka JMX metrics: BytesInPerSec per partition — variance দেখুন।
Top-1% partitions ১০× average হলে — skew problem।
Hotspot partition consumer lag accumulate করবে।

Skew mitigation:

(ক) Sub-partitioning:

VIP user identify (top-100 by volume)।
Their key = {user_phone}#{random(0..9)} — ১০ sub-partition।
Trade-off: per-user ordering নষ্ট, কিন্তু throughput scaled।
Application-level sequence number-এ ordering reconstruct।

(খ) Increased partitions:

Topic-এ ২৪ → ৪৮ partition — distribution improve।
Caveat: existing key-এর hash mod বদলে যায় — re-partition।

(গ) Custom partitioner:

Standard hash-এর বদলে — known hot keys-কে dedicated partition assign।
Application-level smart routing।
Operational complexity বাড়ে।

(ঘ) Compaction:

Per-key latest state রাখা — log compacted topic।
"Current state" lookup-এর জন্য — full history অন্য topic-এ।

bKash-specific reality:

Top ১০০ agent transactions-এর ৩০-৪০% volume — major skew।
Production solution: Hybrid — most users user_phone, top agents {phone}#{shard}।
Fraud detection সঠিক agent-key pattern চিনতে পারে — application logic।

মূল উপলব্ধি: Key choice-এর কোনো universal "right answer" নেই। Domain-specific trade-off — ordering, distribution, analytics-এর balance। অনেক সময় multiple topics ভিন্ন needs serve করে — pure design goal না।

প্র ০২ Auto commit বনাম manual commit — at-least-once vs exactly-once-এর সাথে কী সম্পর্ক? Production-এ কখন কোনটি বাছবেন?

এটি Kafka consumer-এর সবচেয়ে confusing concept। Wrong choice = silent data corruption।

Auto commit-এর mechanism:

enable.auto.commit=true, auto.commit.interval.ms=5000 (default)।
Consumer poll-এর সময় background-এ — সর্বশেষ offset commit।
Application code কিছু না করেই — automatic।

Auto commit-এর dangers:

Scenario 1 — Data loss:

Consumer ১০টি message poll করল।
৫টি process হওয়ার আগে — auto commit ঘটে গেল (সব ১০-এর offset)।
Crash — restart-এ সব ১০ already committed। Last 5 lost।

Scenario 2 — Duplicate (less likely with auto):

Auto commit interval-এর মধ্যে crash — সর্বশেষ commit থেকে restart।
Already-processed messages আবার process।

Manual commit options:

(১) commitSync() after every message:

Maximum safety, minimum throughput।
প্রতি commit network round-trip।
৫০-১০০ msg/sec realistic।

(২) commitSync() per batch:

poll() থেকে ফিরে আসা সব messages process → commit।
Kafka recommended pattern।
৫,০০০-৫০,০০০ msg/sec।

(৩) commitAsync():

Non-blocking — continue processing।
Failure-এ retry tricky (ordering issue)।
Use carefully।

Exactly-once specific:

enable.auto.commit=false — must।
Read offset + process + write — same transaction।
Kafka Streams: processing.guarantee=exactly_once_v2।
External sink: idempotent write (upsert by event_id)।

Production decision matrix:

Logging/metrics consumer: auto commit fine — occasional loss acceptable।
Analytics aggregation: manual commit, idempotent sink।
Financial transactions: manual commit + database transaction + unique constraint।
Stream processing (Flink/Kafka Streams): framework handles — exactly-once enable।

Common mistakes:

Auto commit + critical consumer = silent data loss in production।
Manual commit + non-idempotent DB insert = duplicate রেকর্ড।
commitAsync after each message — overhead similar to sync, complexity বেশি।
Forgetting to commit on shutdown — restart-এ duplicates।

Bangladesh fintech pattern:

bKash: at-least-once + database upsert। Transaction ID unique constraint।
Pathao: at-least-once + cache deduplication।
Daraz: order_id PK in DB — duplicate insert silently ignored।

মূল উপলব্ধি: "Exactly-once" কখনো consumer alone-এ অর্জনযোগ্য না। Always: at-least-once + idempotent sink = effectively-once। Auto commit production-এ rare — manual + business-aware commit pattern সঠিক।

প্র ০৩ একটি consumer group-এ ১০ consumer, কিন্তু topic-এ মাত্র ৪ partition। কী হবে? Throughput কীভাবে বাড়াবেন?

Classic Kafka scaling question — interview-এ frequently আসে।

সরাসরি উত্তর:

৪ partition → max ৪ active consumer।
৬ consumer idle — কিছু পড়বে না।
Hardware spec, network, ভালো code — কিছুই matter করবে না।

কেন এই limitation:

একটি partition-এ ordering guarantee — same key সব sequential।
২ consumer একই partition পড়লে — ordering ভেঙে যায়।
তাই one-partition-one-consumer hard rule।

Throughput বাড়ানোর strategies:

(১) Partition বাড়ান:

kafka-topics --alter --partitions 12 — ৪ → ১২।
এখন ১২ consumer active।
Caveat: existing keys-এর hash mod বদলে — same key new partition-এ যেতে পারে। Already-buffered messages এর ordering broken।
Best practice: Topic create-এর সময়ই overprovision (৩x expected concurrency)। Adding partitions later painful।

(২) Per-consumer throughput optimize:

fetch.min.bytes=1MB — broker batch বেশি data return।
max.poll.records=1000 — per poll() batch।
Async processing — poll-এর পরে records থ্রেড pool-এ।
Light-weight consumer logic — heavy work async background।

(৩) Vertical scaling:

প্রতি consumer larger machine (CPU, memory)।
Per-partition throughput কয়েক MB/s — bottleneck downstream হলে fix।

(৪) Decoupled processing:

Consumer শুধু read + queue (in-memory বা Redis)।
Worker pool queue থেকে process — concurrency consumer count-এর independent।
Caveat: ordering guarantee হারায় — careful design।

(৫) Batch processing within consumer:

প্রতি poll-এ ১০০-১০০০ records — bulk DB insert।
Per-message overhead কম।
Most production-এ ১০-৫০× throughput improvement।

(৬) Compression:

Network bandwidth bottleneck হলে — producer-এ compression.type=zstd।
৪-৭× compression — wire-এ কম data।

(৭) Multiple consumer groups:

একই topic, different responsibilities — different groups।
Parallel work, independent scaling।

Production tips:

Partition count = max(producers throughput need, consumers parallelism need)।
Underscore: ১০× headroom — future growth।
Too many partitions (১,০০০+ per broker) — controller overhead, longer rebalance।
Sweet spot: ১০-৫০ partitions per topic in most cases।

মূল উপলব্ধি: Partition count = parallelism ceiling। Topic design-এ এটি plan করুন। Throughput problem-এ — code optimize আগে, partition যোগ পরে। ১০ consumer + ৪ partition = ৬ paid engineers idle।

প্র ০৪ Pathao-এর driver location stream — প্রতি ১ সেকেন্ডে ১০,০০০ driver-এর GPS update। কীভাবে topic, key, producer config design করবেন?

High-frequency, high-cardinality streaming — modern fintech-এর challenge।

Volume analysis:

১০,০০০ driver × ১/সেকেন্ড = ১০,০০০ msg/sec।
Peak (rush hour): ২০,০০০-৩০,০০০ msg/sec।
Message size: ~১৫০ bytes (driver_id, lat, lng, timestamp, status)।
Throughput: ~৫ MB/sec — Kafka-এ trivial।

Topic design:

Topic name: pathao.driver-locations.v1 — domain.entity.version pattern।
Partitions: ১২। ১০ active driver = ৮৩৩ msg/sec/partition — comfortable।
Replication factor: ৩ — durability over performance।
Retention: ১ hour। GPS history-এর long-term storage S3-এ।
Compression: zstd — repeating fields highly compressible।

Partition key:

Choice: driver_id।
Per-driver location updates ordered (movement trajectory)।
Distribution uniform — driver_id sequential বা UUID।
No skew expected — সব driver সমান rate।

Producer config:

acks=1 — leader-only। GPS update lost OK (১ sec next update)। Latency lower।
compression.type=zstd — best ratio।
linger.ms=50 — ৫০ ms batching। Latency tolerance vs throughput।
batch.size=64KB — relatively large batches।
enable.idempotence=false — duplicate GPS irrelevant। Slight performance gain।

Schema (Avro):

{
  "type": "record",
  "name": "DriverLocation",
  "fields": [
    {"name": "driver_id", "type": "string"},
    {"name": "lat", "type": "double"},
    {"name": "lng", "type": "double"},
    {"name": "speed_kmh", "type": "float"},
    {"name": "heading", "type": "int"},
    {"name": "status", "type": {"type": "enum", "symbols": ["FREE", "BUSY", "OFFLINE"]}},
    {"name": "event_time", "type": "long", "logicalType": "timestamp-millis"}
  ]
}

Consumer groups:

ride-matcher: Free driver-দের location → Redis geospatial index। Real-time matching।
map-tile-renderer: Live driver heatmap dashboard।
compliance-tracker: Speed violation, route deviation alerts।
analytics-aggregator: ১-min window aggregation → ClickHouse।
s3-archiver: ৫-min batch → Parquet → S3।

Optimizations:

Idle driver throttle: client-side — driver stationary হলে updates ১/মিনিট। ৫০-৭০% volume reduction।
Tiered storage: Confluent Tiered Storage — hot data Kafka-তে, cold S3-তে। Retention অসীম।
Geo-sharded topics: Dhaka, Chattogram, Sylhet আলাদা topic — regional consumer locality।
Edge filter: Mobile SDK-তে সাধারণ filter (speed=0, last update < 5s skip) — server load কম।

Monitoring critical:

Producer error rate per minute।
Consumer lag — ride matcher-এ ১ sec-এর বেশি = problem।
Partition skew — driver_id distribution check।
Network bandwidth utilization।

Cost (Bangladesh, AWS Mumbai):

3-broker MSK m5.2xlarge ~৬০K BDT/month।
Storage (1hr retention, 5MB/s × 3) ~৫০ GB — negligible।
Cross-AZ data transfer largest cost — ১০-১৫K BDT।
Total ~৭৫-৯০K BDT/month।

মূল উপলব্ধি: Real-world streaming design = volume math + business priority + cost trade-off। GPS stream-এ idempotence-এর খরচ-এর মূল্য নেই (data lossy by nature), কিন্তু financial transaction-এ অপরিহার্য — same Kafka, ভিন্ন config।

অনুশীলন

Config বাছাই: নিচের use case-এ কোন producer config:
- (ক) bKash transaction
- (খ) IoT sensor (১,০০,০০০ device)
- (গ) Application access logs
- (ক) bKash: acks=all, enable.idempotence=true, compression=lz4, linger.ms=5। Durability সর্বোচ্চ।
- (খ) IoT: acks=1, enable.idempotence=false, compression=zstd, linger.ms=100, batch.size=128KB। High throughput, occasional loss OK।
- (গ) Logs: acks=0, compression=zstd, linger.ms=200। Lowest cost, lossy fine।
Rebalance scenario: ৪ consumer একটি group-এ, ১২ partition। ১ consumer crash করল। কী ঘটবে?
- প্রথমে: প্রতি consumer ৩ partition।
- Crash detect (heartbeat miss after session.timeout.ms, default 45s)।
- Coordinator rebalance trigger।
- Cooperative-sticky হলে: ৩ remaining consumer-দের ১টা করে partition যোগ → প্রতি ৪।
- Range/round-robin হলে: full reassignment, ১-৩ sec "stop the world"।
- Crashed consumer-এর partition-এর consumer lag বাড়বে — recovery-এ catch up।

Code pattern: একটি Python consumer লিখুন যা Daraz orders topic থেকে read করে, prices > 10,000 BDT-কে high-value flag করে, এবং proper offset commit করে।

from confluent_kafka import Consumer
import json

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'high-value-flagger',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,
})
consumer.subscribe(['orders'])

batch = []
try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None or msg.error():
            continue
        order = json.loads(msg.value())
        if order['total_bdt'] > 10000:
            print(f"HIGH-VALUE: order {order['order_id']} ৳{order['total_bdt']}")
            # downstream-এ idempotent insert (order_id PK)
        batch.append(msg)

        # Batch commit per 100 messages
        if len(batch) >= 100:
            consumer.commit(asynchronous=False)
            batch.clear()
finally:
    if batch:
        consumer.commit(asynchronous=False)
    consumer.close()