[GitHub] flink pull request #5239: [FLINK-8360] Implement task-local state recovery

[StefanRRichter]

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

    https://github.com/apache/flink/pull/5239#discussion_r168494204
  
    --- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
 ---
    @@ -553,27 +616,36 @@ public boolean supportsAsynchronousSnapshots() {
                                        metaInfoSnapshots,
                                        
!Objects.equals(UncompressedStreamCompressionDecorator.INSTANCE, 
keyGroupCompressionDecorator));
     
    +                   LocalRecoveryDirectoryProvider directoryProvider =
    +                           
FsStateBackend.LocalRecoveryMode.ENABLE_FILE_BASED.equals(localRecoveryConfig.getLocalRecoveryMode())
 ?
    +                                   
localRecoveryConfig.getLocalStateDirectories() :
    +                                   null;
    +
    +                   final 
ThrowingSupplier<CheckpointStreamWithResultProvider, Exception> 
checkpointStreamSupplier =
    +                           () -> new 
CheckpointStreamWithResultProvider.Factory().create(
    +                                   checkpointId,
    +                                   CheckpointedStateScope.EXCLUSIVE,
    --- End diff --
    
    Already changed in later commit.


---