flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
/**
	 * FLINK-5985
	 *
	 * <p>This test ensures we can restore from a savepoint under modifications to the job graph that only concern
	 * stateless operators.
	 */
FLINK-5985
This test ensures we can restore from a savepoint under modifications to the job graph that only concern
stateless operators.
FLINK - 5985 This test ensures we can restore from a savepoint under modifications to the job graph that only concern stateless operators .
@Test
	public void testCanRestoreWithModifiedStatelessOperators() throws Exception {
		int numTaskManagers = 2;
		int numSlotsPerTaskManager = 2;
		int parallelism = 2;
		final Deadline deadline = new FiniteDuration(5, TimeUnit.MINUTES).fromNow();
		final File tmpDir = folder.getRoot();
		final File savepointDir = new File(tmpDir, "savepoints");
		final Configuration config = new Configuration();
		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, numTaskManagers);
		config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlotsPerTaskManager);
		config.setString(CoreOptions.SAVEPOINT_DIRECTORY,
				savepointDir.toURI().toString());
		String savepointPath;
		LOG.info("Flink configuration: " + config + ".");
		TestingCluster flink = new TestingCluster(config);
		try {
			LOG.info("Starting Flink cluster.");
			flink.start(true);
			LOG.info("Retrieving JobManager.");
			ActorGateway jobManager = Await.result(
					flink.leaderGateway().future(),
					deadline.timeLeft());
			LOG.info("JobManager: " + jobManager + ".");
			final StatefulCounter statefulCounter = new StatefulCounter();
			StatefulCounter.resetForTest(parallelism);
			StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
			env.setParallelism(parallelism);
			env.addSource(new InfiniteTestSource())
					.shuffle()
					.map(new MapFunction<Integer, Integer>() {
						@Override
						public Integer map(Integer value) throws Exception {
							return 4 * value;
						}
					})
					.shuffle()
					.map(statefulCounter).uid("statefulCounter")
					.shuffle()
					.map(new MapFunction<Integer, Integer>() {
						@Override
						public Integer map(Integer value) throws Exception {
							return 2 * value;
						}
					})
					.addSink(new DiscardingSink<Integer>());
			JobGraph originalJobGraph = env.getStreamGraph().getJobGraph();
			JobSubmissionResult submissionResult = flink.submitJobDetached(originalJobGraph);
			JobID jobID = submissionResult.getJobID();
			StatefulCounter.getProgressLatch().await(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
			Future<Object> savepointPathFuture = jobManager.ask(new TriggerSavepoint(jobID, Option.<String>empty()), deadline.timeLeft());
			savepointPath = ((TriggerSavepointSuccess) Await.result(savepointPathFuture, deadline.timeLeft())).savepointPath();
			Future<Object> savepointFuture = jobManager.ask(new RequestSavepoint(savepointPath), deadline.timeLeft());
			((ResponseSavepoint) Await.result(savepointFuture, deadline.timeLeft())).savepoint();
			LOG.info("Retrieved savepoint: " + savepointPath + ".");
		} finally {
			LOG.info("Shutting down Flink cluster.");
			flink.stop();
		}
		flink = new TestingCluster(config);
		try {
			LOG.info("Restarting Flink cluster.");
			flink = new TestingCluster(config);
			flink.start(true);
			LOG.info("Retrieving JobManager.");
			ActorGateway jobManager = Await.result(flink.leaderGateway().future(), deadline.timeLeft());
			LOG.info("JobManager: " + jobManager + ".");
			StatefulCounter.resetForTest(parallelism);
			StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
			env.setParallelism(parallelism);
			env.addSource(new InfiniteTestSource())
					.shuffle()
					.map(new StatefulCounter()).uid("statefulCounter")
					.shuffle()
					.map(new MapFunction<Integer, Integer>() {
						@Override
						public Integer map(Integer value) throws Exception {
							return value;
						}
					})
					.addSink(new DiscardingSink<Integer>());
			JobGraph modifiedJobGraph = env.getStreamGraph().getJobGraph();
			modifiedJobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(savepointPath));
			LOG.info("Resubmitting job " + modifiedJobGraph.getJobID() + " with " +
					"savepoint path " + savepointPath + " in detached mode.");
			flink.submitJobDetached(modifiedJobGraph);
			StatefulCounter.getRestoreLatch().await(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
			StatefulCounter.getProgressLatch().await(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
		} finally {
			flink.stop();
		}
	}
@ Test public void testCanRestoreWithModifiedStatelessOperators ( ) throws Exception { int numTaskManagers = 2 ; int numSlotsPerTaskManager = 2 ; int parallelism = 2 ; final Deadline deadline = new FiniteDuration ( 5 , TimeUnit . MINUTES ) . fromNow ( ) ; final File tmpDir = folder . getRoot ( ) ; final File savepointDir = new File ( tmpDir , "savepoints" ) ; final Configuration config = new Configuration ( ) ; config . setInteger ( ConfigConstants . LOCAL_NUMBER_TASK_MANAGER , numTaskManagers ) ; config . setInteger ( ConfigConstants . TASK_MANAGER_NUM_TASK_SLOTS , numSlotsPerTaskManager ) ; config . setString ( CoreOptions . SAVEPOINT_DIRECTORY , savepointDir . toURI ( ) . toString ( ) ) ; String savepointPath ; LOG . info ( "Flink configuration: " + config + "." ) ; TestingCluster flink = new TestingCluster ( config ) ; try { LOG . info ( "Starting Flink cluster." ) ; flink . start ( true ) ; LOG . info ( "Retrieving JobManager." ) ; ActorGateway jobManager = Await . result ( flink . leaderGateway ( ) . future ( ) , deadline . timeLeft ( ) ) ; LOG . info ( "JobManager: " + jobManager + "." ) ; final StatefulCounter statefulCounter = new StatefulCounter ( ) ; StatefulCounter . resetForTest ( parallelism ) ; StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . setParallelism ( parallelism ) ; env . addSource ( new InfiniteTestSource ( ) ) . shuffle ( ) . map ( new MapFunction < Integer , Integer > ( ) { @ Override public Integer map ( Integer value ) throws Exception { return 4 * value ; } } ) . shuffle ( ) . map ( statefulCounter ) . uid ( "statefulCounter" ) . shuffle ( ) . map ( new MapFunction < Integer , Integer > ( ) { @ Override public Integer map ( Integer value ) throws Exception { return 2 * value ; } } ) . addSink ( new DiscardingSink < Integer > ( ) ) ; JobGraph originalJobGraph = env . getStreamGraph ( ) . getJobGraph ( ) ; JobSubmissionResult submissionResult = flink . submitJobDetached ( originalJobGraph ) ; JobID jobID = submissionResult . getJobID ( ) ; StatefulCounter . getProgressLatch ( ) . await ( deadline . timeLeft ( ) . toMillis ( ) , TimeUnit . MILLISECONDS ) ; Future < Object > savepointPathFuture = jobManager . ask ( new TriggerSavepoint ( jobID , Option . < String > empty ( ) ) , deadline . timeLeft ( ) ) ; savepointPath = ( ( TriggerSavepointSuccess ) Await . result ( savepointPathFuture , deadline . timeLeft ( ) ) ) . savepointPath ( ) ; Future < Object > savepointFuture = jobManager . ask ( new RequestSavepoint ( savepointPath ) , deadline . timeLeft ( ) ) ; ( ( ResponseSavepoint ) Await . result ( savepointFuture , deadline . timeLeft ( ) ) ) . savepoint ( ) ; LOG . info ( "Retrieved savepoint: " + savepointPath + "." ) ; } finally { LOG . info ( "Shutting down Flink cluster." ) ; flink . stop ( ) ; } flink = new TestingCluster ( config ) ; try { LOG . info ( "Restarting Flink cluster." ) ; flink = new TestingCluster ( config ) ; flink . start ( true ) ; LOG . info ( "Retrieving JobManager." ) ; ActorGateway jobManager = Await . result ( flink . leaderGateway ( ) . future ( ) , deadline . timeLeft ( ) ) ; LOG . info ( "JobManager: " + jobManager + "." ) ; StatefulCounter . resetForTest ( parallelism ) ; StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . setParallelism ( parallelism ) ; env . addSource ( new InfiniteTestSource ( ) ) . shuffle ( ) . map ( new StatefulCounter ( ) ) . uid ( "statefulCounter" ) . shuffle ( ) . map ( new MapFunction < Integer , Integer > ( ) { @ Override public Integer map ( Integer value ) throws Exception { return value ; } } ) . addSink ( new DiscardingSink < Integer > ( ) ) ; JobGraph modifiedJobGraph = env . getStreamGraph ( ) . getJobGraph ( ) ; modifiedJobGraph . setSavepointRestoreSettings ( SavepointRestoreSettings . forPath ( savepointPath ) ) ; LOG . info ( "Resubmitting job " + modifiedJobGraph . getJobID ( ) + " with " + "savepoint path " + savepointPath + " in detached mode." ) ; flink . submitJobDetached ( modifiedJobGraph ) ; StatefulCounter . getRestoreLatch ( ) . await ( deadline . timeLeft ( ) . toMillis ( ) , TimeUnit . MILLISECONDS ) ; StatefulCounter . getProgressLatch ( ) . await ( deadline . timeLeft ( ) . toMillis ( ) , TimeUnit . MILLISECONDS ) ; } finally { flink . stop ( ) ; } }
FLINK-5985
 <p>This test ensures we can restore from a savepoint under modifications to the job graph that only concern
 stateless operators.
FLINK - 5985 <p > This test ensures we can restore from a savepoint under modifications to the job graph that only concern stateless operators .
// Config // Test deadline // Flink configuration // Start Flink // Retrieve the job manager // wait for the Tasks to be ready // Shut down the Flink cluster (thereby canceling the job) // create a new TestingCluster to make sure we start with completely // new resources // Retrieve the job manager // Reset static test helpers // Gather all task deployment descriptors // generate a modified job graph that adds a stateless op // Set the savepoint path // Submit the job // Await state is restored // Await some progress after restore