Design a Top-K Trending / Real-Time Leaderboard System

• Top-K leaderboard : ingests millions of engagement events and publishes the K highest-scoring items over a rolling time window.
• Single-owner partitioning : Kafka partitioning by item key so each item has exactly one counting authority.
• Windowed aggregation : hopping windows with event-time watermarks for correct time accounting.
• Atomic snapshot publication : writing a complete versioned result set and flipping a pointer, preventing stale overwrites.
• Space-Saving / Count-Min Sketch : approximate counting structures for memory-constrained environments.

How do you define popularity?

Popularity could be a raw count, a weighted score (share counts more than view), or velocity (how fast engagement is rising). Define it precisely because the whole system depends on it. At National Bank of India, popularity of app features is a weighted score: a completed transaction counts 3 points, a page view counts 1 point, and a share counts 5 points.

package com.bank.leaderboard;

public class PopularityScorer {
    public long score(EngagementEvent event) {
        return switch (event.type()) {
            case VIEW -> 1;
            case TRANSACTION -> 3;
            case SHARE -> 5;
            default -> 0;
        };
    }
}

What is the single-process baseline?

A HashMap from item to count plus a min-heap of size K. Each event updates the map, and if the count beats the heap's smallest, the heap is updated. This is O(N log K) and works for moderate event rates on a single machine.

package com.bank.leaderboard;

import java.util.*;
import java.util.stream.Stream;

public class SingleProcessTopK {
    public List<String> topK(Stream<String> events, int k) {
        Map<String, Long> counts = new HashMap<>();
        events.forEach(e -> counts.merge(e, 1L, Long::sum));
        PriorityQueue<Map.Entry<String, Long>> heap =
            new PriorityQueue<>(Map.Entry.comparingByValue());
        for (var e : counts.entrySet()) {
            heap.offer(e);
            if (heap.size() > k) heap.poll();
        }
        return heap.stream()
            .sorted(Map.Entry.<String, Long>comparingByValue().reversed())
            .map(Map.Entry::getKey).toList();
    }
}

Why partition by item key?

Each item must have exactly one counting owner. If counts are split across workers, local top-K lists can miss a globally popular item that appears below each worker's threshold. Kafka partitioning by item key guarantees all events for one item go to the same consumer, making the local count authoritative.

How do rolling windows work?

A 5-minute window refreshed every few seconds is a hopping window. Use event-time with watermarks and bounded lateness so late events land in the correct window. Events arriving after the watermark has advanced past their window are dropped or counted in a late-arrivals side channel.

package com.bank.leaderboard;

import java.time.Instant;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class WindowedCounter {
    private final long windowMs = 5 * 60 * 1000;  // 5 minutes
    private final Map<String, Map<String, Long>> windowCounts = new ConcurrentHashMap<>();

    public void count(String windowKey, String itemId, long score) {
        windowCounts.computeIfAbsent(windowKey, k -> new ConcurrentHashMap<>())
            .merge(itemId, score, Long::sum);
    }
    public String windowKey(Instant eventTime) {
        long epoch = eventTime.toEpochMilli();
        return String.valueOf(epoch - (epoch % windowMs));
    }
}

How do you handle duplicates?

Deduplicate by stable event id in checkpointed state so a replay (e.g., Kafka consumer restart) never double-counts. The deduplication state is checkpointed alongside the window counts for crash recovery.

package com.bank.leaderboard;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class Deduplicator {
    private final Set<String> seenIds = ConcurrentHashMap.newKeySet();

    public boolean isNew(String eventId) {
        return seenIds.add(eventId);  // returns false if already seen
    }
}

How do you handle hot keys?

A viral feature (e.g., a new "instant loan" button) concentrates millions of events on one partition. Use partial pre-aggregation: split the hot key into N sub-keys (e.g., "instant-loan:0" through "instant-loan:9"), aggregate each sub-key on a separate worker, and sum them in the global merge step.

package com.bank.leaderboard;

import java.util.concurrent.ThreadLocalRandom;

public class HotKeySplitter {
    private static final int SPLIT_FACTOR = 10;

    public String splitKey(String itemId) {
        if (hotKeyRegistry.isHot(itemId)) {
            int shard = ThreadLocalRandom.current().nextInt(SPLIT_FACTOR);
            return itemId + ":" + shard;
        }
        return itemId;
    }
}

How do you publish snapshots?

Write a complete versioned snapshot and flip a pointer atomically. Reject older jobs overwriting newer snapshots by comparing version numbers. The snapshot includes the top-K list, the window timestamp, and the watermark position.

How does the serving layer work?

The serving layer (Redis or a cache) exposes the precomputed snapshot with freshness metadata (generation time, watermark) so a stale ranking is detectable. Filter out ineligible items (e.g., features disabled by ops), break ties deterministically by item id so the ranking is stable across refreshes.

What about approximate counting?

Count-Min Sketch estimates frequencies cheaply but cannot enumerate top-K ids by itself because it does not store keys. It needs a candidate stage (e.g., Space-Saving) to track heavy hitters. Space-Saving directly tracks the top items and is more practical for leaderboards.

What should a senior engineer know?

A senior engineer designs for recovery using checkpointed stream state, Kafka replay, and idempotent sinks. They know that the single-owner property is the correctness crux and design around it. They also monitor end-to-end latency from event emission to snapshot publication and alert when freshness degrades beyond an acceptable threshold.