WindowAssessor

Reproduction artifact for the paper "A Unified Semantic Framework for Window Operators under Late and Out-of-Order Data" (Ahmed Awad). It runs the same windowed count query on five stream processing engines over small, controlled traces so their behavior under late and out-of-order data can be compared tuple-for-tuple.

What this work is about

Stream processing engines (SPEs) frequently disagree on how a window operator behaves once data arrives late or out of order: the same window specification can admit or drop different events, recompute (or not) after a late arrival, and report results in different ways. That undermines portability, correctness reasoning, and maintainability of streaming applications.

The paper builds a unified semantic framework for window operators that:

• separates the window lifecycle into maintenance (which events are admitted into which window instances, and when a window is evicted) versus reporting (when and how results are emitted — stateless, refinable, or retractable);
• distinguishes single-buffer operators (evict-and-compute on progress; one computation per window) from multi-buffer operators (retain window state; can recompute and re-report as late data arrives);
• maps these abstractions onto the concrete APIs and execution models of five widely used SPEs, exposing exactly where and why their results diverge.

This repository is the empirical half of that argument. Using a 14-row trace fed both in-order and out-of-order, it shows the divergences concretely and — via a total-order vs partial-order comparison — isolates a specific, under-discussed failure mode: equal-timestamp ties on a window boundary. When timestamps are unique, cross-engine differences reduce to two clean, documentable axes (sliding boundary-inclusion [start,end) vs (start,end], and session end-labeling); when the boundary carries a tie, engines that fire on watermark advance split it and produce order-dependent — even arithmetically impossible — counts. See data/README.md ("Ordering regimes") for the framing and expected-results/README.md for the measured results.

Engines evaluated

The query is a count per key over a sliding time window (width 10 ms, slide 2 ms) and a session window (gap 10 ms), reported as <window start, window end, key, count>.

Engine	Version	Window maintenance	Notes
Apache Beam	2.68.0 (direct runner)	multi-buffer	configurable triggers / accumulation
Apache Flink	1.18.1	multi-buffer	watermark-driven event time, allowedLateness, custom triggers
Spark Structured Streaming	3.5.1	multi-buffer	watermark state management, output modes; folder-fed source
Kafka Streams	3.9.1	multi-buffer	stream-time progress, grace, optional suppress-on-close
Apache Storm	2.6.0	single-buffer	trigger/evict policies, lag-based lateness; sliding only

Storm has no native session window, so it is excluded from the session assessment. (A hand-written SessionWindowBolt exists in the repo but is discontinued — not run, not compared, and not a Storm-supported window type.)

Repository layout

common/         RunConfig — shared input/window/mode selection (see below)
beam-runner/    beam.BeamTestStreamsSlidingWindow, beam.BeamTestStreamSessions
flink-runner/   flink.StreamingJobTestWindowingSemantics  (--window sliding|session)
kafka-runner/   kafka.KStreamsSlidingDemoDriverCSV, kafka.KStreamsSessionDemoDriverCSV
spark-runner/   spark.CsvSlidingStreaming, spark.CsvSessionStreaming  (folder-fed)
storm-runner/   storm.StormSlidingFromCSV  (sliding only)
scripts/        normalize, compare, spark-feed  (PowerShell + POSIX twins)
data/           input format + datasets + ordering regimes   (data/README.md)
expected-results/  reference tuples + run/compare workflow   (expected-results/README.md)
*.csv           the input datasets (see Datasets)

Prerequisites

• JDK 21. This is the baseline the results were produced on. Java 24/25 break Spark (removed javax.security.auth.Subject.getSubject), so run the engines on 21. Set JAVA_HOME to your JDK 21 per terminal; the default java on PATH may be newer.
• Maven Wrapper is committed (mvnw / mvnw.cmd, Maven 3.9.9) — no Maven install needed. mvnw.cmd is patched to tolerate spaced paths (e.g. C:\Users\Ahmed Awad).
• Spark on Windows only: Hadoop native bits (winutils) — see Platform notes.

Build

Build everything once (compiles all modules and installs common into the local repo so the runners can resolve it):

mvnw.cmd -q -DskipTests install

Then run each engine without -am (the shared exec plugin misfires across modules with -am):

mvnw.cmd -q -pl <module> exec:exec -Dexec.args="-classpath %classpath <MainClass> <args>"

