Lesson 50: Processing Rider Requests — The Spatial Normalization Gate

The Problem Nobody Talks About

Every rider request starts with two floats: latitude and longitude. This is the rawest, most expensive form of spatial data you can put in a stream. If you let those raw floats travel through your Kafka topology without converting them to an H3 index first, you have already lost. You’ve deferred a structural decision — “where does this event belong in space?” — to the part of your system least equipped to answer it cheaply. That’s the matching engine, and it’s already under load.

This lesson is about closing that gap at the consumer boundary. The RiderRequestProcessor is your spatial normalization gate. Its only job: convert lat/lon to an H3 Resolution 9 cell index and re-key the output record so that every downstream operator in the matching topology receives spatially coherent, co-located data — without touching a repartition topic.

---

The Naive Approach (And Why It Craters at 10k Events/sec)

Here’s what a standard developer writes on their first pass:

// NAIVE: Don't do this.
consumer.poll(Duration.ofMillis(100)).forEach(record -> {
    RiderRequest req = deserialize(record.value());
    activeRiders.put(req.riderId(), req); // HashMap in heap
    forwardToMatcher(req);                // keyed by riderId UUID
});

Three compounding failures:

1. UUID-keyed partitioning destroys spatial locality. Kafka assigns partitions by hash(key) % numPartitions. A rider UUID hash has zero correlation with geographic position. The result: a rider in San Francisco and a driver also in San Francisco land on different Kafka partitions, processed by different StreamThread instances. When the matching processor tries to join them, it issues a through-repartition — a full write to an internal compacted topic, a broker round-trip, and a re-read. At 3,000 rider requests/sec, you’re generating 3,000 unnecessary network round-trips per second. Each one adds 15–80ms depending on your broker topology. Your p99 matching latency just became structurally bounded by broker network RTT.

2. Heap state is not fault-tolerant. A HashMap<String, RiderRequest> in heap memory vanishes on StreamThread restart. Kafka Streams’ TaskManager will reassign the task to another thread, which will have no state — you’ve lost every in-flight rider request in that partition’s window. RocksDB-backed state stores avoid this; they checkpoint to a Kafka changelog topic and restore on reassignment.

3. No H3 = no prefix scan capability. The matching engine’s core operation is: “find all drivers within K=1 rings of this rider’s H3 cell.” If your RocksDB keys aren’t structured as {h3Cell}#{driverId}, you can’t do an O(log N) prefix scan. You’re doing a full O(N) table scan on every match request. At 500 active drivers per partition, that’s 500 RocksDB point reads per rider event. At 3,000 rider events/sec, that’s 1.5M RocksDB reads/sec — on a single StreamThread. RocksDB block cache thrash begins. rocksdb-block-cache-miss-ratio spikes. Everything degrades together.

---

The Uber-Lite Architecture: The Spatial Normalization Gate

The architectural decision is simple, but its consequences propagate through the entire system:

Convert lat/lon → H3 cell at the consumer boundary. Re-key every output record by H3 cell. Never let a raw coordinate travel past the first processor node.

This gives us three properties:

• Spatial Affinity: Records for the same geographic cell hash to the same Kafka partition. Riders and drivers in the same H3 cell land on the same StreamThread. The TaskManager never needs to repartition for a geo-join.

