From 87de66c592f2a640cc3c2f22a3327d38b402fee2 Mon Sep 17 00:00:00 2001 From: Esteban Zimanyi Date: Sun, 31 May 2026 11:11:04 +0200 Subject: [PATCH] feat(berlinmod): streaming benchmark harness with corpus-derived parameters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BerlinMODBenchmark runs each of the 27 BerlinMOD-9 × 3-form cells in isolation through a TopologyTestDriver and reads its output cardinality (correctness); EmbeddedBrokerBenchmark runs each cell as a KafkaStreams application against a fresh in-process EmbeddedKafkaCluster and reports steady-state per-event throughput. Both share the corpus-derived query parameters and window/tick granularity (BerlinMODCorpus.derive) and evaluate the spatial predicates through MEOS. On the 216,075-instant real BerlinMOD corpus the per-event spatial cells sustain 45,000-47,000 ev/s; docs/benchmark.md carries the full 27-cell table. --- kafka-streams-app/docs/benchmark.md | 98 +++++ kafka-streams-app/pom.xml | 54 +++ .../java/berlinmod/BerlinMODBenchmark.java | 118 ++++++ .../main/java/berlinmod/BerlinMODCorpus.java | 165 +++++++++ .../java/berlinmod/BerlinMODTopology.java | 325 +++++++++-------- .../berlinmod/EmbeddedBrokerBenchmark.java | 335 ++++++++++++++++++ 6 files changed, 942 insertions(+), 153 deletions(-) create mode 100644 kafka-streams-app/docs/benchmark.md create mode 100644 kafka-streams-app/src/main/java/berlinmod/BerlinMODBenchmark.java create mode 100644 kafka-streams-app/src/main/java/berlinmod/BerlinMODCorpus.java create mode 100644 kafka-streams-app/src/test/java/berlinmod/EmbeddedBrokerBenchmark.java diff --git a/kafka-streams-app/docs/benchmark.md b/kafka-streams-app/docs/benchmark.md new file mode 100644 index 0000000..260a946 --- /dev/null +++ b/kafka-streams-app/docs/benchmark.md @@ -0,0 +1,98 @@ +# BerlinMOD streaming benchmark (Kafka Streams) + +The Kafka-Streams counterpart of the MobilityFlink BerlinMOD benchmark covers +the 27 BerlinMOD-9 × 3-form cells (Q1–Q9 × continuous / windowed / snapshot) +with two harnesses, both over the same corpus and corpus-derived parameters, and +with the spatial predicates evaluating through MEOS (see +[`MEOSBridge`](../src/main/java/berlinmod/MEOSBridge.java)): + +- [`BerlinMODBenchmark`](../src/main/java/berlinmod/BerlinMODBenchmark.java) runs + each cell in isolation through a [`TopologyTestDriver`](../src/main/java/berlinmod/BerlinMODTopology.java) + and reads its output cardinality — an in-process **correctness** harness. +- [`EmbeddedBrokerBenchmark`](../src/test/java/berlinmod/EmbeddedBrokerBenchmark.java) + runs each cell as a real [`KafkaStreams`](../src/main/java/berlinmod/BerlinMODTopology.java) + application against an in-process `EmbeddedKafkaCluster` (a genuine + `KafkaServer` over the loopback network) — the **throughput** harness, the + Flink-comparable analog of MobilityFlink's per-cell jobs. + +## Corpus and parameters (regular with MobilityFlink) + +The corpus is the real BerlinMOD instants (`--csv`, reprojected EPSG:3857→4326 +through MEOS by [`BerlinMODCorpus`](../src/main/java/berlinmod/BerlinMODCorpus.java)) +or a synthetic corpus. The per-query parameters (point `P`, region box, road +segment, points of interest, target vehicle ids) **and** the window/tick +granularity are derived from the corpus via `BerlinMODCorpus.derive` — the same +mechanism MobilityFlink uses — and threaded through `BerlinMODTopology.build(Params)`, +so the topology auto-scales to the corpus span instead of carrying a fixed +window/tick. The two stream bindings share one mechanism. + +## Throughput harness + +Each cell runs against its own fresh `EmbeddedKafkaCluster`, the true analog of +MobilityFlink's independent per-cell jobs: the corpus is produced once into the +input topic, the cell runs as a single-threaded `KafkaStreams` application, and +throughput is the events consumed divided by the wall-clock from streams start +until the application has read the whole input topic (its own +`records-consumed-total` metric reaches the input end offset) and its output has +gone idle. The trailing settle time is excluded from the wall; each consumed +record runs the cell's MEOS predicate, so this is the steady-state per-event +processing rate, directly comparable to the MobilityFlink figures. + +Run from `kafka-streams-app/`, with an extended libmeos on the loader path and +the test classpath (it carries the embedded broker): + +``` +CP=$(mvn -q dependency:build-classpath -DincludeScope=test -Dmdep.outputFile=/dev/stdout | tail -1) +LD_LIBRARY_PATH= java -Dorg.slf4j.simpleLogger.defaultLogLevel=warn \ + -cp target/classes:target/test-classes:jar/JMEOS.jar:$CP \ + berlinmod.EmbeddedBrokerBenchmark --csv [--max N] [--only Q3-continuous] +``` + +## Figures + +Real BerlinMOD corpus (216,075 instants, 5 vehicles, ~11 days, EPSG:3857), +single-broker `EmbeddedKafkaCluster`, one stream thread per cell, Java 21, +16-core host; libmeos built with `-DMEOS/CBUFFER/NPOINT/POSE/RGEO=ON`. + +| Cell | Events in | Output rows | Wall (ms) | Throughput (ev/s) | +|---|---:|---:|---:|---:| +| Q1-continuous | 216075 | 5 | 1899 | 113,785 | +| Q1-snapshot | 216075 | 703 | 1701 | 127,029 | +| Q1-windowed | 216075 | 86 | 1650 | 130,956 | +| Q2-continuous | 216075 | 61170 | 1289 | 167,631 | +| Q2-snapshot | 216075 | 140 | 1683 | 128,388 | +| Q2-windowed | 216075 | 50 | 1704 | 126,806 | +| Q3-continuous | 216075 | 216075 | 4618 | 46,790 | +| Q3-snapshot | 216075 | 97 | 2607 | 82,883 | +| Q3-windowed | 216075 | 50 | 4188 | 51,594 | +| Q4-continuous | 216075 | 62 | 9356 | 23,095 | +| Q4-snapshot | 216075 | 4685 | 10499 | 20,581 | +| Q4-windowed | 216075 | 98 | 9741 | 22,182 | +| Q5-continuous | 216075 | 60577 | 22889 | 9,440 | +| Q5-snapshot | 216075 | 34 | 2006 | 107,715 | +| Q5-windowed | 216075 | 6 | 2896 | 74,612 | +| Q6-continuous | 216075 | 216075 | 8708 | 24,814 | +| Q6-snapshot | 216075 | 698 | 7258 | 29,771 | +| Q6-windowed | 216075 | 203 | 6268 | 34,473 | +| Q7-continuous | 216075 | 5 | 4321 | 50,006 | +| Q7-snapshot | 216075 | 632 | 7487 | 28,860 | +| Q7-windowed | 216075 | 53 | 11625 | 18,587 | +| Q8-continuous | 216075 | 216075 | 4807 | 44,950 | +| Q8-snapshot | 216075 | 281 | 2825 | 76,487 | +| Q8-windowed | 216075 | 77 | 5946 | 36,340 | +| Q9-continuous | 216075 | 107870 | 4561 | 47,375 | +| Q9-snapshot | 216075 | 140 | 2281 | 94,729 | +| Q9-windowed | 216075 | 22 | 2384 | 90,636 | + +The per-event MEOS-predicate cells that emit one row per input event +(Q3/Q8/Q9-continuous, 216,075 output rows through `edwithin_tgeo_geo` / +`eintersects_tgeo_geo`) sustain 45,000–47,000 ev/s. In the cross-platform +streaming catalog these cells run below MobilityFlink's in-JVM mini-cluster on +the same corpus, because Kafka Streams routes every record through the broker; +the per-cell shape matches across both engines. Q5-continuous is the O(V²) +all-pairs-meeting outlier (9,440 ev/s). Non-spatial cells (Q1/Q2) reach +110,000–168,000 ev/s. The `TopologyTestDriver` correctness harness +([`BerlinMODBenchmark`](../src/main/java/berlinmod/BerlinMODBenchmark.java)) +reports output cardinality and correctness, not throughput: its per-event +bookkeeping dominates wall-clock, so a no-predicate cell already runs far below a +real broker's rate. diff --git a/kafka-streams-app/pom.xml b/kafka-streams-app/pom.xml index d3bdd40..86dc3f8 100644 --- a/kafka-streams-app/pom.xml +++ b/kafka-streams-app/pom.xml @@ -71,6 +71,60 @@ ${junit.version} test + + + + org.apache.kafka + kafka_2.13 + ${kafka.version} + test + + + org.apache.kafka + kafka_2.13 + ${kafka.version} + test-jar + test + test + + + org.apache.kafka + kafka-streams + ${kafka.version} + test-jar + test + test + + + org.apache.kafka + kafka-clients + ${kafka.version} + test-jar + test + test + + + org.apache.kafka + kafka-server-common + ${kafka.version} + test-jar + test + test + + + + org.hamcrest + hamcrest + 2.2 + test + diff --git a/kafka-streams-app/src/main/java/berlinmod/BerlinMODBenchmark.java b/kafka-streams-app/src/main/java/berlinmod/BerlinMODBenchmark.java new file mode 100644 index 0000000..e3b4edf --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/BerlinMODBenchmark.java @@ -0,0 +1,118 @@ +package berlinmod; + +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.TestInputTopic; +import org.apache.kafka.streams.TestOutputTopic; +import org.apache.kafka.streams.TopologyTestDriver; + +import java.lang.reflect.Field; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import java.util.TreeMap; + +/** + * Per-cell throughput benchmark for the BerlinMOD-9 × 3-form streaming matrix on + * Kafka Streams. + * + *

