fix(flink): run the engine on flink

.github/workflows/ci.yml

@@ -76,6 +76,7 @@ jobs:
           connectors-gcs/mimaReportBinaryIssues
           connectors-jdbc/mimaReportBinaryIssues
           engines-spark/mimaReportBinaryIssues
+          engines-flink/mimaReportBinaryIssues
           quality-deequ/mimaReportBinaryIssues
           infrastructure/mimaReportBinaryIssues
           contracts-sdk/mimaReportBinaryIssues
@@ -228,6 +229,29 @@ jobs:
       - name: Run engines-spark tests
         run: sbt "-Dsbt.log.noformat=true" engines-spark/test
 
+  flink-engine:
+    name: Flink engine tests
+    needs: quick-checks
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    if: "!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, '[skip release]')"
+    steps:
+      - uses: actions/checkout@v6
+
+      - uses: sbt/setup-sbt@v1
+
+      - uses: actions/setup-java@v5
+        with:
+          distribution: temurin
+          java-version: 17
+          cache: sbt
+
+      # These tests start a local Flink MiniCluster, so a failure here is a Flink failure rather than a
+      # compile error. The root `testQuick` runs them too; this job is what names them.
+      - name: Run engines-flink tests
+        run: sbt "-Dsbt.log.noformat=true" engines-flink/test
+
   spark-it:
     name: Spark integration tests
     needs: [test]

build.sbt

@@ -13,10 +13,7 @@
 ThisBuild / dynverSonatypeSnapshots := true
 // Default Scala stays 2.13 for most modules; Spark/Deequ modules are handled pragmatically via deps.
 ThisBuild / scalaVersion := Dependencies.Versions.scala213
-// Cross-compile defaults: build-level + commands iterate over Scala 2.13 and 3 for speed and stability.
-// Module-specific overrides (e.g., enginesFlink) can still target 2.12 explicitly.
-// Global cross-build: keep 2.13 only for stability and speed.
-// Module overrides handle 2.12 (Flink) and Scala 3 (experimental) explicitly.
+// Global cross-build: keep 2.13 only for stability and speed. `core` additionally builds on Scala 3.
 ThisBuild / crossScalaVersions := Seq(
   Dependencies.Versions.scala213
 )
