[FLINK-3499] [runtime] Clear ZooKeeper references on recovery

This closes #1707.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java

@@ -137,6 +137,11 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
 	public void recover() throws Exception {
 		LOG.info("Recovering checkpoints from ZooKeeper.");
 
+		// Clear local handles in order to prevent duplicates on
+		// recovery. The local handles should reflect the state
+		// of ZooKeeper.
+		checkpointStateHandles.clear();
+
 		// Get all there is first
 		List<Tuple2<StateHandle<CompletedCheckpoint>, String>> initialCheckpoints;
 		while (true) {

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreITCase.java

@@ -87,6 +87,7 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
 
 		// All three should be in ZK
 		assertEquals(3, ZooKeeper.getClient().getChildren().forPath(CheckpointsPath).size());
+		assertEquals(3, checkpoints.getNumberOfRetainedCheckpoints());
 
 		// Recover
 		checkpoints.recover();
@@ -102,6 +103,7 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
 		}
 
 		assertEquals(1, ZooKeeper.getClient().getChildren().forPath(CheckpointsPath).size());
+		assertEquals(1, checkpoints.getNumberOfRetainedCheckpoints());
 		assertEquals(expected[2], checkpoints.getLatestCheckpoint());
 	}
 }

flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerCheckpointRecoveryITCase.java → flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHACheckpointRecoveryITCase.java

@@ -74,7 +74,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
 
-public class JobManagerCheckpointRecoveryITCase extends TestLogger {
+public class JobManagerHACheckpointRecoveryITCase extends TestLogger {
 
 	@Rule
 	public RetryRule retryRule = new RetryRule();