@Test
  public void testInitRecoversFromDeletedException() {
    TaskName taskName = new TaskName("task");
    Set<String> stores = Collections.singleton("store");

    BlobStoreRestoreManager taskRestoreManager = mock(BlobStoreRestoreManager.class);
    Throwable deletedException = new SamzaException(new CompletionException(new DeletedException("410 gone")));
    doThrow(deletedException).when(taskRestoreManager).init(any(Checkpoint.class));
    when(taskRestoreManager.restore()).thenReturn(CompletableFuture.completedFuture(null));
    when(taskRestoreManager.restore(true)).thenReturn(CompletableFuture.completedFuture(null));

    // mock ReflectionUtil.getObj
    PowerMockito.mockStatic(ReflectionUtil.class);
    BlobStoreManagerFactory blobStoreManagerFactory = mock(BlobStoreManagerFactory.class);
    BlobStoreManager blobStoreManager = mock(BlobStoreManager.class);
    PowerMockito.when(ReflectionUtil.getObj(anyString(), eq(BlobStoreManagerFactory.class)))
        .thenReturn(blobStoreManagerFactory);
    when(blobStoreManagerFactory.getRestoreBlobStoreManager(any(Config.class), any(ExecutorService.class)))
        .thenReturn(blobStoreManager);

    Map<String, TaskRestoreManager> storeTaskRestoreManager = ImmutableMap.of("store", taskRestoreManager);
    CheckpointManager checkpointManager = mock(CheckpointManager.class);
    JobContext jobContext = mock(JobContext.class);
    when(jobContext.getJobModel()).thenReturn(mock(JobModel.class));
    TaskModel taskModel = mock(TaskModel.class);
    when(taskModel.getTaskName()).thenReturn(new TaskName("test"));
    ContainerModel containerModel = mock(ContainerModel.class);
    when(containerModel.getTasks()).thenReturn(ImmutableMap.of(taskName, taskModel));
    Checkpoint checkpoint = mock(CheckpointV2.class);
    Map<TaskName, Checkpoint> taskCheckpoints = ImmutableMap.of(taskName, checkpoint);
    Map<TaskName, Map<String, Set<String>>> taskBackendFactoryToStoreNames =
        ImmutableMap.of(taskName, ImmutableMap.of(BlobStoreStateBackendFactory.class.getName(), stores));
    Config config = new MapConfig(ImmutableMap.of("job.name", "test"), ImmutableMap.of("stores.store.backup.factories", BlobStoreStateBackendFactory.class.getName()));
    ExecutorService executor = Executors.newSingleThreadExecutor();
    SystemConsumer systemConsumer = mock(SystemConsumer.class);

    ContainerStorageManagerRestoreUtil.initAndRestoreTaskInstances(ImmutableMap.of(taskName, storeTaskRestoreManager),
        samzaContainerMetrics, checkpointManager, jobContext, containerModel, taskCheckpoints,
        taskBackendFactoryToStoreNames, config, executor, new HashMap<>(), null,
        ImmutableMap.of("store", systemConsumer));

    // verify init() is called twice -> once without getDeleted flag, once with getDeleted flag
    verify(taskRestoreManager, times(1)).init(any(Checkpoint.class));
    verify(taskRestoreManager, times(1)).init(any(Checkpoint.class), anyBoolean());
    // verify restore is called with getDeletedFlag only
    verify(taskRestoreManager, times(0)).restore();
    verify(taskRestoreManager, times(1)).restore(true);
  }
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