Fix test for complete

Author: holdenk

core/src/main/scala/com/high-performance-spark-examples/streaming/structuredstreaming/JsonWindowedAggExample.scala

@@ -27,8 +27,7 @@ object JsonWindowedAggExample {
       checkpointDir = "/tmp/checkpoints/json_windowed_agg",
       outputFormat = "console",
       queryName = None,
-      trigger = Trigger.ProcessingTime("5 seconds"),
-      addWatermark = false
+      trigger = Trigger.ProcessingTime("5 seconds")
     )
   }
 
@@ -39,23 +38,23 @@ object JsonWindowedAggExample {
       checkpointDir: String,
       outputFormat: String,
       queryName: Option[String],
-      trigger: Trigger,
-      addWatermark: Boolean
+      trigger: Trigger
   ): StreamingQuery = {
     import spark.implicits._
 
+    // tag::streaming_ex_json_window[]
     val df = spark.readStream
       .format("json")
       .schema("timestamp TIMESTAMP, word STRING")
       .load(inputPath)
 
-    val base = if (addWatermark) df.withWatermark("timestamp", "5 minutes") else df
-    val windowed = base
+    val windowed = df
       .groupBy(window(col("timestamp"), "10 minutes"), col("word"))
       .count()
+    // end::streaming_ex_json_window[]
 
     val writer = windowed.writeStream
-      .outputMode("append")
+      .outputMode("complete") // Append would need a watermark
       .format(outputFormat)
       .option("checkpointLocation", checkpointDir)
       .trigger(trigger)

core/src/test/scala/com/high-performance-spark-examples/streaming/structuredstreaming/JsonWindowedAggExampleSuite.scala

@@ -35,7 +35,6 @@ class JsonWindowedAggExampleFileIT extends AnyFunSuite {
         outputFormat = "memory",                       // assertable sink
         queryName = Some(qName),
         trigger = Trigger.ProcessingTime("250 milliseconds"),
-        addWatermark = true                            // watermark = 5 minutes (set in builder)
       )
 
       // --- Batch 1: events in [10:00,10:10)
@@ -50,8 +49,10 @@ class JsonWindowedAggExampleFileIT extends AnyFunSuite {
       // Let the stream pick up batch 1
       q.processAllAvailable() // ok in tests
 
-      // Nothing should be emitted yet in append mode (window not closed)
-      assert(spark.table(qName).count() == 0)
+      val initialCount = spark.table(qName).count()
+
+      // We're running in complete mode, we should see some records.
+      assert(initialCount > 0)
 
       // --- Batch 2: later event at 10:16 moves max event time to 10:16
       // Watermark = maxEventTime - 5m = 10:11 >= 10:10, so [10:00,10:10) closes and emits.
@@ -75,7 +76,8 @@ class JsonWindowedAggExampleFileIT extends AnyFunSuite {
 
       val expectedAfterBatch2 = Set(
         ("2025-01-01 10:00:00", "2025-01-01 10:10:00", "hello", 2L),
-        ("2025-01-01 10:00:00", "2025-01-01 10:10:00", "world", 1L)
+        ("2025-01-01 10:00:00", "2025-01-01 10:10:00", "world", 1L),
+        ("2025-01-01 10:10:00", "2025-01-01 10:20:00", "hello", 1L)
       )
       assert(afterBatch2 == expectedAfterBatch2)
 
@@ -100,7 +102,8 @@ class JsonWindowedAggExampleFileIT extends AnyFunSuite {
         .toSet
 
       val expectedFinal = expectedAfterBatch2 ++ Set(
-        ("2025-01-01 10:10:00", "2025-01-01 10:20:00", "hello", 1L)
+        ("2025-01-01 10:10:00", "2025-01-01 10:20:00", "hello", 1L),
+        ("2025-01-01 10:20:00", "2025-01-01 10:30:00", "noop", 1),
       )
       assert(finalOut == expectedFinal)