Revert "[FLINK-14971][checkpointing] Make CompletedCheckpointStore thread-safe...

Revert "[FLINK-14971][checkpointing] Make CompletedCheckpointStore thread-safe to avoid synchronization outside"

This reverts commit f248e6df.

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

@@ -1370,7 +1370,6 @@ public class CheckpointCoordinator {
 		return this.pendingCheckpoints.size();
 	}
 
-	@VisibleForTesting
 	public int getNumberOfRetainedSuccessfulCheckpoints() {
 		synchronized (lock) {
 			return completedCheckpointStore.getNumberOfRetainedCheckpoints();
@@ -1383,7 +1382,6 @@ public class CheckpointCoordinator {
 		}
 	}
 
-	@VisibleForTesting
 	public List<CompletedCheckpoint> getSuccessfulCheckpoints() throws Exception {
 		synchronized (lock) {
 			return completedCheckpointStore.getAllCheckpoints();
@@ -1394,7 +1392,6 @@ public class CheckpointCoordinator {
 		return checkpointStorage;
 	}
 
-	@VisibleForTesting
 	public CompletedCheckpointStore getCheckpointStore() {
 		return completedCheckpointStore;
 	}

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

@@ -23,16 +23,12 @@ import org.apache.flink.api.common.JobStatus;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.annotation.concurrent.ThreadSafe;
-
 import java.util.List;
 import java.util.ListIterator;
 
 /**
  * A bounded LIFO-queue of {@link CompletedCheckpoint} instances.
- * Note that it might be visited by multiple threads. So implementation should keep it thread-safe.
  */
-@ThreadSafe
 public interface CompletedCheckpointStore {
 
 	Logger LOG = LoggerFactory.getLogger(CompletedCheckpointStore.class);

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

@@ -24,19 +24,15 @@ import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.annotation.concurrent.ThreadSafe;
-
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.List;
 
 import static org.apache.flink.util.Preconditions.checkArgument;
-import static org.apache.flink.util.Preconditions.checkState;
 
 /**
  * {@link CompletedCheckpointStore} for JobManagers running in {@link HighAvailabilityMode#NONE}.
  */
-@ThreadSafe
 public class StandaloneCompletedCheckpointStore implements CompletedCheckpointStore {
 
 	private static final Logger LOG = LoggerFactory.getLogger(StandaloneCompletedCheckpointStore.class);
@@ -47,8 +43,6 @@ public class StandaloneCompletedCheckpointStore implements CompletedCheckpointSt
 	/** The completed checkpoints. */
 	private final ArrayDeque<CompletedCheckpoint> checkpoints;
 
-	private boolean shutdown = false;
-
 	/**
 	 * Creates {@link StandaloneCompletedCheckpointStore}.
 	 *
@@ -70,9 +64,6 @@ public class StandaloneCompletedCheckpointStore implements CompletedCheckpointSt
 	@Override
 	public void addCheckpoint(CompletedCheckpoint checkpoint) throws Exception {
 
-		synchronized (checkpoints) {
-			checkState(!shutdown, "StandaloneCompletedCheckpointStore has been shut down");
-
 		checkpoints.addLast(checkpoint);
 
 		if (checkpoints.size() > maxNumberOfCheckpointsToRetain) {
@@ -84,21 +75,16 @@ public class StandaloneCompletedCheckpointStore implements CompletedCheckpointSt
 			}
 		}
 	}
-	}
 
 	@Override
 	public List<CompletedCheckpoint> getAllCheckpoints() {
-		synchronized (checkpoints) {
 		return new ArrayList<>(checkpoints);
 	}
-	}
 
 	@Override
 	public int getNumberOfRetainedCheckpoints() {
-		synchronized (checkpoints) {
 		return checkpoints.size();
 	}
-	}
 
 	@Override
 	public int getMaxNumberOfRetainedCheckpoints() {
@@ -107,9 +93,6 @@ public class StandaloneCompletedCheckpointStore implements CompletedCheckpointSt
 
 	@Override
 	public void shutdown(JobStatus jobStatus) throws Exception {
-		synchronized (checkpoints) {
-			if (!shutdown) {
-				shutdown = true;
 		try {
 			LOG.info("Shutting down");
 
@@ -120,8 +103,6 @@ public class StandaloneCompletedCheckpointStore implements CompletedCheckpointSt
 			checkpoints.clear();
 		}
 	}
-		}
-	}
 
 	@Override
 	public boolean requiresExternalizedCheckpoints() {

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

@@ -29,8 +29,6 @@ import org.apache.flink.util.function.ThrowingConsumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.annotation.concurrent.ThreadSafe;
-
 import java.io.IOException;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
@@ -42,7 +40,6 @@ import java.util.concurrent.Executor;
 
 import static org.apache.flink.util.Preconditions.checkArgument;
 import static org.apache.flink.util.Preconditions.checkNotNull;
-import static org.apache.flink.util.Preconditions.checkState;
 
 /**
  * {@link CompletedCheckpointStore} for JobManagers running in {@link HighAvailabilityMode#ZOOKEEPER}.
@@ -67,7 +64,6 @@ import static org.apache.flink.util.Preconditions.checkState;
  * checkpoints is consistent. Currently, after recovery we start out with only a single
  * checkpoint to circumvent those situations.
  */
-@ThreadSafe
 public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointStore {
 
 	private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperCompletedCheckpointStore.class);
@@ -89,10 +85,6 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
 
 	private final Executor executor;
 
-	private final Object lock = new Object();
-
-	private boolean shutdown = false;
-
 	/**
 	 * Creates a {@link ZooKeeperCompletedCheckpointStore} instance.
 	 *
@@ -135,15 +127,14 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
 	public void recover() throws Exception {
 		LOG.info("Recovering checkpoints from ZooKeeper.");
 
-		synchronized (lock) {
-			checkState(!shutdown, "ZooKeeperCompletedCheckpointStore has been shut down");
 		// Get all there is first
 		List<Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String>> initialCheckpoints;
 		while (true) {
 			try {
 				initialCheckpoints = checkpointsInZooKeeper.getAllAndLock();
 				break;
-				} catch (ConcurrentModificationException e) {
+			}
+			catch (ConcurrentModificationException e) {
 				LOG.warn("Concurrent modification while reading from ZooKeeper. Retrying.");
 			}
 		}
@@ -209,7 +200,6 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
 				numberOfInitialCheckpoints);
 		}
 	}
-	}
 
 	/**
 	 * Synchronously writes the new checkpoints to ZooKeeper and asynchronously removes older ones.
@@ -220,9 +210,6 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
 	public void addCheckpoint(final CompletedCheckpoint checkpoint) throws Exception {
 		checkNotNull(checkpoint, "Checkpoint");
 
-		synchronized (lock) {
-			checkState(!shutdown, "ZooKeeperCompletedCheckpointStore has been shut down");
-
 		final String path = checkpointIdToPath(checkpoint.getCheckpointID());
 
 		// Now add the new one. If it fails, we don't want to loose existing data.
@@ -238,7 +225,6 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
 
 		LOG.debug("Added {} to {}.", checkpoint, path);
 	}
-	}
 
 	private void tryRemoveCompletedCheckpoint(CompletedCheckpoint completedCheckpoint, ThrowingConsumer<CompletedCheckpoint, Exception> discardCallback) {
 		try {
@@ -259,17 +245,13 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
 
 	@Override
 	public List<CompletedCheckpoint> getAllCheckpoints() throws Exception {
-		synchronized (lock) {
 		return new ArrayList<>(completedCheckpoints);
 	}
-	}
 
 	@Override
 	public int getNumberOfRetainedCheckpoints() {
-		synchronized (lock) {
 		return completedCheckpoints.size();
 	}
-	}
 
 	@Override
 	public int getMaxNumberOfRetainedCheckpoints() {
@@ -278,9 +260,6 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
 
 	@Override
 	public void shutdown(JobStatus jobStatus) throws Exception {
-		synchronized (lock) {
-			if (!shutdown) {
-				shutdown = true;
 		if (jobStatus.isGloballyTerminalState()) {
 			LOG.info("Shutting down");
 
@@ -302,8 +281,6 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
 			checkpointsInZooKeeper.releaseAll();
 		}
 	}
-		}
-	}
 
 	// ------------------------------------------------------------------------
 

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

@@ -158,7 +158,6 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
 		assertNull(client.checkExists().forPath(CHECKPOINT_PATH + ZooKeeperCompletedCheckpointStore.checkpointIdToPath(checkpoint.getCheckpointID())));
 
 		sharedStateRegistry.close();
-		store = createCompletedCheckpoints(1);
 		store.recover();
 
 		assertEquals(0, store.getNumberOfRetainedCheckpoints());
@@ -193,7 +192,6 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
 
 		// Recover again
 		sharedStateRegistry.close();
-		store = createCompletedCheckpoints(1);
 		store.recover();
 
 		CompletedCheckpoint recovered = store.getLatestCheckpoint(false);