feat(berlinmod): full BerlinMOD-9 × 3-form parity matrix on Kafka Streams (27/27 cells)

[estebanzimanyi]

Draft. BerlinMOD-9 × 3-form parity-matrix work on MobilityKafka (Kafka-Streams runtime). 27 of 27 cells = 100 % of the MobilityKafka parity-matrix row, matching MobilityFlink #3.

Coverage

Q	Topic	Continuous	Windowed	Snapshot
Q1	"which vehicles have appeared?"	✓	✓	✓
Q2	"where is vehicle X at time T?"	✓	✓	✓
Q3	"vehicles within d of P at time T?"	✓	✓	✓
Q4	"vehicles entered region R?"	✓	✓	✓
Q5	"pairs of vehicles meeting near P"	✓	✓	✓
Q6	"cumulative distance per vehicle"	✓	✓	✓
Q7	"first passage through POIs"	✓	✓	✓
Q8	"vehicles close to road segment"	✓	✓	✓
Q9	"X-Y distance at time T"	✓	✓	✓

Module contents

kafka-streams-app/ — Maven project (Java 21, Kafka Streams 3.6.0).

Per-form processors (27): Q<N>ContinuousProcessor, Q<N>WindowedProcessor, Q<N>SnapshotProcessor for N ∈ {1..9}. Each Q-Windowed/Q-Snapshot processor schedules a PunctuationType.STREAM_TIME callback at the form-appropriate interval (10 000 ms for windowed, 5 000 ms for snapshot).

Shared infrastructure: BerlinMODTrip (data class), BerlinMODTripSerde (JSON Serde), Haversine (great-circle distance, m), SegmentDistance (planar projection point-to-segment, m), PointOfInterest (Q7 record), BerlinMODTopology (unified topology fanning to per-Q-form output topics), BerlinMODQ1LocalTest (TopologyTestDriver-based local driver).

Output counts (TopologyTestDriver, sorted-event-time corpus + dual sentinels)

Q	Continuous	Windowed	Snapshot
Q1	3	2	13
Q2	7	2	4
Q3	21	2	9
Q4	1	2	5
Q5	19	2	4
Q6	21	6	13
Q7	3	6	13
Q8	21	2	9
Q9	13	2	4

Two windows close in the test ([T0+0, T0+10000) and [T0+10000, T0+20000)); Q1/Q2/Q3/Q4/Q5/Q8/Q9 emit 1 line per closed window; Q6/Q7 emit 1 line per vehicle per closed window.

Snapshot semantics vs MobilityFlink #3

Kafka Streams' STREAM_TIME punctuator fires on stream-time advance with state-at-fire-moment, and coalesces multi-interval stream-time jumps into a single fire at the current stream-time. Flink's bounded-source watermark jumps to +∞ at source-close and flushes all keyed timers with final state.

The LocalTest pipes two sentinel records (at t=T0+15001 and t=T0+20001) to step the punctuator through the desired tick boundaries one at a time. The first event's initial-catchup fire also produces a tick at T=T0 — a Kafka Streams runtime feature with no Flink-bounded-source analogue.

State patterns covered

Pattern	Cells
Stateless filter / predicate	Q2-c, Q3-c, Q8-c
Single keyed flag	Q1-c, Q1-s, Q4-c
Single-key state	Q2-s, Q9-s
Per-vehicle accumulator	Q6-c, Q6-s
Per-(vehicle, POI) keyed state	Q7-c, Q7-s
Cross-vehicle shared via selectKey(0)	Q5-c, Q3-s, Q5-s, Q7-s, Q8-s
Paired shared via selectKey(0)	Q9-c, Q4-s, Q9-s
winStart-keyed CSV-vehicleId set	Q1-w, Q3-w, Q8-w
winStart-keyed encoded last-known per vehicle	Q2-w, Q5-w, Q9-w
winStart-keyed per-vehicle accumulator	Q6-w
winStart-keyed per-vehicle entries log	Q4-w, Q7-w

Bump-isolation

Zero MEOS surface anywhere:

• No JMEOS jar dependency
• No libmeos.so linkage
• No PyMEOS-side code
• No file collision with the in-flight MEOS 1.4 bump's scope

Spatial predicates use pure-Java Haversine / planar SegmentDistance / point-in-box, marked TODO(meos) at each call site for JMEOS-bridge migration after the bump settles.

Build verification

mvn -q clean package -DskipTests
... BUILD SUCCESS

java --add-opens java.base/java.lang=ALL-UNNAMED \
     --add-opens java.base/java.util=ALL-UNNAMED \
     --add-opens java.base/java.lang.reflect=ALL-UNNAMED \
     -cp target/mobility-kafka-streams-1.0-SNAPSHOT.jar \
     berlinmod.BerlinMODQ1LocalTest

Single squashed commit, no AI attribution, target/ ignored by .gitignore.

Companion artefacts

• MobilityFlink — full 27-cell row: MobilityDB/MobilityFlink#3
• Ecosystem profile parity-contract: MobilityDB/.github#10