Apache Kafka Applications — Real-World Use Cases and Patterns

// Producer — publish user purchase event to Kafka
Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Enable idempotence — exactly-once delivery guarantee
props.put("enable.idempotence", "true");
props.put("acks", "all");  // wait for all replicas to confirm

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

String event = """
  {
    "event": "purchase",
    "userId": "user-123",
    "productId": "prod-456",
    "amount": 49.99,
    "timestamp": "2026-03-15T10:30:00Z"
  }
""";

// userId as partition key — all events for same user go to same partition
// This guarantees ordering of events per user
producer.send(
    new ProducerRecord<>("purchases", "user-123", event),
    (metadata, exception) -> {
        if (exception != null) log.error("Failed to send event", exception);
        else log.info("Sent to partition {} offset {}", metadata.partition(), metadata.offset());
    }
);

StreamsBuilder builder = new StreamsBuilder();

KStream<String, Purchase> purchases = builder.stream("purchases");

// Count purchases per product in 5-minute tumbling windows
KTable<Windowed<String>, Long> productCounts = purchases
    .groupBy((key, value) -> value.getProductId())
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count();

// Alert when a product receives 100+ purchases in 5 minutes (trending)
productCounts
    .filter((windowedKey, count) -> count > 100)
    .toStream()
    .mapValues((windowedKey, count) ->
        new TrendingAlert(windowedKey.key(), count, windowedKey.window().startTime()))
    .to("hot-products-alerts");

// Real-time revenue aggregation
KTable<String, Double> revenueByCategory = purchases
    .groupBy((key, value) -> value.getCategory())
    .aggregate(
        () -> 0.0,
        (category, purchase, total) -> total + purchase.getAmount(),
        Materialized.as("revenue-by-category-store")  // queryable state store
    );

// Query directly: stateStore.get("electronics") → $1,245,678.90
// This runs as a continuous streaming computation — zero batch jobs

// Kafka message from Debezium when a row is updated in PostgreSQL
// Each database change generates one Kafka message
{
  "op": "u",  // u=update, c=create (insert), d=delete, r=read (snapshot)
  "source": {
    "table": "orders",
    "db": "ecommerce",
    "ts_ms": 1710500000000,
    "lsn": 28690624  // PostgreSQL WAL (Write-Ahead Log) position
  },
  "before": {
    "id": 1001,
    "status": "pending",
    "amount": 49.99
  },
  "after": {
    "id": 1001,
    "status": "shipped",   // ← what changed
    "amount": 49.99
  }
}

// Consumers use this to:
// - Keep Elasticsearch in sync with PostgreSQL in real-time
// - Invalidate Redis cache when data changes
// - Notify downstream services of state changes
// - Build audit logs of every data change

# Key Kafka configuration decisions for production:

# Topic configuration — example for a purchases topic
kafka-topics.sh --create   --topic purchases   --partitions 12            # 12 partitions = 12-way parallelism
  --replication-factor 3     # 3 replicas = survives 2 broker failures
  --config retention.ms=604800000   # 7 days retention
  --config cleanup.policy=delete

# Producer: ensure durability
acks=all                     # Wait for all replicas to confirm
enable.idempotence=true      # Exactly-once producer
max.in.flight.requests.per.connection=5

# Consumer: control offset commits
enable.auto.commit=false     # Commit manually after processing
auto.offset.reset=earliest   # Start from beginning for new consumer groups
max.poll.records=500         # Process in batches

# Monitoring: always track
# consumer_group_lag — messages behind = processing delay
# producer_record_error_total — failed publishes
# broker_under_replicated_partitions — replication health