echauchot commented on a change in pull request #13040:
URL: https://github.com/apache/flink/pull/13040#discussion_r469246313
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointsCleaner.java
##########
@@ -0,0 +1,86 @@
+package org.apache.flink.runtime.checkpoint;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStorageLocation;
+import org.apache.flink.runtime.state.StateUtil;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.function.ThrowingConsumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Delegate class responsible for checkpoints cleaning and counting the number
of checkpoints yet
+ * to clean.
+ */
+public class CheckpointsCleaner {
+ /** The executor used for asynchronous calls, like potentially blocking
I/O. */
+ private final Executor executor;
+ AtomicInteger numberOfCheckpointsToClean;
+
+ Runnable checkpointCleaningFinishedCallback;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CheckpointsCleaner.class);
+
+ public CheckpointsCleaner(Executor executor) {
+ this.executor = executor;
+ this.numberOfCheckpointsToClean = new AtomicInteger(0);
+ }
+
+ public void setCheckpointCleaningFinishedCallback(Runnable
checkpointCleaningFinishedCallback) {
+ this.checkpointCleaningFinishedCallback =
checkpointCleaningFinishedCallback;
+ }
+
+ public int getNumberOfCheckpointsToClean() {
+ return numberOfCheckpointsToClean.get();
+ }
+
+ /**
+ * Asynchronously call a discard callback on on the ioExecutor
+ * (FixedThreadPool of configurable size of default 4*CPU cores)
+ * and count the number of checkpoints that are waiting to clean.
+ * @param completedCheckpoint the checkpoint to discard
+ * @param discardCallback the discard callback to call
+ */
+ public void
asyncDiscardCheckpointAndCountCheckpoints(CompletedCheckpoint
completedCheckpoint, ThrowingConsumer<CompletedCheckpoint, Exception>
discardCallback){
+ numberOfCheckpointsToClean.incrementAndGet();
Review comment:
The problem of this ticket is that discard runnables are stacked into
the executor queue until OOM and not started. So I meant to increment the
counter before submitting the runnable to the executor and not as soon as they
start so that we count stacked runnables. If we put the increment inside the
runnable itself, we will count only running runnables. WDYT ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