exec:exec forks a JVM (required for Flink/Spark/Storm, which start a local cluster and need the system class loader). Kafka/Beam are pure in-memory and also work with exec:java, but the exec:exec form above is uniform and works for all.

Central configuration (common.RunConfig)

Every runner parses its arguments through one class, so the input and window type are selected the same way everywhere. All flags are optional; defaults reproduce the paper's runs.

Flag	Alias	Meaning	Default
--input <path>	-i	Explicit input. A CSV file for Flink/Beam/Kafka/Storm; the watched folder for Spark. A lone positional arg is also treated as the input.	—
--mode <ordered|unordered>	-m	Which bundled dataset to use when --input is absent. ordered → InorderData.csv; unordered → OutOfOrderData.csv.	unordered
--window <sliding|session>	-w	Window type. Only the Flink runner honors this (one class does both). The other engines have a dedicated class per window type, so a conflicting value there is warned about and ignored.	per-runner
--help	-h	Print usage.	—

To run on the total-order dataset, pass --input TotalOrderData.csv (Kafka/Beam/Flink/Storm) or feed Spark with spark-feed -Source total.

Running each engine

All examples assume JAVA_HOME points at JDK 21 and you are at the repo root. --mode ordered|unordered selects the dataset; swap in --input <file> for a custom trace.

Kafka Streams (sliding / session):

mvnw.cmd -q -pl kafka-runner exec:exec -Dexec.args="-classpath %classpath kafka.KStreamsSlidingDemoDriverCSV --mode unordered"
mvnw.cmd -q -pl kafka-runner exec:exec -Dexec.args="-classpath %classpath kafka.KStreamsSessionDemoDriverCSV --mode ordered"

Beam (sliding / session):

mvnw.cmd -q -pl beam-runner exec:exec -Dexec.args="-classpath %classpath beam.BeamTestStreamsSlidingWindow --mode unordered"
mvnw.cmd -q -pl beam-runner exec:exec -Dexec.args="-classpath %classpath beam.BeamTestStreamSessions --mode ordered"

Flink (one class, choose the window with --window). Flink writes results to a file in the repo root, not stdout: FlinkSlidingTimeWindowFor<inputName>Partitioned.txt for sliding and FlinkSessionWindowFor<inputName>Partitioned.txt for session (e.g. FlinkSlidingTimeWindowForInorderData.csvPartitioned.txt). Copy it over the capture before normalizing:

mvnw.cmd -q -pl flink-runner exec:exec -Dexec.args="-classpath %classpath flink.StreamingJobTestWindowingSemantics --window sliding --mode unordered"

Storm (sliding only — see the Storm platform notes about timing):

mvnw.cmd -q -pl storm-runner exec:exec -Dexec.args="-Dstorm.keepalive.ms=1800000 -classpath %classpath storm.StormSlidingFromCSV --mode ordered" > expected-results\actual\storm_sliding_ordered.out 2>&1

Spark (folder-fed — two terminals; needs the --add-opens list and Hadoop bits below). Terminal A starts Spark watching watchDir/; Terminal B feeds it:

mvnw.cmd -q -pl spark-runner exec:exec -Dexec.args="-Dhadoop.home.dir=C:\hadoop --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=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.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 --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED -classpath %classpath spark.CsvSlidingStreaming" > expected-results\actual\spark_sliding_unordered.out 2>&1

powershell -NoProfile -ExecutionPolicy Bypass -File scripts\spark-feed.ps1 -Source ooo

Capturing and comparing results (normalize each .out to canonical tuples, then compare) is documented in expected-results/README.md.

Observing transient (per-step) results (optional, diagnostic)

By default each runner reports every window once, at closure. That terminal result is what the paper compares and what expected-results/ records. Behind those numbers a multi-buffer engine actually updates a window repeatedly as elements are admitted (for a session window, growing and merging: one element, then two, then three). To watch that per-step sequence you must switch the engine into a report-on-every-update mode. These modes are diagnostic only: they are not the reported results, they are not part of the normalize/compare workflow, and each is a source edit with no command-line flag. In every case, rebuild after editing (see Build) and revert the edit afterwards to reproduce the reported numbers.

Kafka Streams (kafka.KStreamsSessionDemoDriverCSV). Per-record forwarding is the Kafka Streams default; the runner adds .suppress(Suppressed.untilWindowCloses(...)) to hold back everything but the final result. Comment out that .suppress(...) call and every update is forwarded as a changelog record, including null-valued retractions emitted when a session merge changes the window key.

