Re: [PR] [FLINK-32806][runtime] Support eviction of clean job result entries in EmbeddedJobResultStore [flink]

["dalelane (via GitHub)" <gi...@apache.org>]

dalelane commented on code in PR #29304:
URL: https://github.com/apache/flink/pull/29304#discussion_r4200535396


##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/AbstractNonHaServices.java:
##########
@@ -58,7 +59,15 @@ public abstract class AbstractNonHaServices implements HighAvailabilityServices
     private boolean shutdown;
 
     public AbstractNonHaServices() {
-        this.jobResultStore = new EmbeddedJobResultStore();
+        this(new EmbeddedJobResultStore());

Review Comment:
   Should the setting reach the store used by MiniCluster?
   
   The change passes the configuration through StandaloneHaServices. However, I think EmbeddedHaServices may still be using the original constructor, which doesn't receive the configuration. That class is used when high availability is turned off (HighAvailabilityServicesUtils) and in MiniCluster.
   
   In both places the configuration is available but doesn't seem to be passed on. 



##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedJobResultStore.java:
##########
@@ -18,28 +18,66 @@
 
 package org.apache.flink.runtime.highavailability.nonha.embedded;
 
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.highavailability.AbstractThreadsafeJobResultStore;
 import org.apache.flink.runtime.highavailability.JobResultEntry;
 import org.apache.flink.runtime.highavailability.JobResultStore;
+import org.apache.flink.runtime.highavailability.JobResultStoreOptions;
 import org.apache.flink.runtime.jobmaster.JobResult;
 import org.apache.flink.util.concurrent.Executors;
 
+import org.apache.flink.shaded.guava33.com.google.common.cache.Cache;
+import org.apache.flink.shaded.guava33.com.google.common.cache.CacheBuilder;
+
+import javax.annotation.Nullable;
+
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
 /** A thread-safe in-memory implementation of the {@link JobResultStore}. */
 public class EmbeddedJobResultStore extends AbstractThreadsafeJobResultStore {
 
     private final Map<JobID, JobResultEntry> dirtyJobResults = new HashMap<>();
 
-    private final Map<JobID, JobResultEntry> cleanJobResults = new HashMap<>();
+    private final Cache<JobID, JobResultEntry> cleanJobResults;
 
+    /**
+     * Creates a store that retains the clean job results indefinitely. This corresponds to the
+     * behaviour this class always had.
+     */
     public EmbeddedJobResultStore() {
+        // a null TTL means "no expiration" which keeps the previous behaviour intact
+        this((Duration) null);
+    }
+
+    /**
+     * Creates a store that evicts the clean job results after {@link
+     * JobResultStoreOptions#CLEAN_JOB_RESULT_TTL} has passed. If the option is not configured, the
+     * clean job results are retained indefinitely.
+     */
+    public EmbeddedJobResultStore(Configuration configuration) {
+        this(
+                checkNotNull(configuration, "configuration")
+                        .get(JobResultStoreOptions.CLEAN_JOB_RESULT_TTL));
+    }
+
+    @VisibleForTesting
+    EmbeddedJobResultStore(@Nullable Duration cleanJobResultTtl) {
         super(Executors.directExecutor());
+        final CacheBuilder<Object, Object> cacheBuilder = CacheBuilder.newBuilder();
+        if (cleanJobResultTtl != null) {
+            cacheBuilder.expireAfterAccess(cleanJobResultTtl.toMillis(), TimeUnit.MILLISECONDS);

Review Comment:
   worth validating that the value is positive?



##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedJobResultStore.java:
##########
@@ -67,7 +105,7 @@ public boolean hasDirtyJobResultEntryInternal(JobID jobId) {
 
     @Override
     public boolean hasCleanJobResultEntryInternal(JobID jobId) {
-        return cleanJobResults.containsKey(jobId);
+        return cleanJobResults.asMap().containsKey(jobId);

Review Comment:
   Could a finished job be run a second time after its entry is removed?
   
   IIUC, the job result store does two jobs. It records results, and it stops the same job from being run twice. When a job is submitted, the Dispatcher asks the store whether it already has an entry for that job ID. If it does, the job is turned away (Dispatcher#isInGloballyTerminalState, which calls hasJobResultEntryAsync).
   
   With this change, once the time limit has passed, the clean entry is removed and the store no longer knows the job existed. If the same job ID is submitted again after that, it looks like the job would be accepted and run again.
   
   There is a separate check at the application level, and the application result store has no time limit, so submitting the same application again is still turned away. The case I have in mind is the same job ID arriving under a different application. If that's known/intentional, maybe it's just worth a sentence in the option's description to make it clear?



-- 
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.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org