Kafka Operations & Tuning

Architecture:

  Producer                Schema Registry              Consumer
  ┌──────┐              ┌──────────────┐              ┌──────┐
  │      │──register──▶ │  Stores      │ ◀──lookup──  │      │
  │      │  schema      │  schemas by  │   schema     │      │
  │      │              │  subject +   │   by ID      │      │
  │      │              │  version     │              │      │
  └──────┘              └──────────────┘              └──────┘

Flow:
  1. Producer registers schema with Schema Registry (gets schema ID)
  2. Producer embeds schema ID in each Kafka record (first 5 bytes)
  3. Consumer reads schema ID from record
  4. Consumer fetches schema from Registry (cached locally)
  5. Consumer deserializes record using the fetched schema

Record format with Schema Registry:
  [0]         Magic byte (0x00)
  [1-4]       Schema ID (4 bytes, big-endian int)
  [5...]      Serialized payload (Avro/Protobuf/JSON)

Subjects (how schemas are organized):
  - TopicNameStrategy (default): subject = "<topic>-value" or "<topic>-key"
  - RecordNameStrategy: subject = fully qualified record name
  - TopicRecordNameStrategy: subject = "<topic>-<record-name>"

Safe changes (BACKWARD compatible):
  ✓ Add a new field WITH a default value
  ✓ Remove a field that has a default value
  ✓ Rename a field using an alias (Avro)

Unsafe changes (BREAKING):
  ✗ Remove a required field (no default)
  ✗ Change a field's type (int → string)
  ✗ Rename a field without alias
  ✗ Add a required field without default

Example — Avro schema evolution:

Version 1:
  {"type": "record", "name": "User", "fields": [
    {"name": "id", "type": "string"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": "string"}
  ]}

Version 2 (BACKWARD compatible — added optional field):
  {"type": "record", "name": "User", "fields": [
    {"name": "id", "type": "string"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": "string"},
    {"name": "phone", "type": ["null", "string"], "default": null}  ← optional
  ]}

New consumers can read old records (phone defaults to null).
Old consumers can still read old records (they ignore phone).

Scaling consumers for throughput:

  1. Start: 1 consumer, measure processing rate per partition
  2. If consumer can't keep up (lag increasing):
     a. First: optimize processing (batch DB writes, async I/O)
     b. Then: add consumers up to partition count
     c. If still not enough: increase partitions, then add consumers

  Max parallelism = min(partition count, consumer count in group)

  Example:
    Topic: 12 partitions
    1 consumer  → handles 12 partitions (max throughput: 1x processing speed)
    3 consumers → each handles 4 partitions (3x throughput)
    6 consumers → each handles 2 partitions (6x throughput)
    12 consumers → each handles 1 partition (12x throughput — maximum)
    15 consumers → 12 active + 3 idle (wasted resources)

# Threading
num.network.threads = 8          # Threads handling network requests
num.io.threads = 16              # Threads handling disk I/O
num.replica.fetchers = 4         # Threads for follower replication

# Log management
log.segment.bytes = 1073741824   # 1 GB segment files
log.retention.hours = 168        # 7 days retention
log.retention.bytes = -1         # No size limit (use time-based)
log.cleanup.policy = delete      # Or "compact" for changelog topics

# Replication
default.replication.factor = 3
min.insync.replicas = 2
unclean.leader.election.enable = false

# Allow user "order-service" to produce to "orders" topic
kafka-acls.sh --add \
  --allow-principal User:order-service \
  --operation Write \
  --topic orders

# Allow user "analytics" to consume from "orders" topic
kafka-acls.sh --add \
  --allow-principal User:analytics \
  --operation Read \
  --topic orders \
  --group analytics-group

# Deny all other access (default deny)
# Best practice: start with deny-all, explicitly allow what's needed

# Multi-tenant isolation:
#   - Separate topics per tenant (tenant-a.orders, tenant-b.orders)
#   - ACLs restrict each tenant to their own topics
#   - Quotas limit throughput per client to prevent noisy neighbors