@@ -41,7 +38,7 @@
 ThisBuild / coverageHighlighting := true
 
 // ===== REPOSITORY RESOLVERS =====
 resolvers ++= Resolver.sonatypeOssRepos("public") ++ Seq(
   Resolver.mavenCentral,
   "Confluent" at "https://packages.confluent.io/maven/",
   "Apache Releases" at "https://repository.apache.org/content/repositories/releases/",
@@ -171,6 +168,7 @@
     connectorsGcs,
     connectorsJdbc,
     enginesSpark,
+    enginesFlink,
     qualityDeequ, // Removed empty quality module per v1.0-2 plan
     examples,
     compileFailTests,
@@ -199,7 +197,7 @@
 lazy val core = moduleProject("core")
   .settings(
     description := "Core abstractions and custom type system",
-    // Core is the only module published for both. Flink-specific modules handle 2.12 separately.
+    // Core is the only module published for both.
     crossScalaVersions := Seq(Dependencies.Versions.scala213, Dependencies.Versions.scala3),
     libraryDependencies ++= Dependencies.forModule("core"),
     // Minimal, justified excludes only
@@ -317,10 +315,25 @@
 lazy val enginesFlink = moduleProject("engines-flink")
   .dependsOn(core, connectors, enginesSpark % "test->compile")
   .settings(
-    description        := "Apache Flink execution engine",
-    // Flink Scala API is 2.12-only; avoid invalid 2.13 cross build
-    crossScalaVersions := Seq(Dependencies.Versions.scala212),
+    description := "Apache Flink execution engine",
     libraryDependencies ++= Dependencies.forModule("engines-flink"),
+    // Flink serializes records with Kryo, which reflects into `java.base`. On Java 17 and later that is
+    // closed by default, and the failure arrives as an `InaccessibleObjectException` from inside a running
+    // job rather than at startup. These are the opens Flink's own scripts set for the same reason.
+    Test / javaOptions ++= Seq(
+      "--add-opens=java.base/java.lang=ALL-UNNAMED",
+      "--add-opens=java.base/java.lang.invoke=ALL-UNNAMED",
+      "--add-opens=java.base/java.io=ALL-UNNAMED",
+      "--add-opens=java.base/java.net=ALL-UNNAMED",
+      "--add-opens=java.base/java.nio=ALL-UNNAMED",
+      "--add-opens=java.base/java.text=ALL-UNNAMED",
+      "--add-opens=java.base/java.time=ALL-UNNAMED",
+      "--add-opens=java.base/java.util=ALL-UNNAMED",
+      "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED",
+      "--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED",
+      "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED",
+    ),
+    Test / fork := true,
   )
   .settings(mimaSettings("engines-flink"): _*)
 
@@ -537,14 +550,15 @@
 import sbtunidoc.ScalaUnidocPlugin
 import sbtunidoc.ScalaUnidocPlugin.autoImport._
 
-// Only aggregate Scala 2.13 modules (Flink is 2.12-only). This keeps unidoc stable.
+// Only aggregate the Scala 2.13 modules. This keeps unidoc stable.
 lazy val unidocProjects = Seq(
   core,
   contracts,
   connectors,
   connectorsGcs,
   connectorsJdbc,
   enginesSpark,
+  enginesFlink,
   qualityDeequ,
   infrastructure,
   examples,

docs/plan/release-criteria.md

@@ -7,7 +7,7 @@ one place a readiness claim belongs.
 ## Build & Platform
 - CI runs on JDK 17 (Ubuntu 22.04), sbt 1.9+.
 - Scala 2.13 primary; Scala 3 sources compile in `core` (no Spark dependencies).
-- `engines-flink`: builds and resolves on a single axis, or the module is removed.
+- `engines-flink`: builds and resolves on Scala 2.13, and its tests run a Flink job rather than a delegation.
  - Spark IT job pinned to Java 17; println banned in production sources via CI guard.
 
 ## Contracts & Compile‑Time Safety

docs/plan/v1.0-readiness.md

@@ -19,20 +19,17 @@ change before the version number means anything.
   filter followed by a write wrote unfiltered rows and a violation in record 101 went unreported. The
   operations that read every record now carry a `DataDecoder` bound, which is what lets the engine push them
   into the frame.
-- `engines-flink` does not build. `build.sbt` pins it to Scala 2.12 for the Flink Scala API, but `core` and
-  `connectors` publish 2.13 and 3 only, so dependency resolution fails before compilation starts. No CI job
-  builds the module.
+- `engines-flink` builds and tests on Scala 2.13 against Flink 1.18. It used to be pinned to 2.12 for the
+  Flink Scala API, which `core` does not publish for, so resolution failed before compilation started; the
+  module now uses Flink's Java artifacts, which carry no Scala suffix. It was also missing from the root
+  aggregate, so no build or test command reached it. `engines-flink/test` runs on every CI build and starts a
+  local MiniCluster.
+- `read`, `write`, `filter`, `map`, `flatMap`, `union`, `validate` and `runQualityChecks` on Flink run as
+  batch jobs over every record. A Flink-backed dataset holds a plan rather than the records, so the driver
+  keeps at most 100 of them.
 
 ## Blocking gaps
 
-### The Flink engine runs no Flink
-`FlinkDataAlgebra` forwards every method to `InMemoryDataAlgebra`. Work happens on the driver heap and
-nothing is submitted to a cluster. The module also cannot be compiled (see above), so this is not covered by
-any test.
-
-Before v1.0, either implement the engine on the Flink DataStream or Table API and give it a 2.12 build that
-resolves, or remove the module and say Spark is the only engine.
-
 ### Connector surfaces that are not implemented
 Several connector paths exist as types without behaviour: the S3 read and write path on Spark raises a
 failure, and the S3 connector body is commented out. The capability set advertises `Read` and `Write` without
@@ -64,8 +61,29 @@ These are deliberate for v1.0 and recorded so nobody rediscovers them as bugs.
   pinned there: on an unsupported format Spark raises `DataProcessingError.ProcessingFailure` and the
   in-memory engine raises a plain `UnsupportedOperationException`. A caller that handles one does not handle
   the other.
-- The Flink spec named for parity checks delegation, which is all it can check while the engine forwards to
-  the in-memory one.
+- Flink parity is tested against the in-memory engine in
+  `modules/engines-flink/src/test/scala/com/flowforge/engines/flink/CrossEngineParitySpec.scala`, with the
+  same unsupported-format divergence pinned there.
+- `groupBy`, `join`, `sortBy`, `take`, `drop`, `profile`, the CDC operations and the table operations still
+  forward to the in-memory engine on Flink, so on a Flink-backed dataset they answer for at most the 100
+  sampled records. They are the same operations that run on the driver for Spark, for the same reason.
+- The Flink sink writes JSON and JSONL only. A dataflow sink cannot order a CSV header line ahead of the rows
+  it writes, and a headerless CSV is a file this engine cannot read back, so a CSV sink raises rather than
+  writing one. The sink runs at parallelism one so the output is the single file the caller named.
+- Both engines hand every row to the caller's `DataDecoder` as JSON, including rows read from a CSV file,
+  because JSON is the one representation the codecs and the engines both speak. A decoder that supports only
+  one other format therefore decodes nothing, and the dataset's `data` is empty while its count is not. The
+  decoder's `supportsFormat` is not consulted, so the caller gets no signal.
+- A record count on both engines counts the rows the source holds, not the records that decoded. A file with
+  rows the decoder rejects reports a count larger than the records anything can see, which inflates the
+  denominator `validate` scores against. The in-memory engine counts decoded records instead, so the three
+  engines agree only on a source every row of which decodes.
+- The 1000-message cap on `validate` truncates deterministically on Spark, which orders by message before
+  capping, and not on Flink, which has no cheap global sort over a `DataStream`. Which messages survive is
+  unspecified there. On both engines the cap cannot change `passed` or `score`.
+- The Flink engine runs batch jobs, not streaming ones. Every job sets `RuntimeExecutionMode.BATCH`, which is
+  what makes a keyed count report one final value per key, and which means no unbounded source can be
+  consumed. Every source it reads is a bounded file.
 - Four `TODO` markers remain in `core`: record-level validation and the Reader-based pipeline in
   `syntax/PipelineSyntax.scala`, and the environment fallback and per-source loading in
   `algebra/ConfigurationAlgebra.scala`. Each one returns something rather than failing, so a caller gets a