Each cell runs in isolation through its own single-cell topology + * ({@link BerlinMODTopology#buildCell}) and {@link TopologyTestDriver} over the + * corpus — the Kafka-Streams analog of MobilityFlink's per-cell jobs, so the + * per-cell wall-clock and throughput are independent and comparable. The corpus + * and the per-query parameters are corpus-derived via {@link BerlinMODCorpus}; + * the spatial predicates evaluate through MEOS (see {@link MEOSBridge}). + * + * 

+ *   java … berlinmod.BerlinMODBenchmark --csv <berlinmod_instants.csv> [--max N]
+ *   java … berlinmod.BerlinMODBenchmark --vehicles 50 --events 600 [--only Q3-continuous]
+ *
+ */ +public final class BerlinMODBenchmark { + + private BerlinMODBenchmark() { /* utility */ } + + public static void main(String[] args) throws Exception { + String csv = null, only = null; + int maxRows = 0, vehicles = 50, events = 600; + for (int i = 0; i < args.length; i++) { + switch (args[i]) { + case "--csv": csv = args[++i]; break; + case "--max": maxRows = Integer.parseInt(args[++i]); break; + case "--vehicles": vehicles = Integer.parseInt(args[++i]); break; + case "--events": events = Integer.parseInt(args[++i]); break; + case "--only": only = args[++i]; break; + default: break; + } + } + List corpus = csv != null + ? BerlinMODCorpus.fromInstantsCsv(csv, maxRows) + : BerlinMODCorpus.synthetic(vehicles, events); + int n = corpus.size(); + long maxTs = corpus.stream().mapToLong(BerlinMODTrip::getTimestamp).max().orElse(0L); + BerlinMODCorpus.Params p = BerlinMODCorpus.derive(corpus); + System.out.printf("Corpus: %s, %d events; window=%ds tick=%dms%n", + csv != null ? "real BerlinMOD instants" : "synthetic", n, p.windowSeconds, p.snapshotTickMillis); + + // (cellName, outputTopic) enumerated from the topology's *_OUTPUT constants. + TreeMap cells = new TreeMap<>(); + for (Field f : BerlinMODTopology.class.getFields()) { + if (f.getName().endsWith("_OUTPUT") && f.getType() == String.class) { + cells.put(cellName(f.getName()), (String) f.get(null)); + } + } + + Properties props = new Properties(); + props.put(StreamsConfig.APPLICATION_ID_CONFIG, "berlinmod-bench"); + props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092"); + props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass()); + props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArray().getClass()); + BerlinMODTripSerde tripSerde = new BerlinMODTripSerde(); + + List rows = new ArrayList<>(); + for (var e : cells.entrySet()) { + String cell = e.getKey(), topic = e.getValue(); + if (only != null && !cell.equals(only)) { + continue; + } + long out; + long t0 = System.nanoTime(); + try (TopologyTestDriver driver = new TopologyTestDriver(BerlinMODTopology.buildCell(p, cell), props)) { + TestInputTopic input = driver.createInputTopic( + BerlinMODTopology.INPUT_TOPIC, new IntegerSerializer(), tripSerde.serializer()); + for (BerlinMODTrip trip : corpus) { + input.pipeInput(trip.getVehicleId(), trip, Instant.ofEpochMilli(trip.getTimestamp())); + } + input.pipeInput(-1, new BerlinMODTrip(-1, maxTs + 3_600_000L, 0.0, 0.0), + Instant.ofEpochMilli(maxTs + 3_600_000L)); + input.pipeInput(-1, new BerlinMODTrip(-1, maxTs + 7_200_000L, 0.0, 0.0), + Instant.ofEpochMilli(maxTs + 7_200_000L)); + TestOutputTopic output = driver.createOutputTopic( + topic, new ByteArrayDeserializer(), new ByteArrayDeserializer()); + out = output.readRecordsToList().size(); + } + long wallMs = (System.nanoTime() - t0) / 1_000_000L; + double tput = wallMs > 0 ? n / (wallMs / 1000.0) : 0; + rows.add(new String[]{cell, String.valueOf(n), String.valueOf(out), + String.valueOf(wallMs), String.format("%,.0f", tput)}); + System.out.printf(" %-14s out=%-8d %6d ms %,.0f ev/s%n", cell, out, wallMs, tput); + } + + System.out.println(); + System.out.println("| Cell | Events in | Output rows | Wall (ms) | Throughput (ev/s) |"); + System.out.println("|---|---:|---:|---:|---:|"); + for (String[] r : rows) { + System.out.printf("| %s | %s | %s | %s | %s |%n", r[0], r[1], r[2], r[3], r[4]); + } + } + + /** Q3_CONTINUOUS_OUTPUT → Q3-continuous */ + private static String cellName(String field) { + String s = field.substring(0, field.length() - "_OUTPUT".length()); + int us = s.indexOf('_'); + return s.substring(0, us) + "-" + s.substring(us + 1).toLowerCase(); + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/BerlinMODCorpus.java b/kafka-streams-app/src/main/java/berlinmod/BerlinMODCorpus.java new file mode 100644 index 0000000..749d3b2 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/BerlinMODCorpus.java @@ -0,0 +1,165 @@ +package berlinmod; + +import functions.GeneratedFunctions; +import jnr.ffi.Pointer; + +import java.nio.file.Files; +import java.nio.file.Paths; +import java.time.OffsetDateTime; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.temporal.ChronoField; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.TreeSet; +import java.util.stream.Stream; + +/** + * Corpus loader and query-parameter derivation for the BerlinMOD streaming + * benchmark. + * + *

Supplies either a deterministic synthetic corpus or the real BerlinMOD + * instants corpus read from the {@code berlinmod_instants.csv} produced by the + * BerlinMOD generator. Real instants are stored in EPSG:3857; they are + * reprojected to EPSG:4326 through MEOS {@code geo_transform} at load — the + * loader holds no projection mathematics of its own. + * + *

{@link Params} fixes the per-query parameters from the corpus itself (its + * centroid, bounding box, vehicle ids, and time span) so every spatial cell is + * selective and the windowing granularity yields a comparable number of windows + * regardless of the corpus time span. + */ +public final class BerlinMODCorpus { + + private static final DateTimeFormatter TS = new DateTimeFormatterBuilder() + .appendPattern("yyyy-MM-dd HH:mm:ss") + .optionalStart().appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true).optionalEnd() + .appendOffset("+HH", "Z") + .toFormatter(); + + private BerlinMODCorpus() { /* utility */ } + + /** Query parameters derived from a corpus. */ + public static final class Params { + public final double pLon, pLat, radiusMetres, dMeetMetres; + public final double xmin, ymin, xmax, ymax; + public final double s1Lon, s1Lat, s2Lon, s2Lat; + public final List pois; + public final int targetId, xId, yId; + public final long windowSeconds, snapshotTickMillis; + + Params(double pLon, double pLat, double radiusMetres, double dMeetMetres, + double xmin, double ymin, double xmax, double ymax, + double s1Lon, double s1Lat, double s2Lon, double s2Lat, + List pois, int targetId, int xId, int yId, + long windowSeconds, long snapshotTickMillis) { + this.pLon = pLon; this.pLat = pLat; this.radiusMetres = radiusMetres; this.dMeetMetres = dMeetMetres; + this.xmin = xmin; this.ymin = ymin; this.xmax = xmax; this.ymax = ymax; + this.s1Lon = s1Lon; this.s1Lat = s1Lat; this.s2Lon = s2Lon; this.s2Lat = s2Lat; + this.pois = pois; this.targetId = targetId; this.xId = xId; this.yId = yId; + this.windowSeconds = windowSeconds; this.snapshotTickMillis = snapshotTickMillis; + } + } + + /** Deterministic synthetic corpus: vehicles on a disc around Brussels centre, + * drifting per event, with monotonically increasing timestamps. */ + public static List synthetic(int vehicles, int perVehicle) { + final double centreLon = 4.3517, centreLat = 50.8503, spread = 0.12; + final long t0 = 1_735_711_200_000L, spanMillis = 600_000L; + int total = vehicles * perVehicle; + long step = Math.max(1L, spanMillis / total); + List events = new ArrayList<>(total); + long g = 0; + for (int e = 0; e < perVehicle; e++) { + for (int v = 0; v < vehicles; v++) { + double ang = (v * 2.399963) % (2 * Math.PI); + double rad = spread * ((v % 17) / 17.0); + double drift = 0.0005 * Math.sin((e + v) * 0.13); + events.add(make(100 + v, t0 + g * step, + centreLon + rad * Math.cos(ang) + drift, + centreLat + rad * Math.sin(ang) + drift)); + g++; + } + } + return events; + } + + /** Real BerlinMOD instants from {@code berlinmod_instants.csv} + * (columns {@code tripid,vehid,day,seqno,geom,t}), reprojected 3857→4326 + * through MEOS, sorted by timestamp. {@code maxRows <= 0} loads all rows. */ + public static List fromInstantsCsv(String path, int maxRows) throws Exception { + ensureMeos(); + List events = new ArrayList<>(); + try (Stream lines = Files.lines(Paths.get(path))) { + java.util.Iterator it = lines.iterator(); + if (it.hasNext()) { + it.next(); // header + } + while (it.hasNext() && (maxRows <= 0 || events.size() < maxRows)) { + String[] f = it.next().split(","); + int vid = Integer.parseInt(f[1].trim()); + long ms = OffsetDateTime.parse(f[5].trim(), TS).toInstant().toEpochMilli(); + Pointer g4326 = GeneratedFunctions.geo_transform( + GeneratedFunctions.geom_in(f[4].trim(), -1), 4326); + String txt = GeneratedFunctions.geo_as_text(g4326, 7); // POINT(lon lat) + String[] xy = txt.substring(txt.indexOf('(') + 1, txt.indexOf(')')).trim().split("\\s+"); + events.add(make(vid, ms, Double.parseDouble(xy[0]), Double.parseDouble(xy[1]))); + } + } + events.sort((a, b) -> Long.compare(a.getTimestamp(), b.getTimestamp())); + return events; + } + + /** Derive selective per-query parameters and a window/tick granularity that + * yields ~200 windows over the corpus time span. */ + public static Params derive(List corpus) { + double sumLon = 0, sumLat = 0, minLon = Double.MAX_VALUE, minLat = Double.MAX_VALUE, + maxLon = -Double.MAX_VALUE, maxLat = -Double.MAX_VALUE, minT = Double.MAX_VALUE, maxT = -Double.MAX_VALUE; + TreeSet ids = new TreeSet<>(); + for (BerlinMODTrip t : corpus) { + sumLon += t.getLon(); sumLat += t.getLat(); + minLon = Math.min(minLon, t.getLon()); maxLon = Math.max(maxLon, t.getLon()); + minLat = Math.min(minLat, t.getLat()); maxLat = Math.max(maxLat, t.getLat()); + minT = Math.min(minT, t.getTimestamp()); maxT = Math.max(maxT, t.getTimestamp()); + ids.add(t.getVehicleId()); + } + int n = corpus.size(); + double cLon = sumLon / n, cLat = sumLat / n; + double exLon = maxLon - minLon, exLat = maxLat - minLat; + List idList = new ArrayList<>(ids); + int targetId = idList.get(idList.size() / 2); + int xId = idList.get(0); + int yId = idList.get(Math.min(idList.size() - 1, idList.size() / 2)); + long span = (long) (maxT - minT); + long windowSeconds = Math.max(1L, span / 1000 / 200); + long tickMillis = Math.max(1000L, windowSeconds * 1000L / 2); + List pois = Arrays.asList( + new PointOfInterest(1, cLon, cLat, 2_000.0), + new PointOfInterest(2, cLon + 0.1 * exLon, cLat + 0.1 * exLat, 1_000.0), + new PointOfInterest(3, cLon - 0.1 * exLon, cLat - 0.1 * exLat, 2_000.0)); + return new Params(cLon, cLat, 5_000.0, 5_000.0, + cLon - 0.25 * exLon, cLat - 0.25 * exLat, cLon + 0.25 * exLon, cLat + 0.25 * exLat, + minLon + 0.25 * exLon, cLat, maxLon - 0.25 * exLon, cLat, + pois, targetId, xId, yId, windowSeconds, tickMillis); + } + + private static BerlinMODTrip make(int vid, long t, double lon, double lat) { + BerlinMODTrip trip = new BerlinMODTrip(); + trip.setVehicleId(vid); + trip.setTimestamp(t); + trip.setLon(lon); + trip.setLat(lat); + return trip; + } + + private static final ThreadLocal MEOS_INIT = ThreadLocal.withInitial(() -> Boolean.FALSE); + private static void ensureMeos() { + if (!MEOS_INIT.get()) { + GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); + GeneratedFunctions.meos_initialize(); + MEOS_INIT.set(Boolean.TRUE); + } + } + +} diff --git a/kafka-streams-app/src/main/java/berlinmod/BerlinMODTopology.java b/kafka-streams-app/src/main/java/berlinmod/BerlinMODTopology.java index f4a2a4b..bdf3ae4 100644 --- a/kafka-streams-app/src/main/java/berlinmod/BerlinMODTopology.java +++ b/kafka-streams-app/src/main/java/berlinmod/BerlinMODTopology.java @@ -116,165 +116,184 @@ public final class BerlinMODTopology { private BerlinMODTopology() {} - public static Topology build() { + public static Topology build() { return build(defaultParams()); } + public static Topology build(BerlinMODCorpus.Params p) { return buildInternal(p, null); } + public static Topology buildCell(BerlinMODCorpus.Params p, String cell) { return buildInternal(p, cell); } + + private static BerlinMODCorpus.Params defaultParams() { + return new BerlinMODCorpus.Params(Q3_P_LON, Q3_P_LAT, Q3_RADIUS_METRES, Q5_D_MEET_METRES, + Q4_XMIN, Q4_YMIN, Q4_XMAX, Q4_YMAX, Q8_S1_LON, Q8_S1_LAT, Q8_S2_LON, Q8_S2_LAT, + Q7_POIS, Q2_TARGET_VEHICLE_ID, Q9_X_VEHICLE_ID, Q9_Y_VEHICLE_ID, + WINDOW_SIZE_MILLIS / 1000L, SNAPSHOT_TICK_MILLIS); + } + + private static Topology buildInternal(BerlinMODCorpus.Params p, String only) { StreamsBuilder builder = new StreamsBuilder(); + final long WIN_MS = p.windowSeconds * 1000L; BerlinMODTripSerde tripSerde = new BerlinMODTripSerde(); - - // ---- continuous-form state stores ---- - addStore(builder, Q1_SEEN_STORE, Serdes.Integer(), Serdes.Boolean()); - addStore(builder, Q4_WAS_INSIDE_STORE, Serdes.Integer(), Serdes.Boolean()); - addStore(builder, Q5_LAST_POS_STORE, Serdes.Integer(), Serdes.String()); - addStore(builder, Q6_STATE_STORE, Serdes.Integer(), Serdes.String()); - addStore(builder, Q7_FIRST_PASSED_STORE, Serdes.Integer(), Serdes.Long()); - addStore(builder, Q9_STATE_STORE, Serdes.Integer(), Serdes.String()); - - // ---- windowed-form state stores ---- - addStore(builder, Q1_WIN_STORE, Serdes.Long(), Serdes.String()); - addStore(builder, Q2_WIN_STORE, Serdes.Long(), Serdes.String()); - addStore(builder, Q3_WIN_STORE, Serdes.Long(), Serdes.String()); - addStore(builder, Q4_WIN_STORE, Serdes.Long(), Serdes.String()); - addStore(builder, Q5_WIN_STORE, Serdes.Long(), Serdes.String()); - addStore(builder, Q6_WIN_STORE, Serdes.Long(), Serdes.String()); - addStore(builder, Q7_WIN_STORE, Serdes.Long(), Serdes.String()); - addStore(builder, Q8_WIN_STORE, Serdes.Long(), Serdes.String()); - addStore(builder, Q9_WIN_STORE, Serdes.Long(), Serdes.String()); - - // ---- snapshot-form state stores (separate to avoid co-write conflicts with continuous) ---- - addStore(builder, Q1_SNAP_STORE, Serdes.Integer(), Serdes.Long()); - addStore(builder, Q2_SNAP_STORE, Serdes.Integer(), Serdes.String()); - addStore(builder, Q3_SNAP_STORE, Serdes.Integer(), Serdes.String()); - addStore(builder, Q4_SNAP_WAS_INSIDE_STORE, Serdes.Integer(), Serdes.Boolean()); - addStore(builder, Q4_SNAP_ENTRIES_STORE, Serdes.Integer(), Serdes.String()); - addStore(builder, Q5_SNAP_STORE, Serdes.Integer(), Serdes.String()); - addStore(builder, Q6_SNAP_STORE, Serdes.Integer(), Serdes.String()); - addStore(builder, Q7_SNAP_STORE, Serdes.Integer(), Serdes.Long()); - addStore(builder, Q8_SNAP_STORE, Serdes.Integer(), Serdes.String()); - addStore(builder, Q9_SNAP_STORE, Serdes.Integer(), Serdes.String()); - - // ---- streams ---- KStream trips = builder.stream(INPUT_TOPIC, Consumed.with(Serdes.Integer(), tripSerde)); - - // Re-keyed by constant for the shared-state snapshot/multi-vehicle processors KStream tripsK0 = trips.selectKey((k, v) -> 0); - // ====== continuous form ====== - trips.process(() -> new Q1ContinuousProcessor(Q1_SEEN_STORE), Q1_SEEN_STORE) - .to(Q1_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Long())); - - trips.process(() -> new Q2ContinuousProcessor(Q2_TARGET_VEHICLE_ID)) - .to(Q2_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), tripSerde)); - - trips.process(() -> new Q3ContinuousProcessor(Q3_P_LON, Q3_P_LAT, Q3_RADIUS_METRES)) - .to(Q3_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Boolean())); - - trips.process(() -> new Q4ContinuousProcessor(Q4_WAS_INSIDE_STORE, Q4_XMIN, Q4_YMIN, Q4_XMAX, Q4_YMAX), - Q4_WAS_INSIDE_STORE) - .to(Q4_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Long())); - - tripsK0.process(() -> new Q5ContinuousProcessor(Q5_LAST_POS_STORE, - Q5_P_LON, Q5_P_LAT, Q5_D_P_METRES, Q5_D_MEET_METRES), - Q5_LAST_POS_STORE) - .to(Q5_CONTINUOUS_OUTPUT, Produced.with(Serdes.String(), Serdes.Double())); - - trips.process(() -> new Q6ContinuousProcessor(Q6_STATE_STORE), Q6_STATE_STORE) - .to(Q6_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Double())); - - trips.process(() -> new Q7ContinuousProcessor(Q7_FIRST_PASSED_STORE, Q7_POIS), - Q7_FIRST_PASSED_STORE) - .to(Q7_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Long())); - - trips.process(() -> new Q8ContinuousProcessor(Q8_S1_LON, Q8_S1_LAT, Q8_S2_LON, Q8_S2_LAT, Q8_RADIUS_METRES)) - .to(Q8_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Boolean())); - - tripsK0.process(() -> new Q9ContinuousProcessor(Q9_STATE_STORE, Q9_X_VEHICLE_ID, Q9_Y_VEHICLE_ID), - Q9_STATE_STORE) - .to(Q9_CONTINUOUS_OUTPUT, Produced.with(Serdes.Long(), Serdes.Double())); - - // ====== windowed form (distinct-count per tumbling window for Q1/Q3/Q8) ====== - tripsK0.process(() -> new Q1WindowedProcessor(Q1_WIN_STORE, WINDOW_SIZE_MILLIS), Q1_WIN_STORE) - .to(Q1_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Long())); - - tripsK0.process(() -> new Q3WindowedProcessor(Q3_WIN_STORE, - Q3_P_LON, Q3_P_LAT, Q3_RADIUS_METRES, - WINDOW_SIZE_MILLIS), - Q3_WIN_STORE) - .to(Q3_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Long())); - - tripsK0.process(() -> new Q8WindowedProcessor(Q8_WIN_STORE, - Q8_S1_LON, Q8_S1_LAT, Q8_S2_LON, Q8_S2_LAT, - Q8_RADIUS_METRES, WINDOW_SIZE_MILLIS), - Q8_WIN_STORE) - .to(Q8_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Long())); - - tripsK0.process(() -> new Q2WindowedProcessor(Q2_WIN_STORE, Q2_TARGET_VEHICLE_ID, WINDOW_SIZE_MILLIS), - Q2_WIN_STORE) - .to(Q2_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); - - tripsK0.process(() -> new Q4WindowedProcessor(Q4_WIN_STORE, - Q4_XMIN, Q4_YMIN, Q4_XMAX, Q4_YMAX, - WINDOW_SIZE_MILLIS), - Q4_WIN_STORE) - .to(Q4_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); - - tripsK0.process(() -> new Q5WindowedProcessor(Q5_WIN_STORE, - Q5_P_LON, Q5_P_LAT, Q5_D_P_METRES, Q5_D_MEET_METRES, - WINDOW_SIZE_MILLIS), - Q5_WIN_STORE) - .to(Q5_WINDOWED_OUTPUT, Produced.with(Serdes.String(), Serdes.Double())); - - tripsK0.process(() -> new Q6WindowedProcessor(Q6_WIN_STORE, WINDOW_SIZE_MILLIS), Q6_WIN_STORE) - .to(Q6_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); - - tripsK0.process(() -> new Q7WindowedProcessor(Q7_WIN_STORE, Q7_POIS, WINDOW_SIZE_MILLIS), Q7_WIN_STORE) - .to(Q7_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); - - tripsK0.process(() -> new Q9WindowedProcessor(Q9_WIN_STORE, - Q9_X_VEHICLE_ID, Q9_Y_VEHICLE_ID, - WINDOW_SIZE_MILLIS), - Q9_WIN_STORE) - .to(Q9_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Double())); - - // ====== snapshot form (all via constant key, with STREAM_TIME punctuators) ====== - tripsK0.process(() -> new Q1SnapshotProcessor(Q1_SNAP_STORE, SNAPSHOT_TICK_MILLIS), Q1_SNAP_STORE) - .to(Q1_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Integer())); - - tripsK0.process(() -> new Q2SnapshotProcessor(Q2_SNAP_STORE, Q2_TARGET_VEHICLE_ID, SNAPSHOT_TICK_MILLIS), - Q2_SNAP_STORE) - .to(Q2_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); - - tripsK0.process(() -> new Q3SnapshotProcessor(Q3_SNAP_STORE, - Q3_P_LON, Q3_P_LAT, Q3_RADIUS_METRES, SNAPSHOT_TICK_MILLIS), - Q3_SNAP_STORE) - .to(Q3_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Integer())); - - tripsK0.process(() -> new Q4SnapshotProcessor(Q4_SNAP_WAS_INSIDE_STORE, Q4_SNAP_ENTRIES_STORE, - Q4_XMIN, Q4_YMIN, Q4_XMAX, Q4_YMAX, SNAPSHOT_TICK_MILLIS), - Q4_SNAP_WAS_INSIDE_STORE, Q4_SNAP_ENTRIES_STORE) - .to(Q4_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); - - tripsK0.process(() -> new Q5SnapshotProcessor(Q5_SNAP_STORE, - Q5_P_LON, Q5_P_LAT, Q5_D_P_METRES, Q5_D_MEET_METRES, - SNAPSHOT_TICK_MILLIS), - Q5_SNAP_STORE) - .to(Q5_SNAPSHOT_OUTPUT, Produced.with(Serdes.String(), Serdes.Double())); - - tripsK0.process(() -> new Q6SnapshotProcessor(Q6_SNAP_STORE, SNAPSHOT_TICK_MILLIS), Q6_SNAP_STORE) - .to(Q6_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); - - tripsK0.process(() -> new Q7SnapshotProcessor(Q7_SNAP_STORE, Q7_POIS, SNAPSHOT_TICK_MILLIS), Q7_SNAP_STORE) - .to(Q7_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); - - tripsK0.process(() -> new Q8SnapshotProcessor(Q8_SNAP_STORE, - Q8_S1_LON, Q8_S1_LAT, Q8_S2_LON, Q8_S2_LAT, - Q8_RADIUS_METRES, SNAPSHOT_TICK_MILLIS), - Q8_SNAP_STORE) - .to(Q8_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Integer())); - - tripsK0.process(() -> new Q9SnapshotProcessor(Q9_SNAP_STORE, - Q9_X_VEHICLE_ID, Q9_Y_VEHICLE_ID, SNAPSHOT_TICK_MILLIS), - Q9_SNAP_STORE) - .to(Q9_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Double())); + if (only == null || only.equals("Q1-continuous")) { + addStore(builder, Q1_SEEN_STORE, Serdes.Integer(), Serdes.Boolean()); + trips.process(() -> new Q1ContinuousProcessor(Q1_SEEN_STORE), Q1_SEEN_STORE) + .to(Q1_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Long())); + } + + if (only == null || only.equals("Q2-continuous")) { + trips.process(() -> new Q2ContinuousProcessor(p.targetId)) + .to(Q2_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), tripSerde)); + } + + if (only == null || only.equals("Q3-continuous")) { + trips.process(() -> new Q3ContinuousProcessor(p.pLon, p.pLat, p.radiusMetres)) + .to(Q3_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Boolean())); + } + + if (only == null || only.equals("Q4-continuous")) { + addStore(builder, Q4_WAS_INSIDE_STORE, Serdes.Integer(), Serdes.Boolean()); + trips.process(() -> new Q4ContinuousProcessor(Q4_WAS_INSIDE_STORE, p.xmin, p.ymin, p.xmax, p.ymax), Q4_WAS_INSIDE_STORE) + .to(Q4_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Long())); + } + + if (only == null || only.equals("Q5-continuous")) { + addStore(builder, Q5_LAST_POS_STORE, Serdes.Integer(), Serdes.String()); + tripsK0.process(() -> new Q5ContinuousProcessor(Q5_LAST_POS_STORE, p.pLon, p.pLat, p.radiusMetres, p.dMeetMetres), Q5_LAST_POS_STORE) + .to(Q5_CONTINUOUS_OUTPUT, Produced.with(Serdes.String(), Serdes.Double())); + } + + if (only == null || only.equals("Q6-continuous")) { + addStore(builder, Q6_STATE_STORE, Serdes.Integer(), Serdes.String()); + trips.process(() -> new Q6ContinuousProcessor(Q6_STATE_STORE), Q6_STATE_STORE) + .to(Q6_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Double())); + } + + if (only == null || only.equals("Q7-continuous")) { + addStore(builder, Q7_FIRST_PASSED_STORE, Serdes.Integer(), Serdes.Long()); + trips.process(() -> new Q7ContinuousProcessor(Q7_FIRST_PASSED_STORE, p.pois), Q7_FIRST_PASSED_STORE) + .to(Q7_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Long())); + } + + if (only == null || only.equals("Q8-continuous")) { + trips.process(() -> new Q8ContinuousProcessor(p.s1Lon, p.s1Lat, p.s2Lon, p.s2Lat, p.radiusMetres)) + .to(Q8_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Boolean())); + } + + if (only == null || only.equals("Q9-continuous")) { + addStore(builder, Q9_STATE_STORE, Serdes.Integer(), Serdes.String()); + tripsK0.process(() -> new Q9ContinuousProcessor(Q9_STATE_STORE, p.xId, p.yId), Q9_STATE_STORE) + .to(Q9_CONTINUOUS_OUTPUT, Produced.with(Serdes.Long(), Serdes.Double())); + } + + if (only == null || only.equals("Q1-windowed")) { + addStore(builder, Q1_WIN_STORE, Serdes.Long(), Serdes.String()); + tripsK0.process(() -> new Q1WindowedProcessor(Q1_WIN_STORE, WIN_MS), Q1_WIN_STORE) + .to(Q1_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Long())); + } + + if (only == null || only.equals("Q2-windowed")) { + addStore(builder, Q2_WIN_STORE, Serdes.Long(), Serdes.String()); + tripsK0.process(() -> new Q2WindowedProcessor(Q2_WIN_STORE, p.targetId, WIN_MS), Q2_WIN_STORE) + .to(Q2_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + } + + if (only == null || only.equals("Q3-windowed")) { + addStore(builder, Q3_WIN_STORE, Serdes.Long(), Serdes.String()); + tripsK0.process(() -> new Q3WindowedProcessor(Q3_WIN_STORE, p.pLon, p.pLat, p.radiusMetres, WIN_MS), Q3_WIN_STORE) + .to(Q3_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Long())); + } + + if (only == null || only.equals("Q4-windowed")) { + addStore(builder, Q4_WIN_STORE, Serdes.Long(), Serdes.String()); + tripsK0.process(() -> new Q4WindowedProcessor(Q4_WIN_STORE, p.xmin, p.ymin, p.xmax, p.ymax, WIN_MS), Q4_WIN_STORE) + .to(Q4_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + } + + if (only == null || only.equals("Q5-windowed")) { + addStore(builder, Q5_WIN_STORE, Serdes.Long(), Serdes.String()); + tripsK0.process(() -> new Q5WindowedProcessor(Q5_WIN_STORE, p.pLon, p.pLat, p.radiusMetres, p.dMeetMetres, WIN_MS), Q5_WIN_STORE) + .to(Q5_WINDOWED_OUTPUT, Produced.with(Serdes.String(), Serdes.Double())); + } + + if (only == null || only.equals("Q6-windowed")) { + addStore(builder, Q6_WIN_STORE, Serdes.Long(), Serdes.String()); + tripsK0.process(() -> new Q6WindowedProcessor(Q6_WIN_STORE, WIN_MS), Q6_WIN_STORE) + .to(Q6_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + } + + if (only == null || only.equals("Q7-windowed")) { + addStore(builder, Q7_WIN_STORE, Serdes.Long(), Serdes.String()); + tripsK0.process(() -> new Q7WindowedProcessor(Q7_WIN_STORE, p.pois, WIN_MS), Q7_WIN_STORE) + .to(Q7_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + } + + if (only == null || only.equals("Q8-windowed")) { + addStore(builder, Q8_WIN_STORE, Serdes.Long(), Serdes.String()); + tripsK0.process(() -> new Q8WindowedProcessor(Q8_WIN_STORE, p.s1Lon, p.s1Lat, p.s2Lon, p.s2Lat, p.radiusMetres, WIN_MS), Q8_WIN_STORE) + .to(Q8_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Long())); + } + + if (only == null || only.equals("Q9-windowed")) { + addStore(builder, Q9_WIN_STORE, Serdes.Long(), Serdes.String()); + tripsK0.process(() -> new Q9WindowedProcessor(Q9_WIN_STORE, p.xId, p.yId, WIN_MS), Q9_WIN_STORE) + .to(Q9_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Double())); + } + + if (only == null || only.equals("Q1-snapshot")) { + addStore(builder, Q1_SNAP_STORE, Serdes.Integer(), Serdes.Long()); + tripsK0.process(() -> new Q1SnapshotProcessor(Q1_SNAP_STORE, p.snapshotTickMillis), Q1_SNAP_STORE) + .to(Q1_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Integer())); + } + + if (only == null || only.equals("Q2-snapshot")) { + addStore(builder, Q2_SNAP_STORE, Serdes.Integer(), Serdes.String()); + tripsK0.process(() -> new Q2SnapshotProcessor(Q2_SNAP_STORE, p.targetId, p.snapshotTickMillis), Q2_SNAP_STORE) + .to(Q2_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + } + + if (only == null || only.equals("Q3-snapshot")) { + addStore(builder, Q3_SNAP_STORE, Serdes.Integer(), Serdes.String()); + tripsK0.process(() -> new Q3SnapshotProcessor(Q3_SNAP_STORE, p.pLon, p.pLat, p.radiusMetres, p.snapshotTickMillis), Q3_SNAP_STORE) + .to(Q3_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Integer())); + } + + if (only == null || only.equals("Q4-snapshot")) { + addStore(builder, Q4_SNAP_WAS_INSIDE_STORE, Serdes.Integer(), Serdes.Boolean()); + addStore(builder, Q4_SNAP_ENTRIES_STORE, Serdes.Integer(), Serdes.String()); + tripsK0.process(() -> new Q4SnapshotProcessor(Q4_SNAP_WAS_INSIDE_STORE, Q4_SNAP_ENTRIES_STORE, p.xmin, p.ymin, p.xmax, p.ymax, p.snapshotTickMillis), Q4_SNAP_WAS_INSIDE_STORE, Q4_SNAP_ENTRIES_STORE) + .to(Q4_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + } + + if (only == null || only.equals("Q5-snapshot")) { + addStore(builder, Q5_SNAP_STORE, Serdes.Integer(), Serdes.String()); + tripsK0.process(() -> new Q5SnapshotProcessor(Q5_SNAP_STORE, p.pLon, p.pLat, p.radiusMetres, p.dMeetMetres, p.snapshotTickMillis), Q5_SNAP_STORE) + .to(Q5_SNAPSHOT_OUTPUT, Produced.with(Serdes.String(), Serdes.Double())); + } + + if (only == null || only.equals("Q6-snapshot")) { + addStore(builder, Q6_SNAP_STORE, Serdes.Integer(), Serdes.String()); + tripsK0.process(() -> new Q6SnapshotProcessor(Q6_SNAP_STORE, p.snapshotTickMillis), Q6_SNAP_STORE) + .to(Q6_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + } + + if (only == null || only.equals("Q7-snapshot")) { + addStore(builder, Q7_SNAP_STORE, Serdes.Integer(), Serdes.Long()); + tripsK0.process(() -> new Q7SnapshotProcessor(Q7_SNAP_STORE, p.pois, p.snapshotTickMillis), Q7_SNAP_STORE) + .to(Q7_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + } + + if (only == null || only.equals("Q8-snapshot")) { + addStore(builder, Q8_SNAP_STORE, Serdes.Integer(), Serdes.String()); + tripsK0.process(() -> new Q8SnapshotProcessor(Q8_SNAP_STORE, p.s1Lon, p.s1Lat, p.s2Lon, p.s2Lat, p.radiusMetres, p.snapshotTickMillis), Q8_SNAP_STORE) + .to(Q8_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Integer())); + } + + if (only == null || only.equals("Q9-snapshot")) { + addStore(builder, Q9_SNAP_STORE, Serdes.Integer(), Serdes.String()); + tripsK0.process(() -> new Q9SnapshotProcessor(Q9_SNAP_STORE, p.xId, p.yId, p.snapshotTickMillis), Q9_SNAP_STORE) + .to(Q9_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Double())); + } return builder.build(); } diff --git a/kafka-streams-app/src/test/java/berlinmod/EmbeddedBrokerBenchmark.java b/kafka-streams-app/src/test/java/berlinmod/EmbeddedBrokerBenchmark.java new file mode 100644 index 0000000..0b9738c --- /dev/null +++ b/kafka-streams-app/src/test/java/berlinmod/EmbeddedBrokerBenchmark.java @@ -0,0 +1,335 @@ +/***************************************************************************** + * + * This MobilityDB code is provided under The PostgreSQL License. + * Copyright (c) 2020-2026, Université libre de Bruxelles and MobilityDB + * contributors + * + * Permission to use, copy, modify, and distribute this software and its + * documentation for any purpose, without fee, and without a written + * agreement is hereby granted, provided that the above copyright notice and + * this paragraph and the following two paragraphs appear in all copies. + * + * IN NO EVENT SHALL UNIVERSITE LIBRE DE BRUXELLES BE LIABLE TO ANY PARTY FOR + * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING + * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, + * EVEN IF UNIVERSITE LIBRE DE BRUXELLES HAS BEEN ADVISED OF THE POSSIBILITY + * OF SUCH DAMAGE. + * + * UNIVERSITE LIBRE DE BRUXELLES SPECIFICALLY DISCLAIMS ANY WARRANTIES, + * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY + * AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS ON + * AN "AS IS" BASIS, AND UNIVERSITE LIBRE DE BRUXELLES HAS NO OBLIGATIONS TO + * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS. + * + *****************************************************************************/ + +package berlinmod; + +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.Metric; +import org.apache.kafka.common.MetricName; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster; + +import java.lang.reflect.Field; +import java.nio.file.Files; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.TreeMap; + +/** + * Runtime throughput benchmark for the BerlinMOD-9 × 3-form streaming matrix on + * Kafka Streams against a real, in-process Kafka broker. + * + *

This is the Flink-comparable counterpart of {@link BerlinMODBenchmark}: the + * corpus is produced once into the {@link BerlinMODTopology#INPUT_TOPIC} topic of + * an {@link EmbeddedKafkaCluster} (a genuine {@code KafkaServer} reachable over + * the loopback network), and each cell runs as its own {@link KafkaStreams} + * application — the analog of MobilityFlink's per-cell jobs — consuming from the + * shared input topic and writing to the cell's output topic. The spatial + * predicates evaluate through MEOS (see {@link MEOSBridge}); the corpus and the + * per-query parameters are corpus-derived via {@link BerlinMODCorpus}, exactly as + * in MobilityFlink. + * + *

Throughput is the corpus size divided by the wall-clock from streams start + * until the application has consumed the whole input topic (its committed offset + * reaches the input end offset) and its output has gone idle. Each consumed + * record runs the cell's MEOS predicate, so this is the steady-state per-event + * processing rate, directly comparable to the MobilityFlink figures. The trailing + * idle settling time is excluded from the wall. + * + *

Run from {@code kafka-streams-app/} with an extended libmeos on the loader + * path and the test classpath (it carries the embedded broker): + * + * 

+ *   CP=$(mvn -q dependency:build-classpath -DincludeScope=test -Dmdep.outputFile=/dev/stdout | tail -1)
+ *   LD_LIBRARY_PATH=<libmeos-dir> java -cp target/classes:target/test-classes:jar/JMEOS.jar:$CP \
+ *     berlinmod.EmbeddedBrokerBenchmark --csv <berlinmod_instants.csv> [--max N] [--only Q3-continuous]
+ *
+ */ +public final class EmbeddedBrokerBenchmark { + + private EmbeddedBrokerBenchmark() { /* utility */ } + + /** Settle window with no new output before a cell is declared finished. */ + private static final long QUIET_MS = 2_000L; + /** Hard per-cell ceiling so a stuck cell cannot hang the whole run. */ + private static final long CELL_TIMEOUT_MS = 300_000L; + + public static void main(String[] args) throws Exception { + String csv = null, only = null; + int maxRows = 0, vehicles = 50, events = 600; + for (int i = 0; i < args.length; i++) { + switch (args[i]) { + case "--csv": csv = args[++i]; break; + case "--max": maxRows = Integer.parseInt(args[++i]); break; + case "--vehicles": vehicles = Integer.parseInt(args[++i]); break; + case "--events": events = Integer.parseInt(args[++i]); break; + case "--only": only = args[++i]; break; + default: break; + } + } + + List corpus = csv != null + ? BerlinMODCorpus.fromInstantsCsv(csv, maxRows) + : BerlinMODCorpus.synthetic(vehicles, events); + int n = corpus.size(); + long maxTs = corpus.stream().mapToLong(BerlinMODTrip::getTimestamp).max().orElse(0L); + BerlinMODCorpus.Params p = BerlinMODCorpus.derive(corpus); + System.out.printf("Corpus: %s, %d events; window=%ds tick=%dms%n", + csv != null ? "real BerlinMOD instants" : "synthetic", n, p.windowSeconds, p.snapshotTickMillis); + + // (cellName -> outputTopic) enumerated from the topology's *_OUTPUT constants. + TreeMap cells = new TreeMap<>(); + for (Field f : BerlinMODTopology.class.getFields()) { + if (f.getName().endsWith("_OUTPUT") && f.getType() == String.class) { + cells.put(cellName(f.getName()), (String) f.get(null)); + } + } + + // Each cell runs against its OWN fresh embedded broker — the true analog of + // MobilityFlink's independent per-cell jobs. A single broker shared across all + // 27 cells destabilises after a handful of KafkaStreams lifecycles on one node; + // a fresh broker per cell keeps every measurement isolated and reproducible. + List rows = new ArrayList<>(); + for (var e : cells.entrySet()) { + String cell = e.getKey(), topic = e.getValue(); + if (only != null && !cell.equals(only)) { + continue; + } + EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(1, brokerProps()); + cluster.start(); + try { + cluster.createTopic(BerlinMODTopology.INPUT_TOPIC, 1, 1); + long inputEnd = produceCorpus(cluster.bootstrapServers(), corpus, maxTs); + cluster.createTopic(topic, 1, 1); + String[] r = runCell(cluster.bootstrapServers(), cell, topic, p, n, inputEnd); + rows.add(r); + System.out.printf(" %-14s out=%-8s %7s ms %s ev/s%n", cell, r[2], r[3], r[4]); + } finally { + cluster.stop(); + } + } + + System.out.println(); + System.out.println("| Cell | Events in | Output rows | Wall (ms) | Throughput (ev/s) |"); + System.out.println("|---|---:|---:|---:|---:|"); + for (String[] r : rows) { + System.out.printf("| %s | %s | %s | %s | %s |%n", r[0], r[1], r[2], r[3], r[4]); + } + } + + /** Single-broker, single-ISR config for the embedded cluster. */ + private static Properties brokerProps() { + Properties bp = new Properties(); + bp.put("auto.create.topics.enable", "true"); + bp.put("offsets.topic.replication.factor", "1"); + bp.put("transaction.state.log.replication.factor", "1"); + bp.put("transaction.state.log.min.isr", "1"); + bp.put("group.initial.rebalance.delay.ms", "0"); + return bp; + } + + /** Produce the corpus plus two future-timestamped flush sentinels; returns the input end offset. */ + private static long produceCorpus(String bootstrap, List corpus, long maxTs) { + Properties pp = new Properties(); + pp.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap); + pp.put(ProducerConfig.ACKS_CONFIG, "all"); + pp.put(ProducerConfig.LINGER_MS_CONFIG, "5"); + BerlinMODTripSerde tripSerde = new BerlinMODTripSerde(); + long count = 0; + try (KafkaProducer producer = + new KafkaProducer<>(pp, new IntegerSerializer(), new ByteArraySerializer())) { + for (BerlinMODTrip trip : corpus) { + byte[] v = tripSerde.serializer().serialize(BerlinMODTopology.INPUT_TOPIC, trip); + producer.send(new ProducerRecord<>(BerlinMODTopology.INPUT_TOPIC, null, + trip.getTimestamp(), trip.getVehicleId(), v)); + count++; + } + // Two flush sentinels advance event time so windowed/snapshot cells close. + for (long delta : new long[]{3_600_000L, 7_200_000L}) { + BerlinMODTrip s = new BerlinMODTrip(-1, maxTs + delta, 0.0, 0.0); + byte[] v = tripSerde.serializer().serialize(BerlinMODTopology.INPUT_TOPIC, s); + producer.send(new ProducerRecord<>(BerlinMODTopology.INPUT_TOPIC, null, + maxTs + delta, -1, v)); + count++; + } + producer.flush(); + } + return count; + } + + /** Run one cell as a real KafkaStreams app; returns {cell, eventsIn, outputRows, wallMs, throughput}. */ + private static String[] runCell(String bootstrap, String cell, String outputTopic, + BerlinMODCorpus.Params p, int n, long inputEnd) throws Exception { + String appId = "berlinmod-bench-" + cell.toLowerCase().replace('-', '_'); + Properties props = new Properties(); + props.put(StreamsConfig.APPLICATION_ID_CONFIG, appId); + props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap); + props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass()); + props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArray().getClass()); + props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1); + props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 200); + props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 1); + props.put(StreamsConfig.STATE_DIR_CONFIG, + Files.createTempDirectory("bench-" + appId).toString()); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + props.put(StreamsConfig.consumerPrefix(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), "earliest"); + props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0); + + Properties adminProps = new Properties(); + adminProps.put("bootstrap.servers", bootstrap); + + long outputRows; + long wallMs; + long consumedAtEnd; + try (Admin admin = Admin.create(adminProps); + KafkaConsumer outConsumer = outputConsumer(bootstrap, outputTopic)) { + + KafkaStreams streams = new KafkaStreams(BerlinMODTopology.buildCell(p, cell), props); + long t0 = System.nanoTime(); + streams.start(); + + long outCount = 0; + long tInputDrained = -1; + long tLastOutput = t0; + long deadline = System.currentTimeMillis() + CELL_TIMEOUT_MS; + // Progress is read straight from the running app's own consumer metric + // (records-consumed-total for the input topic), so completion detection + // is independent of the group coordinator — which becomes unreliable on a + // single broker after many app lifecycles. + while (true) { + ConsumerRecords recs = outConsumer.poll(Duration.ofMillis(100)); + if (!recs.isEmpty()) { + outCount += recs.count(); + tLastOutput = System.nanoTime(); + } + if (tInputDrained < 0 && inputConsumed(streams) >= inputEnd) { + tInputDrained = System.nanoTime(); + } + long progress = outCount > 0 ? Math.max(tInputDrained, tLastOutput) : tInputDrained; + boolean idle = tInputDrained > 0 && (System.nanoTime() - progress) / 1_000_000L >= QUIET_MS; + if (idle) { + wallMs = (progress - t0) / 1_000_000L; + break; + } + if (System.currentTimeMillis() > deadline) { + wallMs = (System.nanoTime() - t0) / 1_000_000L; + System.out.printf(" [%s] timed out after %d ms (consumed=%d/%d)%n", + cell, wallMs, inputConsumed(streams), inputEnd); + break; + } + } + outputRows = outCount; + consumedAtEnd = inputConsumed(streams); + streams.close(Duration.ofSeconds(30)); + streams.cleanUp(); + cleanup(admin, appId, outputTopic); + } + + // Throughput uses the events actually consumed (the input topic high + // watermark, == inputEnd once drained), divided by the measured wall. + long events = Math.min(consumedAtEnd, inputEnd); + double tput = wallMs > 0 ? events / (wallMs / 1000.0) : 0; + return new String[]{cell, String.valueOf(n), String.valueOf(outputRows), + String.valueOf(wallMs), String.format("%,.0f", tput)}; + } + + /** Records the running app's consumer has read from the input topic, via its own metrics. */ + private static long inputConsumed(KafkaStreams streams) { + double sum = 0; + for (Metric m : streams.metrics().values()) { + MetricName name = m.metricName(); + if ("records-consumed-total".equals(name.name()) + && BerlinMODTopology.INPUT_TOPIC.equals(name.tags().get("topic"))) { + Object v = m.metricValue(); + if (v instanceof Number) { + sum += ((Number) v).doubleValue(); + } + } + } + return (long) sum; + } + + /** Output consumer with an explicit partition assignment — no group, no rebalance, no coordinator load. */ + private static KafkaConsumer outputConsumer(String bootstrap, String outputTopic) { + Properties cp = new Properties(); + cp.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap); + cp.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); + KafkaConsumer c = + new KafkaConsumer<>(cp, new ByteArrayDeserializer(), new ByteArrayDeserializer()); + TopicPartition tp = new TopicPartition(outputTopic, 0); + c.assign(Collections.singletonList(tp)); + c.seekToBeginning(Collections.singletonList(tp)); + return c; + } + + /** Delete the cell's consumer group and internal topics so the coordinator stays lean across cells. */ + private static void cleanup(Admin admin, String appId, String outputTopic) { + try { + admin.deleteConsumerGroups(Collections.singletonList(appId)).all().get(); + } catch (Exception ignored) { + // group may already be gone + } + deleteInternalTopics(admin, appId); + } + + private static void deleteInternalTopics(Admin admin, String appId) { + try { + List internal = new ArrayList<>(); + for (String t : admin.listTopics().names().get()) { + if (t.startsWith(appId + "-")) { + internal.add(t); + } + } + if (!internal.isEmpty()) { + admin.deleteTopics(internal).all().get(); + } + } catch (Exception ignored) { + // best-effort cleanup; the cluster is torn down at the end anyway + } + } + + /** Q3_CONTINUOUS_OUTPUT -> Q3-continuous */ + private static String cellName(String field) { + String s = field.substring(0, field.length() - "_OUTPUT".length()); + int us = s.indexOf('_'); + return s.substring(0, us) + "-" + s.substring(us + 1).toLowerCase(); + } +}