[FLINK-3107] [runtime] Defer start of checkpoint ID counter

apache · Feb 12, 2016 · 937963e · 937963e
1 parent 8e55bbd
commit 937963e
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 1 deletion.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
@@ -230,6 +230,7 @@ public ApplicationID restoreSavepoint(
 
 			// Reset the checkpoint ID counter
 			long nextCheckpointId = checkpoint.getCheckpointID();
+			checkpointIdCounter.start();
 			checkpointIdCounter.setCount(nextCheckpointId + 1);
 			LOG.info("Reset the checkpoint ID to {}", nextCheckpointId);
 

diff --git a/...k-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorTest.java b/...k-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorTest.java
@@ -175,6 +175,7 @@ public void testSimpleRollbackSavepoint() throws Exception {
 			}
 		}
 
+		MockCheckpointIdCounter idCounter = new MockCheckpointIdCounter();
 		StateStore<Savepoint> savepointStore = new HeapStateStore<>();
 
 		SavepointCoordinator coordinator = createSavepointCoordinator(
@@ -184,7 +185,7 @@ public void testSimpleRollbackSavepoint() throws Exception {
 				triggerVertices,
 				ackVertices,
 				new ExecutionVertex[] {},
-				new MockCheckpointIdCounter(),
+				idCounter,
 				savepointStore);
 
 		Future<String> savepointPathFuture = coordinator.triggerSavepoint(1231273123);
@@ -213,6 +214,9 @@ public void testSimpleRollbackSavepoint() throws Exception {
 		// Verify all promises removed
 		assertEquals(0, getSavepointPromises(coordinator).size());
 
+		// Verify checkpoint ID counter started
+		assertTrue(idCounter.isStarted());
+
 		coordinator.shutdown();
 	}
 
@@ -1083,15 +1087,18 @@ private static ExecutionVertex mockExecutionVertex(
 
 	private static class MockCheckpointIdCounter implements CheckpointIDCounter {
 
+		private boolean started;
 		private long count;
 		private long lastReturnedCount;
 
 		@Override
 		public void start() throws Exception {
+			started = true;
 		}
 
 		@Override
 		public void stop() throws Exception {
+			started = false;
 		}
 
 		@Override
@@ -1108,5 +1115,9 @@ public void setCount(long newCount) {
 		long getLastReturnedCount() {
 			return lastReturnedCount;
 		}
+
+		public boolean isStarted() {
+			return started;
+		}
 	}
 }