Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/19416#discussion_r142303254
--- Diff:
sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateSuite.scala
---
@@ -376,9 +388,35 @@ class FlatMapGroupsWithStateSuite extends
StateStoreMetricsTest with BeforeAndAf
expectedTimeoutTimestamp = currentBatchTimestamp + 5000) //
timestamp should change
testStateUpdateWithData(
+ s"ProcessingTimeTimeout - $testName - timeout updated after state
removed",
+ stateUpdates = state => { state.remove();
state.setTimeoutDuration(5000) },
+ timeoutConf = ProcessingTimeTimeout,
+ priorState = priorState,
+ priorTimeoutTimestamp = priorTimeoutTimestamp,
+ expectedState = None,
+ expectedTimeoutTimestamp = currentBatchTimestamp + 5000)
+
+ // Tests with EventTimeTimeout
+
+ if (priorState == None) {
+ testStateUpdateWithData(
+ s"EventTimeTimeout - $testName - setting timeout without init
state not allowed",
+ stateUpdates = state => {
+ state.setTimeoutTimestamp(10000)
+ },
+ timeoutConf = EventTimeTimeout,
+ priorState = None,
+ priorTimeoutTimestamp = priorTimeoutTimestamp,
+ expectedState = None,
+ expectedTimeoutTimestamp = 10000)
+ }
+
+ testStateUpdateWithData(
s"EventTimeTimeout - $testName - state and timeout timestamp
updated",
stateUpdates =
- (state: GroupState[Int]) => { state.update(5);
state.setTimeoutTimestamp(5000) },
+ (state: GroupState[Int]) => {
--- End diff --
undo this change.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