Spark Structured Streaming (spark.CsvSessionStreaming). Change the sink from .outputMode("append") to .outputMode("complete"). In complete mode Spark retains all window state without eviction and re-emits the whole result table on every trigger, exposing the transient session growth. Note that complete never evicts, so it is a testing device, not a valid production reporting mode; the reported Spark results use append.

Apache Flink (flink.StreamingJobTestWindowingSemantics). Two edits:

1. In main (around line 274), enable the early trigger: jobForWindowTypeAndData(type, fileName, true, false, true);
2. In the partitioned session branch (around line 315), fire on every element: EarlyCountEventTimeTrigger.create(1);. create(2) produces nothing for session windows, because a merging window's per-instance element count is reset on each merge and never reaches two.

With the early trigger on, Flink appends WithEarlyTrigger to the output file name (e.g. FlinkSessionWindowWithEarlyTriggerForOutOfOrderData.csvPartitioned.txt). Revert both edits to reproduce the reported terminal-trigger numbers.

For example, the out-of-order session run then shows the second key-A session forming step by step, (26,36,A,1), (23,36,A,2), (14,36,A,3), as the late elements 23 and 14 are merged in, which is the mechanism behind the single reported result (14,36,A,3).

Platform notes (Windows)

The engines were run on Windows; these are the sharp edges.

JDK version. Use JDK 21. Newer JDKs (24/25) break Spark. If your default java is newer, set JAVA_HOME to JDK 21 in each terminal.

Spark needs Hadoop native bits. Spark's local file source lists the watched folder through Hadoop's RawLocalFileSystem, which calls into hadoop.dll. Without it Spark starts but dies with HADOOP_HOME and hadoop.home.dir are unset or UnsatisfiedLinkError: NativeIO$Windows.access0. Spark 3.5.1 pulls Hadoop 3.3.4, so you need Hadoop 3.3.x winutils.exe + hadoop.dll:

• put both in %HADOOP_HOME%\bin (e.g. C:\hadoop\bin),
• copy hadoop.dll into C:\Windows\System32, and
• either set HADOOP_HOME in the terminal or pass -Dhadoop.home.dir=C:\hadoop as the first token of the Spark -Dexec.args (as shown above) — the -D form is env-independent and preferred.

Spark also needs the JDK-17+ --add-opens list (shown in the Spark command); keep it on one line before -classpath. Under WSL/Linux none of this is needed.

Spark reads a folder, not a file. It watches watchDir/ with maxFilesPerTrigger=1; the dataset is chosen by the feeder (spark-feed -Source ooo|inorder|total), which clears watchDir/ and drops the single-record files in numeric order. Don't point Spark at a pre-filled folder.

Storm is slow and self-terminating. Storm sliding grinds (MAX_SPOUT_PENDING=1 plus a far-future flush watermark firing every 2 ms slide) and can take many minutes. main sleeps a fixed keep-alive before shutting the cluster down and flushing output; the duration is a non-semantic knob: -Dstorm.keepalive.ms=<millis> (default 30 min). Set it comfortably above the processing time — too short truncates the capture. Storm's watermark seed (_PRIME_ at event-time 0) assumes all event times are ≥ 0 (see data/README.md).

Ctrl+C does not kill forked JVMs. exec:exec forks a grandchild java.exe; Ctrl+C tears down mvnw/Maven but leaves the forked engine JVM running (and holding file handles). Runs pile up. Kill them explicitly, e.g.:

powershell -NoProfile -Command "Get-CimInstance Win32_Process | ? { $_.CommandLine -match 'StormSlidingFromCSV' } | % { Stop-Process -Id $_.ProcessId -Force }"

(or taskkill /PID <pid> /F). In Task Manager they all appear as java.exe; add the Command line column in the Details tab to tell them apart.

Datasets and expected results

• Input format, the three datasets (out-of-order / partial-order / total-order), and the Spark folder convention: data/README.md.
• Per-combination reference tuples and the normalize/compare workflow: expected-results/README.md.

Citation

If you use this artifact, please cite the paper "A Unified Semantic Framework for Window Operators under Late and Out-of-Order Data" (Ahmed Awad). (Full citation to be added on publication.)