build(flink): migrate to Apache Flink 2.0 (stacks on #31)

flink-processor/docs/benchmark-results.md

@@ -0,0 +1,89 @@
+# BerlinMOD streaming-matrix throughput
+
+Throughput of the BerlinMOD-9 × 3-form streaming matrix (9 queries ×
+{continuous, windowed, snapshot} = 27 cells) on the Flink local mini-cluster
+over the real BerlinMOD instants corpus. The spatial predicates evaluate through
+MEOS: within-distance through `edwithin_tgeo_geo`, region containment through
+`eintersects_tgeo_geo`, and distances through `geog_distance` (see
+[`MEOSBridge`](../src/main/java/berlinmod/MEOSBridge.java)).
+
+## Method
+
+The corpus is the BerlinMOD `berlinmod_instants.csv` produced by the BerlinMOD
+generator — 216 075 instants, 5 vehicles, over ~11 days. Instants are stored in
+EPSG:3857 and reprojected to EPSG:4326 through MEOS `geo_transform` at load (see
+[`BerlinMODCorpus`](../src/main/java/berlinmod/BerlinMODCorpus.java)); the
+per-query parameters (point `P` = corpus centroid, region box, road segment,
+points of interest, target vehicle ids) and the window/tick granularity are
+derived from the corpus so each spatial cell is selective and the matrix
+produces a comparable number of windows. Each cell runs as its own Flink job
+terminated by a counting sink; throughput is input events ÷ wall-clock and
+`output rows` is the sink cardinality. Parallelism 1, Flink 1.16, Java 21,
+16-core x86-64 Linux; libmeos built `-DMEOS=ON -DCBUFFER=ON -DNPOINT=ON
+-DPOSE=ON -DRGEO=ON`.
+
+Run from `flink-processor/`:
+
+```
+LD_LIBRARY_PATH=<libmeos-dir> 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 \
+  --add-opens=java.base/java.io=ALL-UNNAMED \
+  --add-opens=java.base/java.time=ALL-UNNAMED \
+  -cp target/classes:jar/JMEOS.jar:<deps> \
+  berlinmod.BerlinMODBenchmark --csv <berlinmod_instants.csv>
+```
+
+## Results — real BerlinMOD instants (216 075 events)
+
+| Cell | Events in | Output rows | Wall (ms) | Throughput (ev/s) |
+|---|---:|---:|---:|---:|
+| Q1-continuous | 216075 | 5 | 2508 | 86,154 |
+| Q1-windowed | 216075 | 86 | 1294 | 166,982 |
+| Q1-snapshot | 216075 | 274 | 1056 | 204,616 |
+| Q2-continuous | 216075 | 61170 | 1074 | 201,187 |
+| Q2-windowed | 216075 | 50 | 1027 | 210,394 |
+| Q2-snapshot | 216075 | 71 | 985 | 219,365 |
+| Q3-continuous | 216075 | 216075 | 2928 | 73,796 |
+| Q3-windowed | 216075 | 86 | 2507 | 86,189 |
+| Q3-snapshot | 216075 | 0 | 926 | 233,342 |
+| Q4-continuous | 216075 | 62 | 3254 | 66,403 |
+| Q4-windowed | 216075 | 98 | 3234 | 66,814 |
+| Q4-snapshot | 216075 | 1944 | 3223 | 67,042 |
+| Q5-continuous | 216075 | 73063 | 9161 | 23,586 |
+| Q5-windowed | 216075 | 6 | 954 | 226,494 |
+| Q5-snapshot | 216075 | 0 | 915 | 236,148 |
+| Q6-continuous | 216075 | 216075 | 2382 | 90,712 |
+| Q6-windowed | 216075 | 203 | 2637 | 81,940 |
+| Q6-snapshot | 216075 | 274 | 2214 | 97,595 |
+| Q7-continuous | 216075 | 5 | 3973 | 54,386 |
+| Q7-windowed | 216075 | 53 | 5004 | 43,180 |
+| Q7-snapshot | 216075 | 288 | 3931 | 54,967 |
+| Q8-continuous | 216075 | 216075 | 2883 | 74,948 |
+| Q8-windowed | 216075 | 86 | 2864 | 75,445 |
+| Q8-snapshot | 216075 | 126 | 928 | 232,839 |
+| Q9-continuous | 216075 | 107870 | 1858 | 116,294 |
+| Q9-windowed | 216075 | 22 | 924 | 233,847 |
+| Q9-snapshot | 216075 | 95 | 992 | 217,818 |
+
+## Parity — streaming continuous form ≡ batch MEOS predicate
+
+The continuous form emits `predicate(event)` for every event, so it is checked
+event-for-event against a batch pass over the same corpus through the same
+`MEOSBridge` call ([`BerlinMODParity`](../src/main/java/berlinmod/BerlinMODParity.java)).
+Both spatial-membership queries match exactly.
+
+| Query | Events | Streaming-true | Batch-true | Mismatches | Parity |
+|---|---:|---:|---:|---:|---|
+| Q3 (within `d` of `P`) | 216075 | 56086 | 56086 | 0 | exact |
+| Q8 (within `d` of segment) | 216075 | 118498 | 118498 | 0 | exact |
+
+## Characteristics
+
+Q5-continuous enumerates every meeting pair across all vehicles on each event
+(O(V²) per event, keyed to a single subtask); it is the lowest throughput of the
+matrix. The snapshot form is a sampled form — it evaluates each vehicle's
+last-known position at tick instants — so a within-`P` snapshot can be empty when
+no vehicle is within `d` of `P` at a tick boundary even though the continuous
+form reports near-`P` events between boundaries.

flink-processor/pom.xml

@@ -12,7 +12,14 @@
         <maven.compiler.source>21</maven.compiler.source>
         <maven.compiler.target>21</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <flink.version>1.16.0</flink.version>
+        <flink.version>2.0.0</flink.version>
+        <!-- The Kafka connector is externalized (own release line) since Flink 1.17;
+             it no longer follows ${flink.version}. This is its Flink-2.0 build. -->
+        <flink.kafka.version>4.0.1-2.0</flink.kafka.version>
+        <!-- Apache Sedona has no Flink-2.0 build (latest targets Flink 1.19), so the
+             sedona/** comparison queries are excluded by default and re-enabled with
+             the "sedona" profile once an upstream Sedona/Flink-2.0 release exists. -->
+        <sedona.source.excludes>**/sedona/**</sedona.source.excludes>
         <jackson.version>2.13.4</jackson.version>
         <junit.jupiter.version>5.8.2</junit.jupiter.version>
         <kafka.version>3.2.0</kafka.version>
@@ -42,20 +49,10 @@
             <artifactId>flink-core</artifactId>
             <version>${flink.version}</version>
         </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-java</artifactId>
-            <version>${flink.version}</version>
-        </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-kafka</artifactId>
-            <version>${flink.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-jdbc</artifactId>
-            <version>${flink.version}</version>
+            <version>${flink.kafka.version}</version>
         </dependency>
 
         <!-- ======================================================= -->
@@ -90,19 +87,10 @@
         </dependency>
         -->
 
-        <!-- =================================================================================   -->
-        <!-- Apache Sedona: https://sedona.apache.org/latest/setup/maven-coordinates/            -->
-        <!-- =================================================================================   -->
-        <dependency>
-            <groupId>org.apache.sedona</groupId>
-            <artifactId>sedona-flink-shaded_2.12</artifactId>
-            <version>1.8.1</version>
-        </dependency>
-        <dependency>
-            <groupId>org.datasyslab</groupId>
-            <artifactId>geotools-wrapper</artifactId>
-            <version>1.8.1-33.1</version>
-        </dependency>
+        <!-- Apache Sedona (sedona-flink-shaded + geotools-wrapper) and the sedona/**
+             comparison queries live in the "sedona" profile — see the bottom of this
+             POM. They are omitted from the default build because Sedona has no Flink-2.0
+             release yet. -->
 
         <!-- ======================================================= -->
         <!-- JTS: 2D geometries (Point, Polygon, LineString, …)      -->
@@ -225,6 +213,12 @@
                 <version>3.12.1</version>
                 <configuration>
                     <release>${target.java.version}</release>
+                    <excludes>
+                        <exclude>${sedona.source.excludes}</exclude>
+                    </excludes>
+                    <testExcludes>
+                        <testExclude>${sedona.source.excludes}</testExclude>
+                    </testExcludes>
                 </configuration>
             </plugin>
             <plugin>
@@ -290,6 +284,31 @@
         </plugins>
     </build>
 
+    <profiles>
+        <!-- Apache Sedona comparison queries (sedona/**). Sedona has no Flink-2.0
+             release yet (latest targets Flink 1.19), so these are excluded from the
+             default build. Enable with -Psedona once an upstream Sedona/Flink-2.0
+             artifact exists; this profile re-adds the sources and the dependencies. -->
+        <profile>
+            <id>sedona</id>
+            <properties>
+                <sedona.source.excludes>nothing-excluded/**</sedona.source.excludes>
+            </properties>
+            <dependencies>
+                <dependency>
+                    <groupId>org.apache.sedona</groupId>
+                    <artifactId>sedona-flink-shaded_2.12</artifactId>
+                    <version>1.9.0</version>
+                </dependency>
+                <dependency>
+                    <groupId>org.datasyslab</groupId>
+                    <artifactId>geotools-wrapper</artifactId>
+                    <version>1.8.1-33.1</version>
+                </dependency>
+            </dependencies>
+        </profile>
+    </profiles>
+
     <!-- Optional extended temporal-type families, mirroring the MobilityDB/MEOS
          CMake build flags. Family inclusion is selected at build time with the
          same uppercase flag names and ON|OFF (also 1|0) values as MEOS:

flink-processor/src/main/java/aisdata/Main.java

@@ -42,7 +42,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
-import org.apache.flink.streaming.api.windowing.time.Time;
 
 import functions.*;
 import functions.GeneratedFunctions;
@@ -104,8 +103,8 @@ public static void main(String[] args) throws Exception {
             DataStream<TGeomPointSeq> trajectories = source
                 .map(new AISDataToTuple4MapFunction())
                 .keyBy(tuple -> tuple.f0) 
-                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
-                //.allowedLateness(Time.seconds(0))
+                .window(TumblingEventTimeWindows.of(Duration.ofSeconds(10)))
+                //.allowedLateness(Duration.ofSeconds(0))
                 .process(new TrajectoryWindowFunction());
 
             //trajectories.print();

flink-processor/src/main/java/aisdata/Query1_Main.java

@@ -8,14 +8,13 @@
 import java.util.Properties;
 
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.api.common.functions.OpenContext;
 import org.apache.flink.connector.kafka.source.KafkaSource;
 import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
 import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
-import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.util.Collector;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -61,7 +60,7 @@
  *       threshold is set to 500 m for the AIS dataset (the paper uses 20 m for SNCB trains,
  *       see {@link #ALERT_DISTANCE_METERS}).</li>
  *   <li><b>Line 3 (10-second tumbling window)</b>: Implemented with
- *       {@code TumblingEventTimeWindows.of(Time.seconds(10))}, using the AIS message timestamp
+ *       {@code TumblingEventTimeWindows.of(Duration.ofSeconds(10))}, using the AIS message timestamp
  *       as event time.</li>
  *   <li><b>Line 4 (print sink)</b>: Alerts are emitted via {@code .print()} and also logged
  *       at WARN level by {@link HighRiskZoneWindowFunction}.</li>
@@ -218,7 +217,7 @@ public static void main(String[] args) throws Exception {
             //   print()                 → paper Line 4: equivalent to PrintSinkDescriptor.
             source
                     .keyBy(AISData::getMmsi)
-                    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
+                    .window(TumblingEventTimeWindows.of(Duration.ofSeconds(10)))
                     .process(new HighRiskZoneWindowFunction(
                             HIGH_RISK_ZONES_WKT, ALERT_DISTANCE_METERS))
                     .print();
@@ -273,7 +272,7 @@ public HighRiskZoneWindowFunction(String[] zoneWkt, double distanceMeters) {
 
         /** Initialises the MEOS library and the hazard zones for this operator instance. */
         @Override
-        public void open(Configuration parameters) throws Exception {
+        public void open(OpenContext parameters) throws Exception {
             super.open(parameters);
             errorHandler = new error_handler();
             functions.meos_initialize_timezone("UTC");

flink-processor/src/main/java/aisdata/Query2_Main.java

@@ -10,14 +10,13 @@
 import java.util.Properties;
 
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.api.common.functions.OpenContext;
 import org.apache.flink.connector.kafka.source.KafkaSource;
 import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
 import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
-import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.util.Collector;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -69,7 +68,7 @@
  *       skipped. This is the inverse of Query 1's {@code edwithin_tgeo_geo}: here we exclude
  *       points that are inside a zone rather than alerting on proximity.</li>
  *   <li><b>Line 3 (sliding window 10s / 10ms)</b>: Implemented with
- *       {@code SlidingEventTimeWindows.of(Time.seconds(10), Time.milliseconds(10))}. A 10ms
+ *       {@code SlidingEventTimeWindows.of(Duration.ofSeconds(10), Duration.ofMillis(10))}. A 10ms
  *       step produces approximately 1000 overlapping windows per second.</li>
  *   <li><b>Line 4 (variation(FA), variation(FF))</b>: The {@code variation} operator computes
  *       statistical variance (E[X²] − E[X]²) over the window. MEOS does not expose a
@@ -214,7 +213,7 @@ public static void main(String[] args) throws Exception {
             //   print()                   → output brake anomaly alerts.
             source
                     .keyBy(AISData::getMmsi)
-                    .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.milliseconds(10)))
+                    .window(SlidingEventTimeWindows.of(Duration.ofSeconds(10), Duration.ofMillis(10)))
                     .process(new BrakeMonitoringWindowFunction(
                             MAINTENANCE_AREAS_WKT, VAR_FA_THRESHOLD, VAR_FF_THRESHOLD))
                     .print();
@@ -306,7 +305,7 @@ public BrakeMonitoringWindowFunction(
 
         /** Initialises the MEOS library for this operator instance. */
         @Override
-        public void open(Configuration parameters) throws Exception {
+        public void open(OpenContext parameters) throws Exception {
             super.open(parameters);
             errorHandler = new error_handler();
             functions.meos_initialize_timezone("UTC");

flink-processor/src/main/java/aisdata/Query3_Main.java

@@ -10,14 +10,13 @@
 import java.util.Properties;
 
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.api.common.functions.OpenContext;
 import org.apache.flink.connector.kafka.source.KafkaSource;
 import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
 import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
-import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.util.Collector;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -54,7 +53,7 @@
  * <p><b>Mapping to this implementation:</b>
  * <ul>
  *   <li><b>Line 2 (sliding window 10s / 10ms)</b>: Implemented with
- *       {@code SlidingEventTimeWindows.of(Time.seconds(10), Time.milliseconds(10))},
+ *       {@code SlidingEventTimeWindows.of(Duration.ofSeconds(10), Duration.ofMillis(10))},
  *       identical to Query 2. Each window collects up to 10 seconds of GPS points for one
  *       vessel, and a new window opens every 10 milliseconds, producing overlapping snapshots
  *       of the vessel's recent trajectory.</li>
@@ -143,7 +142,7 @@ public static void main(String[] args) throws Exception {
             //   print()                   → outputs each trajectory as a human-readable EWKT string.
             source
                     .keyBy(AISData::getMmsi)
-                    .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.milliseconds(10)))
+                    .window(SlidingEventTimeWindows.of(Duration.ofSeconds(10), Duration.ofMillis(10)))
                     .process(new TrajectoryCreationWindowFunction())
                     .print();
 
@@ -188,7 +187,7 @@ public static class TrajectoryCreationWindowFunction
 
         /** Initialises the MEOS library for this operator instance. */
         @Override
-        public void open(Configuration parameters) throws Exception {
+        public void open(OpenContext parameters) throws Exception {
             super.open(parameters);
             errorHandler = new error_handler();
             functions.meos_initialize_timezone("UTC");