@Test
  public void testGetStoreActionsForLoggedPersistentStore_DeleteStoreCheckpointIfLocalOffsetHigherThanCheckpointed() {
    TaskModel mockTaskModel = mock(TaskModel.class);
    TaskName taskName = new TaskName("Partition 0");
    when(mockTaskModel.getTaskName()).thenReturn(taskName);
    Partition taskChangelogPartition = new Partition(0);
    when(mockTaskModel.getChangelogPartition()).thenReturn(taskChangelogPartition);

    String store1Name = "store1";
    StorageEngine store1Engine = mock(StorageEngine.class);
    StoreProperties mockStore1Properties = mock(StoreProperties.class);
    when(store1Engine.getStoreProperties()).thenReturn(mockStore1Properties);
    when(mockStore1Properties.isLoggedStore()).thenReturn(true);
    when(mockStore1Properties.isPersistedToDisk()).thenReturn(true);
    Map<String, StorageEngine> mockStoreEngines = ImmutableMap.of(store1Name, store1Engine);

    String changelog1SystemName = "system1";
    String changelog1StreamName = "store1Changelog";
    SystemStream changelog1SystemStream = new SystemStream(changelog1SystemName, changelog1StreamName);
    SystemStreamPartition changelog1SSP = new SystemStreamPartition(changelog1SystemStream, taskChangelogPartition);
    SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("0", "10", "11");
    Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream);

    String changelog1CheckpointedOffset = "5";
    CheckpointId checkpointId = CheckpointId.create();
    KafkaStateCheckpointMarker kafkaStateCheckpointMarker =
        new KafkaStateCheckpointMarker(changelog1SSP, changelog1CheckpointedOffset);
    ImmutableMap<String, KafkaStateCheckpointMarker> mockCheckpointedChangelogOffset =
        ImmutableMap.of(store1Name, kafkaStateCheckpointMarker);
    Map<SystemStreamPartition, SystemStreamPartitionMetadata> mockCurrentChangelogOffsets =
        ImmutableMap.of(changelog1SSP, changelog1SSPMetadata);

    SystemAdmins mockSystemAdmins = mock(SystemAdmins.class);
    SystemAdmin mockSystemAdmin = mock(SystemAdmin.class);
    when(mockSystemAdmins.getSystemAdmin(changelog1SSP.getSystem())).thenReturn(mockSystemAdmin);
    StorageManagerUtil mockStorageManagerUtil = mock(StorageManagerUtil.class);
    File mockLoggedStoreBaseDir = mock(File.class);
    File mockNonLoggedStoreBaseDir = mock(File.class);
    Config mockConfig = mock(Config.class);
    Clock mockClock = mock(Clock.class);

    File mockCurrentStoreDir = mock(File.class);
    File mockStoreNewerCheckpointDir = mock(File.class);
    File mockStoreOlderCheckpointDir = mock(File.class);
    when(mockStorageManagerUtil.getTaskStoreDir(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any()))
        .thenReturn(mockCurrentStoreDir);
    when(mockStorageManagerUtil.getTaskStoreCheckpointDirs(eq(mockLoggedStoreBaseDir), eq(store1Name), eq(taskName), any()))
        .thenReturn(ImmutableList.of(mockStoreNewerCheckpointDir, mockStoreOlderCheckpointDir));
    when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreNewerCheckpointDir), any(),
        eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true);
    when(mockStorageManagerUtil.isLoggedStoreValid(eq(store1Name), eq(mockStoreOlderCheckpointDir), any(),
        eq(mockStoreChangelogs), eq(mockTaskModel), any(), eq(mockStoreEngines))).thenReturn(true);
    Set<SystemStreamPartition> mockChangelogSSPs = ImmutableSet.of(changelog1SSP);
    when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreNewerCheckpointDir), eq(mockChangelogSSPs), eq(false)))
        .thenReturn(ImmutableMap.of(changelog1SSP, "10"));  // greater than checkpointed offset (5)
    when(mockStorageManagerUtil.readOffsetFile(eq(mockStoreOlderCheckpointDir), eq(mockChangelogSSPs), eq(false)))
        .thenReturn(ImmutableMap.of(changelog1SSP, changelog1CheckpointedOffset));

    Mockito.when(mockSystemAdmin.offsetComparator(anyString(), anyString()))
        .thenAnswer((Answer<Integer>) invocation -> {
          String offset1 = (String) invocation.getArguments()[0];
          String offset2 = (String) invocation.getArguments()[1];
          return Long.valueOf(offset1).compareTo(Long.valueOf(offset2));
        });

    StoreActions storeActions = TransactionalStateTaskRestoreManager.getStoreActions(
        mockTaskModel, mockStoreEngines, mockStoreChangelogs, mockCheckpointedChangelogOffset, checkpointId,
        mockCurrentChangelogOffsets, mockSystemAdmins, mockStorageManagerUtil,
        mockLoggedStoreBaseDir, mockNonLoggedStoreBaseDir, mockConfig, mockClock);

    // ensure that both the current dir and newer checkpoint dir are marked for deletion
    assertEquals(2, storeActions.storeDirsToDelete.get(store1Name).size());
    assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockCurrentStoreDir));
    assertTrue(storeActions.storeDirsToDelete.get(store1Name).contains(mockStoreNewerCheckpointDir));
    // ensure that the older store checkpoint is marked for retention
    assertEquals(mockStoreOlderCheckpointDir, storeActions.storeDirsToRetain.get(store1Name));
    // ensure that we mark the store for restore even if local offset == checkpointed offset
    // this is required since there may be message we need to trim
    assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).startingOffset);
    assertEquals(changelog1CheckpointedOffset, storeActions.storesToRestore.get(store1Name).endingOffset);
  }
This can happen if container failed after checkpointing store but before writing newest changelog offset to
checkpoint topic. In this case, the previously checkpointed (older) store directory should be used.
samza-core/src/test/java/org/apache/samza/storage/TestTransactionalStateTaskRestoreManager.java