Skip to content

Commit

Permalink
[FLINK-3499] [runtime] Clear ZooKeeper references on recovery
Browse files Browse the repository at this point in the history
This closes #1707.
  • Loading branch information
uce committed Feb 24, 2016
1 parent be68b17 commit af2091c
Show file tree
Hide file tree
Showing 3 changed files with 8 additions and 1 deletion.
Expand Up @@ -137,6 +137,11 @@ public ZooKeeperCompletedCheckpointStore(
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) {
Expand Down
Expand Up @@ -87,6 +87,7 @@ public void testRecover() throws Exception {

// All three should be in ZK
assertEquals(3, ZooKeeper.getClient().getChildren().forPath(CheckpointsPath).size());
assertEquals(3, checkpoints.getNumberOfRetainedCheckpoints());

// Recover
checkpoints.recover();
Expand All @@ -102,6 +103,7 @@ public void testRecover() throws Exception {
}

assertEquals(1, ZooKeeper.getClient().getChildren().forPath(CheckpointsPath).size());
assertEquals(1, checkpoints.getNumberOfRetainedCheckpoints());
assertEquals(expected[2], checkpoints.getLatestCheckpoint());
}
}
Expand Up @@ -74,7 +74,7 @@
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();
Expand Down

0 comments on commit af2091c

Please sign in to comment.