SEPURI-SAI-KRISHNA commented on code in PR #29345:
URL: https://github.com/apache/flink/pull/29345#discussion_r4284123416
##########
flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java:
##########
@@ -234,6 +239,58 @@ public static <OUT> Builder<OUT> ofClass(
return new Builder<>(functionClass);
}
+ //
-------------------------------------------------------------------------
+ // Test State Snapshotting
+ //
-------------------------------------------------------------------------
+
+ /**
+ * Takes a snapshot of this harness: the configuration it was built with,
the state of every
+ * partition, the pending and the fired timers, the watermark of every
table argument, and the
+ * collected output.
+ *
+ * <p>The snapshot is isolated from this harness, so processing further
elements, advancing a
+ * watermark or mutating state afterwards does not change what the
snapshot holds. State is
+ * copied in its internal representation, which means no user-facing state
object is touched.
+ * The one part that is not copied is the values collected by the PTF
itself (see {@link
+ * #getFunctionOutput()}), since those are the instances the function
emitted.
+ *
+ * @see #restoreFromSnapshot(TestSnapshot)
+ */
+ public TestSnapshot<OUT> snapshot() {
+ return new TestSnapshot<>(
+ builder,
+ stateManager.snapshotState(),
+ timerManager.snapshot(),
+ new ArrayList<>(functionOutput),
+ copyRows(output));
+ }
+
+ /**
+ * Creates a new harness that continues from the given snapshot.
+ *
+ * <p>The returned harness runs a fresh instance of the PTF, opened as
{@link Builder#build()}
+ * would, and starts out with the state, timers, watermarks and output
that the snapshot was
+ * taken with. A snapshot can be restored more than once, and each
restored harness is
+ * independent of the others.
+ *
+ * @see #snapshot()
+ */
+ public static <OUT> ProcessTableFunctionTestHarness<OUT>
restoreFromSnapshot(
+ TestSnapshot<OUT> snapshot) throws Exception {
+ checkNotNull(snapshot, "snapshot must not be null");
+
+ final ProcessTableFunctionTestHarness<OUT> harness =
snapshot.builder.build();
+ harness.stateManager.restoreState(snapshot.state);
+ harness.timerManager.restore(snapshot.timers);
+ harness.functionOutput.addAll(snapshot.functionOutput);
+ harness.output.addAll(copyRows(snapshot.output));
Review Comment:
Done. A private `build(@Nullable TestSnapshot)` seeds the state manager and
the timer manager before the harness is constructed, and `restoreFromSnapshot`
is now `snapshot.spec.toBuilder().build(snapshot)`. I used an overload rather
than a separate `buildFromSnapshot` so there is one construction path, happy to
use your name instead. On `close()`: the restore work all happens before
`openFunction()`, so a failure while copying cannot leave an opened function
behind. `openFunction()` itself does `function.open(..)` then
`setCollector(..)`, so a failure in that second step would still leak one, but
that is the same for an ordinary `build()` and unchanged here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
