@Test
  public void testGetStoreActionsForLoggedNonPersistentStore_FullRestoreIfCheckpointedOffsetOlderThanOldest() {
    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(false); // non-persistent store
    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);
    // checkpointed changelog offset > newest offset (e.g. changelog topic got changed)
    SystemStreamPartitionMetadata changelog1SSPMetadata = new SystemStreamPartitionMetadata("10", "20", "21");
    Map<String, SystemStream> mockStoreChangelogs = ImmutableMap.of(store1Name, changelog1SystemStream);

    String changelog1CheckpointedOffset = "5";
    CheckpointId checkpointId = CheckpointId.create();
    KafkaStateCheckpointMarker kafkaStateCheckpointMarker =
        new KafkaStateCheckpointMarker(changelog1SSP, changelog1CheckpointedOffset);
    Map<String, KafkaStateCheckpointMarker> mockCheckpointedChangelogOffset =
        new HashMap<String, KafkaStateCheckpointMarker>() { {
          put(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);

    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 there is nothing to retain or delete
    assertEquals(0, storeActions.storeDirsToDelete.size());
    assertEquals(0, storeActions.storeDirsToRetain.size());
    // ensure that we mark the store for full restore (from current oldest to current newest)
    assertEquals("10", storeActions.storesToRestore.get(store1Name).startingOffset);
    assertEquals("20", storeActions.storesToRestore.get(store1Name).endingOffset);
  }
This can happen if the changelog topic gets compacted and the local store offset was written prior to the
compaction. If so, we do a full restore.
samza-core/src/test/java/org/apache/samza/storage/TestTransactionalStateTaskRestoreManager.java