• Structural RocksDB Keys: Downstream processors receive h3Cell as the Kafka record key. They can immediately construct composite RocksDB keys ({h3Cell}#{entityId}) and do prefix scans — O(log N) lookup, not O(N) scan.

• Single Conversion Cost: H3’s latLngToCell is a pure mathematical operation (~200ns). Paying it once at the boundary is correct; paying it in every downstream operator is waste.

H3 Resolution 9: Why This Resolution?

H3 Resolution 9 produces hexagonal cells with an average area of 0.1053 km² and an average edge length of 174 meters. At this resolution:

• A single cell comfortably contains 5–15 drivers during normal urban density.

• K-Ring radius 1 (the 6 immediate neighbors + center cell = 7 cells) covers ~0.74 km² — roughly a 430-meter radius. That’s the right pickup search radius for dense urban environments.

• K-Ring radius 2 covers 19 cells (~2.0 km²) — suitable for sparse suburban demand.

Resolution 8 cells (~0.74 km²) are too coarse: too many drivers per cell degrades prefix scan selectivity. Resolution 10 (~0.015 km²) is too fine: in low-density areas, you’ll get zero drivers per cell and need K-Ring 5+ to find anyone, expanding fan-out dramatically.

---

Implementation Deep Dive

GitHub Link :

https://github.com/sysdr/uber-lite-p/tree/main/lesson50/uber-lite-lesson50

The Processor API Node

We use the Processor API (PAPI), not the DSL. The DSL’s mapValues and selectKey hide two operations that matter here: when the re-keying happens relative to the serialization boundary, and whether an internal repartition topic is created. With PAPI, we control this explicitly.

public class RiderRequestProcessor
        implements Processor<String, RiderRequest, String, EnrichedRiderRequest> {

    private ProcessorContext<String, EnrichedRiderRequest> context;
    private H3Core h3;

    @Override
    public void init(ProcessorContext<String, EnrichedRiderRequest> context) {
        this.context = context;
        try {
            this.h3 = H3Core.newInstance();
        } catch (IOException e) {
            throw new StreamsException("H3Core init failed", e);
        }
        // No state store here — this node is a stateless transformation gate.
        // State lives in the downstream DriverIndexProcessor.
    }

    @Override
    public void process(Record<String, RiderRequest> record) {
        RiderRequest req = record.value();

        // THE GATE: one call, paid once, never again downstream.
        long h3Cell = h3.latLngToCell(req.latitude(), req.longitude(), 9);
        String h3Key = Long.toUnsignedString(h3Cell);

        var enriched = new EnrichedRiderRequest(
            req.riderId(),
            req.latitude(),
            req.longitude(),
            h3Cell,
            h3Key,
            req.requestedAt()
        );

        // Re-key: new Kafka record key IS the H3 cell string.
        context.forward(record.withKey(h3Key).withValue(enriched));
    }
}

Why Long.toUnsignedString? H3 cell indices are 64-bit unsigned integers. Java’s long is signed. Raw long serialization of negative H3 values sorts incorrectly against their unsigned interpretation. Unsigned decimal string representation keeps key ordering predictable and human-readable in Kafka topic inspectors.

The Custom StreamPartitioner

This is the piece most engineers miss. Forwarding with the H3 cell as the record key is necessary but not sufficient. You also need to ensure the sink topic partitions by that key, and that the number of partitions matches between enriched-rider-requests and driver-locations topics. If partition counts differ, hash-modulo routing diverges and co-location breaks.

public class H3CellPartitioner implements StreamPartitioner<String, EnrichedRiderRequest> {
    @Override
    public Optional<Set<Integer>> partitions(String topic, String key,
                                              EnrichedRiderRequest value, int numPartitions) {
        int partition = Math.abs(Murmur2.hash(key.getBytes(StandardCharsets.UTF_8)))
                        % numPartitions;
        return Optional.of(Set.of(partition));
    }
}

Constraint: enriched-rider-requests, driver-locations, and matched-rides must all be created with the same partition count (12 in this lesson). Enforced by Docker Compose topic creation in the project script.

Topology Wiring

Topology topology = new Topology();

topology.addSource("RiderSource",
    new StringDeserializer(),
    new JsonDeserializer<>(RiderRequest.class),
    "rider-requests");

topology.addProcessor("RiderNormalizer",
    RiderRequestProcessor::new,
    "RiderSource");

topology.addSink("EnrichedRiderSink",
    "enriched-rider-requests",
    new StringSerializer(),
    new JsonSerializer<>(),
    new H3CellPartitioner(),   // explicit spatial partitioner
    "RiderNormalizer");

No DSL. No hidden repartition topic. The TaskManager sees exactly 3 nodes: source → processor → sink. One StreamThread handles all three in a tight loop.

RecordAccumulator Tuning

The sink write path goes through Kafka’s RecordAccumulator. Default batch.size=16384 (16KB) and linger.ms=0 means every context.forward() attempts an immediate flush — fine for latency, but at 3,000 events/sec you’re making 3,000 syscalls/sec. Tune for throughput/latency tradeoff:

batch.size=32768        # 32KB batches — fits ~100 EnrichedRiderRequest records
linger.ms=5             # Wait 5ms to fill batch; adds at most 5ms to p50 latency
compression.type=lz4    # LZ4: lower CPU than Snappy, similar compression ratio

At 3,000 events/sec with linger.ms=5, the RecordAccumulator accumulates ~15 records per batch before flushing. That’s 15× fewer broker write RPCs for a 5ms p50 latency budget — a reasonable trade when downstream join latency dominates anyway.

---

Working Demo Link :

Production Metrics: What to Watch

Metric Source Alert Threshold What It Tells You process-latency-max Kafka Streams JMX > 50ms H3 conversion or serialization bottleneck records-consumed-rate Kafka Streams JMX < 2,800/sec Consumer lag building consumer-lag kafka-consumer-groups > 5,000 records StreamThread falling behind rocksdb-block-cache-miss-ratio Streams JMX (downstream) > 0.15 Cache too small for working set partition-skew Custom / Burrow > 10% H3 partitioner distributing unevenly batch-size-avg Producer JMX < 8KB linger.ms too low; batch not filling

Partition skew deserves special attention. H3 cells at Resolution 9 are not uniformly distributed — dense urban grids generate far more events than rural cells. Monitor per-partition consumer lag in Grafana. If one partition consistently runs 3× behind others, you have a hot H3 region. The mitigation is sub-cell sharding: append a modulo suffix to the H3 key (h3Key + "#" + (riderId.hashCode() % SHARD_FACTOR)) to spread intra-cell load.

Step-by-Step Execution Guide

Prerequisites

• Docker 24+ and Docker Compose v2

• Java 21 (Temurin or GraalVM)

• Maven 3.9+

Setup & Run

cd uber-lite-lesson50
./start.sh
docker compose up -d && sleep 15
mvn clean package -q
java --enable-preview -jar target/lesson50.jar

Verification

chmod +x verify.sh && ./verify.sh

Expected output:

[VERIFY] Sending 100 test rider requests...
[VERIFY] Consuming from enriched-rider-requests...
[VERIFY] Sample record key: 617700169958293503  (H3 cell index)
[VERIFY] h3Cell field matches key: TRUE
[VERIFY] All 100 records enriched with valid H3 Res 9 cells
[VERIFY] Partition distribution skew: < 5%
PASS

The Kafka record key IS the H3 cell index. Not a UUID. Not a hash. A geographic address. That’s the gate working correctly.

What’s Next

Lesson 51 builds directly on this output. The enriched-rider-requests topic, partitioned by H3 cell, becomes the trigger stream for the K-Ring expansion processor. Because we paid the H3 conversion cost here and established spatial partition affinity, Lesson 51’s DriverMatchProcessor immediately issues 7 RocksDB prefix scans (one per K-Ring=1 cell) with zero repartition overhead. The work done here — boring, precise, invisible — is what makes the matching engine possible at scale.