@Test public void testSnapshotUtilsLifecycle() throws Exception { StreamOperator<Void> operator = new LifecycleOperator(); CheckpointStorageWorkerView storage = new MockStateBackend().createCheckpointStorage(new JobID()); Path path = new Path(folder.newFolder().getAbsolutePath()); SnapshotUtils.snapshot(operator, 0, 0L, storage, path); Assert.assertEquals(EXPECTED_CALL_OPERATOR_SNAPSHOT, ACTUAL_ORDER_TRACKING); }
public static <OUT, OP extends StreamOperator<OUT>> TaggedOperatorSubtaskState snapshot( OP operator, int index, long timestamp, CheckpointStorageWorkerView checkpointStorage, Path savepointPath) throws Exception { CheckpointOptions options = new CheckpointOptions( CheckpointType.SAVEPOINT, AbstractFsCheckpointStorage.encodePathAsReference(savepointPath)); operator.prepareSnapshotPreBarrier(CHECKPOINT_ID); CheckpointStreamFactory storage = checkpointStorage.resolveCheckpointStorageLocation( CHECKPOINT_ID, options.getTargetLocation()); OperatorSnapshotFutures snapshotInProgress = operator.snapshotState( CHECKPOINT_ID, timestamp, options, storage); OperatorSubtaskState state = new OperatorSnapshotFinalizer(snapshotInProgress).getJobManagerOwnedState(); operator.notifyCheckpointComplete(CHECKPOINT_ID); return new TaggedOperatorSubtaskState(index, state); }
SnapshotUtils { public static <OUT, OP extends StreamOperator<OUT>> TaggedOperatorSubtaskState snapshot( OP operator, int index, long timestamp, CheckpointStorageWorkerView checkpointStorage, Path savepointPath) throws Exception { CheckpointOptions options = new CheckpointOptions( CheckpointType.SAVEPOINT, AbstractFsCheckpointStorage.encodePathAsReference(savepointPath)); operator.prepareSnapshotPreBarrier(CHECKPOINT_ID); CheckpointStreamFactory storage = checkpointStorage.resolveCheckpointStorageLocation( CHECKPOINT_ID, options.getTargetLocation()); OperatorSnapshotFutures snapshotInProgress = operator.snapshotState( CHECKPOINT_ID, timestamp, options, storage); OperatorSubtaskState state = new OperatorSnapshotFinalizer(snapshotInProgress).getJobManagerOwnedState(); operator.notifyCheckpointComplete(CHECKPOINT_ID); return new TaggedOperatorSubtaskState(index, state); } }
SnapshotUtils { public static <OUT, OP extends StreamOperator<OUT>> TaggedOperatorSubtaskState snapshot( OP operator, int index, long timestamp, CheckpointStorageWorkerView checkpointStorage, Path savepointPath) throws Exception { CheckpointOptions options = new CheckpointOptions( CheckpointType.SAVEPOINT, AbstractFsCheckpointStorage.encodePathAsReference(savepointPath)); operator.prepareSnapshotPreBarrier(CHECKPOINT_ID); CheckpointStreamFactory storage = checkpointStorage.resolveCheckpointStorageLocation( CHECKPOINT_ID, options.getTargetLocation()); OperatorSnapshotFutures snapshotInProgress = operator.snapshotState( CHECKPOINT_ID, timestamp, options, storage); OperatorSubtaskState state = new OperatorSnapshotFinalizer(snapshotInProgress).getJobManagerOwnedState(); operator.notifyCheckpointComplete(CHECKPOINT_ID); return new TaggedOperatorSubtaskState(index, state); } private  SnapshotUtils(); }
SnapshotUtils { public static <OUT, OP extends StreamOperator<OUT>> TaggedOperatorSubtaskState snapshot( OP operator, int index, long timestamp, CheckpointStorageWorkerView checkpointStorage, Path savepointPath) throws Exception { CheckpointOptions options = new CheckpointOptions( CheckpointType.SAVEPOINT, AbstractFsCheckpointStorage.encodePathAsReference(savepointPath)); operator.prepareSnapshotPreBarrier(CHECKPOINT_ID); CheckpointStreamFactory storage = checkpointStorage.resolveCheckpointStorageLocation( CHECKPOINT_ID, options.getTargetLocation()); OperatorSnapshotFutures snapshotInProgress = operator.snapshotState( CHECKPOINT_ID, timestamp, options, storage); OperatorSubtaskState state = new OperatorSnapshotFinalizer(snapshotInProgress).getJobManagerOwnedState(); operator.notifyCheckpointComplete(CHECKPOINT_ID); return new TaggedOperatorSubtaskState(index, state); } private  SnapshotUtils(); static TaggedOperatorSubtaskState snapshot(
		OP operator,
		int index,
		long timestamp,
		CheckpointStorageWorkerView checkpointStorage,
		Path savepointPath); }
SnapshotUtils { public static <OUT, OP extends StreamOperator<OUT>> TaggedOperatorSubtaskState snapshot( OP operator, int index, long timestamp, CheckpointStorageWorkerView checkpointStorage, Path savepointPath) throws Exception { CheckpointOptions options = new CheckpointOptions( CheckpointType.SAVEPOINT, AbstractFsCheckpointStorage.encodePathAsReference(savepointPath)); operator.prepareSnapshotPreBarrier(CHECKPOINT_ID); CheckpointStreamFactory storage = checkpointStorage.resolveCheckpointStorageLocation( CHECKPOINT_ID, options.getTargetLocation()); OperatorSnapshotFutures snapshotInProgress = operator.snapshotState( CHECKPOINT_ID, timestamp, options, storage); OperatorSubtaskState state = new OperatorSnapshotFinalizer(snapshotInProgress).getJobManagerOwnedState(); operator.notifyCheckpointComplete(CHECKPOINT_ID); return new TaggedOperatorSubtaskState(index, state); } private  SnapshotUtils(); static TaggedOperatorSubtaskState snapshot(
		OP operator,
		int index,
		long timestamp,
		CheckpointStorageWorkerView checkpointStorage,
		Path savepointPath);  }