@Test
  public void testCheckpointV2BasedRestoreFactoryCreation() {
    Set<String> storeNames = ImmutableSet.of("storeName0", "storeName1", "storeName2");

    StorageConfig mockConfig = mock(StorageConfig.class);
    when(mockConfig.getStoreRestoreFactories("storeName0"))
        .thenReturn(ImmutableList.of("factory0", "factory1", "factory2"));
    when(mockConfig.getStoreRestoreFactories("storeName1"))
        .thenReturn(ImmutableList.of("factory2", "factory1"));
    when(mockConfig.getStoreRestoreFactories("storeName2"))
        .thenReturn(Collections.emptyList());

    when(mockConfig.getChangelogStream("storeName0"))
        .thenReturn(Optional.empty());
    when(mockConfig.getChangelogStream("storeName1"))
        .thenReturn(Optional.of("changelog"));
    when(mockConfig.getChangelogStream("storeName2"))
        .thenReturn(Optional.of("changelog"));

    CheckpointV2 checkpointV2 = mock(CheckpointV2.class);
    when(checkpointV2.getVersion()).thenReturn((short) 2);
    when(checkpointV2.getStateCheckpointMarkers())
        .thenReturn(ImmutableMap.of(
            "factory0", ImmutableMap.of("storeName0", "", "storeName1", "", "storeName2", ""),
            "factory1", ImmutableMap.of("storeName0", "", "storeName1", "", "storeName2", ""),
            "factory2", ImmutableMap.of("storeName0", "", "storeName1", "", "storeName2", "")));

    Map<String, Set<String>> factoriesToStores =
        ContainerStorageManagerUtil.getBackendFactoryStoreNames(storeNames, checkpointV2, mockConfig);
    Assert.assertEquals(2, factoriesToStores.size());
    Assert.assertEquals(ImmutableSet.of("storeName0"),
        factoriesToStores.get("factory0"));
    Assert.assertEquals(ImmutableSet.of("storeName1"),
        factoriesToStores.get("factory2"));

    when(checkpointV2.getStateCheckpointMarkers())
        .thenReturn(ImmutableMap.of(
            "factory2", ImmutableMap.of("storeName0", "", "storeName1", "", "storeName2", "")));
    factoriesToStores = ContainerStorageManagerUtil.getBackendFactoryStoreNames(storeNames, checkpointV2, mockConfig);
    Assert.assertEquals(1, factoriesToStores.size());
    Assert.assertEquals(ImmutableSet.of("storeName1", "storeName0"),
        factoriesToStores.get("factory2"));

    when(checkpointV2.getStateCheckpointMarkers())
        .thenReturn(ImmutableMap.of(
            "factory1", ImmutableMap.of("storeName0", "", "storeName1", "", "storeName2", ""),
            "factory2", ImmutableMap.of("storeName0", "", "storeName1", "", "storeName2", "")));
    factoriesToStores = ContainerStorageManagerUtil.getBackendFactoryStoreNames(storeNames, checkpointV2, mockConfig);
    Assert.assertEquals(2, factoriesToStores.size());
    Assert.assertEquals(ImmutableSet.of("storeName0"),
        factoriesToStores.get("factory1"));
    Assert.assertEquals(ImmutableSet.of("storeName1"),
        factoriesToStores.get("factory2"));

    when(checkpointV2.getStateCheckpointMarkers())
        .thenReturn(ImmutableMap.of(
            "factory1", ImmutableMap.of("storeName0", "", "storeName1", "", "storeName2", ""),
            "factory2", ImmutableMap.of("storeName0", "", "storeName2", "")));
    factoriesToStores = ContainerStorageManagerUtil.getBackendFactoryStoreNames(storeNames, checkpointV2, mockConfig);
    Assert.assertEquals(1, factoriesToStores.size());
    Assert.assertEquals(ImmutableSet.of("storeName0", "storeName1"),
        factoriesToStores.get("factory1"));

    when(checkpointV2.getStateCheckpointMarkers())
        .thenReturn(ImmutableMap.of(
            "factory1", ImmutableMap.of("storeName0", "", "storeName1", "", "storeName2", "")));
    factoriesToStores = ContainerStorageManagerUtil.getBackendFactoryStoreNames(storeNames, checkpointV2, mockConfig);
    Assert.assertEquals(1, factoriesToStores.size());
    Assert.assertEquals(ImmutableSet.of("storeName0", "storeName1"),
        factoriesToStores.get("factory1"));

    when(checkpointV2.getStateCheckpointMarkers())
        .thenReturn(Collections.emptyMap());
    factoriesToStores = ContainerStorageManagerUtil.getBackendFactoryStoreNames(storeNames, checkpointV2, mockConfig);
    Assert.assertEquals(2, factoriesToStores.size());
    Assert.assertEquals(ImmutableSet.of("storeName0"),
        factoriesToStores.get("factory0"));
    Assert.assertEquals(ImmutableSet.of("storeName1"),
        factoriesToStores.get("factory2"));

    when(checkpointV2.getStateCheckpointMarkers())
        .thenReturn(ImmutableMap.of(
            "factory0", ImmutableMap.of("storeName1", "", "storeName2", ""),
            "factory1", ImmutableMap.of("storeName1", "", "storeName2", ""),
            "factory2", ImmutableMap.of("storeName0", "", "storeName2", "")));
    factoriesToStores = ContainerStorageManagerUtil.getBackendFactoryStoreNames(storeNames, checkpointV2, mockConfig);
    Assert.assertEquals(2, factoriesToStores.size());
    Assert.assertEquals(ImmutableSet.of("storeName1"),
        factoriesToStores.get("factory1"));
    Assert.assertEquals(ImmutableSet.of("storeName0"),
        factoriesToStores.get("factory2"));
  }
This test will attempt to verify if logged stores are deleted if the input checkpoints are empty.
samza-core/src/test/scala/org/apache/samza/storage/TestContainerStorageManager.java