flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Clay Teeter <clay.tee...@maalka.com>
Subject UnitTests and ProcessTimeWindows - Missing results
Date Thu, 26 Sep 2019 18:49:11 GMT
What is the best way to run unit tests on streams that contain
ProcessTimeWindows?

Example:

def bufferDataStreamByProcessId(ds: DataStream[MaalkaRecord]):
DataStream[MaalkaRecord] = {
  ds.map { r =>
    println(s"data in: $r") // Data shows up here
    r
  }.keyBy { mr =>
    val r = mr.asInstanceOf[MaalkaDataRecord]
    r.dataProcessingId -> r.isBefore
  }
    .window(ProcessingTimeSessionWindows.withGap(Time.milliseconds(10)))
    .reduce { (_, r2) =>
      println(s"Reducing: r2") // Data does NOT show up here
      r2
    }
    .map { r =>
      println(s"Emitted: $r") // Data does NOT show up here
      r
    }
}


This stream completes about 100ms after the first element is received by
the ProcessingTimeSessionWindow but no data is emitted from the
sessionWindow.

If i change the window to use a
TumblingProcessTimeWindow.of(Time.milliseconds(1)) some of windows do emit,
but the many of the expected values are missing.

Any ideas?

Cheers,
Clay

Mime
View raw message