Github user StefanRRichter commented on a diff in the pull request:

    https://github.com/apache/flink/pull/5239#discussion_r168168381
  
    --- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskLocalStateStore.java
 ---
    @@ -46,26 +52,63 @@
        /** */
        private final int subtaskIndex;
     
    +   /** */
    +   private final Map<Long, TaskStateSnapshot> 
storedTaskStateByCheckpointID;
    +
    +   /** This is the base directory for all local state of the subtask that 
owns this {@link TaskLocalStateStore}. */
    +   private final File subtaskLocalStateBaseDirectory;
    +
        public TaskLocalStateStore(
                JobID jobID,
                JobVertexID jobVertexID,
    -           int subtaskIndex) {
    +           int subtaskIndex,
    +           File localStateRootDir) {
     
                this.jobID = jobID;
                this.jobVertexID = jobVertexID;
                this.subtaskIndex = subtaskIndex;
    +           this.storedTaskStateByCheckpointID = new HashMap<>();
    +           this.subtaskLocalStateBaseDirectory =
    +                   new File(localStateRootDir, createSubtaskPath(jobID, 
jobVertexID, subtaskIndex));
    +   }
    +
    +   static String createSubtaskPath(JobID jobID, JobVertexID jobVertexID, 
int subtaskIndex) {
    +           return "jid-" + jobID + "_vtx-" + jobVertexID + "_sti-" + 
subtaskIndex;
        }
     
        public void storeLocalState(
                @Nonnull CheckpointMetaData checkpointMetaData,
                @Nullable TaskStateSnapshot localState) {
     
    -           if (localState != null) {
    -                   throw new UnsupportedOperationException("Implement this 
before actually providing local state!");
    +           TaskStateSnapshot previous =
    +                   
storedTaskStateByCheckpointID.put(checkpointMetaData.getCheckpointId(), 
localState);
    +
    +           if (previous != null) {
    +                   throw new IllegalStateException("Found previously 
registered local state for checkpoint with id " +
    +                           checkpointMetaData.getCheckpointId() + "! This 
indicated a problem.");
                }
        }
     
    -   public void dispose() {
    -           //TODO
    +   public void dispose() throws Exception {
    +
    +           Exception collectedException = null;
    +
    +           for (TaskStateSnapshot snapshot : 
storedTaskStateByCheckpointID.values()) {
    +                   try {
    +                           snapshot.discardState();
    +                   } catch (Exception discardEx) {
    +                           collectedException = 
ExceptionUtils.firstOrSuppressed(discardEx, collectedException);
    +                   }
    +           }
    +
    +           if (collectedException != null) {
    +                   throw collectedException;
    +           }
    +
    +           
FileUtils.deleteDirectoryQuietly(subtaskLocalStateBaseDirectory);
    --- End diff --
    
    This already works different in later commits.


---

Reply via email to