Streaming Systems Deep Dive

How Partitions Work

ISR (In-Sync Replicas)

Replicas that are caught up with the leader. A message is "committed" when all ISRs have it.

1. Idempotent Producer

Kafka assigns a unique ID to each producer. Retried messages with same sequence number are deduplicated.

# Enable idempotent producer
enable.idempotence=true

# Kafka tracks: (producer_id, sequence_number) per partition
# Duplicate detected if same sequence_number arrives twice

2. Transactional Producer

Atomic writes across multiple partitions/topics. All or nothing.

producer = KafkaProducer(
    transactional_id="order-processor-1",  # Unique ID
    enable_idempotence=True
)

producer.init_transactions()

try:
    producer.begin_transaction()

    # These writes are atomic
    producer.send("orders", value=order)
    producer.send("inventory", value=inventory_update)
    producer.send("notifications", value=notification)

    producer.commit_transaction()  # All visible at once
except Exception:
    producer.abort_transaction()   # All rolled back

3. Consume-Transform-Produce Pattern

# Exactly-once stream processing
producer.begin_transaction()

# Read from input
records = consumer.poll()

for record in records:
    result = transform(record)

    # Write to output
    producer.send("output-topic", value=result)

# Commit offset AND output in same transaction
producer.send_offsets_to_transaction(
    consumer.get_offsets(),
    consumer.group_id
)
producer.commit_transaction()

Windowing

Group events by time windows for aggregation.

// Tumbling window: Count orders per 5-minute window
orders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count();

// Sliding window: Count orders in last 5 minutes, updated every 1 minute
orders
    .groupByKey()
    .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)))
    .count();

// Session window: Group user activity with 30-minute inactivity gap
userEvents
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
    .count();

Event Time vs Processing Time

// Handle late events with watermarks
.windowedBy(TimeWindows
    .ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1)))
// Events up to 1 minute late will still be included in correct window

Stateful Processing

Store state as a sequence of events, not current state.

# Event sourcing with Kafka
# Topic: account-events (source of truth)
{"event": "AccountCreated", "account_id": "123", "timestamp": "..."}
{"event": "MoneyDeposited", "account_id": "123", "amount": 100}
{"event": "MoneyWithdrawn", "account_id": "123", "amount": 30}
{"event": "MoneyDeposited", "account_id": "123", "amount": 50}

# Consumer rebuilds current state by replaying events
balance = 0
for event in events:
    if event.type == "MoneyDeposited":
        balance += event.amount
    elif event.type == "MoneyWithdrawn":
        balance -= event.amount
# balance = 120

1. Real-Time Analytics Dashboard

User Events → Kafka → Flink (aggregate per minute) → Redis → Dashboard

# Flink SQL
SELECT
    window_start,
    page_url,
    COUNT(*) as page_views,
    COUNT(DISTINCT user_id) as unique_visitors
FROM TABLE(
    TUMBLE(TABLE page_views, DESCRIPTOR(event_time), INTERVAL '1' MINUTE)
)
GROUP BY window_start, page_url;

2. Fraud Detection

Transactions → Kafka → Flink CEP → Alerts

# Complex Event Processing: Detect 3+ transactions in 5 minutes from different countries
Pattern<Transaction> pattern = Pattern
    .<Transaction>begin("first")
    .where(tx -> true)
    .followedBy("second")
    .where(tx -> !tx.country.equals(prev.country))
    .followedBy("third")
    .where(tx -> !tx.country.equals(prev.country))
    .within(Time.minutes(5));

3. Log Aggregation

App Logs → Kafka → Logstash → Elasticsearch → Kibana

# High throughput: 100K+ logs/second
# Kafka handles backpressure, Elasticsearch can catch up

4. Event-Driven Microservices

# Order placed → multiple services react
order-events:
  - OrderService publishes OrderPlaced
  - InventoryService: reserve stock
  - PaymentService: charge card
  - NotificationService: send confirmation
  - AnalyticsService: update metrics
  - RecommendationService: update user profile

When to Use Streaming

Kafka Configuration Cheat Sheet