@Test
  public void testRestoreRecoversFromDeletedException() throws Exception {
    TaskName taskName = new TaskName("task");
    String storeName = "store";
    Set<String> stores = Collections.singleton(storeName);
    String jobName = "job";
    String jobId = "jobId";

    BlobStoreManager blobStoreManager = mock(BlobStoreManager.class);
    doNothing().when(blobStoreManager).init();
    doNothing().when(blobStoreManager).close();

    BlobStoreRestoreManager taskRestoreManager = mock(BlobStoreRestoreManager.class);
    doAnswer(invocation -> {
      blobStoreManager.init();
      return null;
    }).when(taskRestoreManager).init(any(Checkpoint.class));
    doAnswer(invocation -> {
      blobStoreManager.close();
      return null;
    }).when(taskRestoreManager).close();

    CompletableFuture<Void> failedFuture = CompletableFuture.completedFuture(null)
        .thenCompose(v -> { throw new DeletedException("410 Gone"); });
    when(taskRestoreManager.restore()).thenReturn(failedFuture);
    when(taskRestoreManager.restore(true)).thenReturn(CompletableFuture.completedFuture(null));

    Map<String, TaskRestoreManager> factoryToTaskRestoreManager = ImmutableMap.of(
        BlobStoreStateBackendFactory.class.getName(), taskRestoreManager);

    JobContext jobContext = mock(JobContext.class);
    TaskModel taskModel = mock(TaskModel.class);
    when(taskModel.getTaskName()).thenReturn(taskName);
    when(taskModel.getTaskMode()).thenReturn(TaskMode.Active);

    ContainerModel containerModel = mock(ContainerModel.class);
    when(containerModel.getTasks()).thenReturn(ImmutableMap.of(taskName, taskModel));

    CheckpointV2 checkpoint = mock(CheckpointV2.class);
    when(checkpoint.getOffsets()).thenReturn(ImmutableMap.of());
    when(checkpoint.getCheckpointId()).thenReturn(CheckpointId.create());
    when(checkpoint.getStateCheckpointMarkers()).thenReturn(ImmutableMap.of(
        KafkaChangelogStateBackendFactory.class.getName(), new HashMap<>()));

    CheckpointManager checkpointManager = mock(CheckpointManager.class);
    when(checkpointManager.readLastCheckpoint(taskName)).thenReturn(checkpoint);

    String expectedOldBlobId = "oldBlobId";
    when(checkpoint.getStateCheckpointMarkers()).thenReturn(ImmutableMap.of(
        BlobStoreStateBackendFactory.class.getName(), ImmutableMap.of(storeName, expectedOldBlobId)));

    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(
        "blob.store.manager.factory", BlobStoreStateBackendFactory.class.getName(),
        "job.name", jobName));

    ExecutorService executor = Executors.newFixedThreadPool(5);

    SystemConsumer systemConsumer = mock(SystemConsumer.class);

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

    // To verify order of operations
    InOrder inOrder = inOrder(blobStoreManager, taskRestoreManager);

    // mock ContainerStorageManagerRestoreUtil.backupRecoveredStore
    String expectedBlobId = "blobId";
    PowerMockito.spy(ContainerStorageManagerRestoreUtil.class);
    PowerMockito.doReturn(CompletableFuture.completedFuture(ImmutableMap.of(storeName, expectedBlobId)))
        .when(ContainerStorageManagerRestoreUtil.class, "backupRecoveredStore",
            any(JobContext.class), any(ContainerModel.class), any(Config.class),
            any(TaskName.class), any(Set.class), any(Checkpoint.class), any(File.class),
            any(BlobStoreManager.class), any(MetricsRegistry.class), any(ExecutorService.class));

    SnapshotIndex snapshotIndex = new SnapshotIndex(System.currentTimeMillis(),
        new SnapshotMetadata(CheckpointId.create(), jobName, jobId, taskName.getTaskName(), storeName),
        new DirIndex("test", new ArrayList<>(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>()),
        Optional.empty());

    ArgumentCaptor<String> getBlobIdCaptor = ArgumentCaptor.forClass(String.class);
    ArgumentCaptor<ByteArrayOutputStream> outputStreamCaptor = ArgumentCaptor.forClass(ByteArrayOutputStream.class);
    when(blobStoreManager.get(getBlobIdCaptor.capture(), outputStreamCaptor.capture(),
        any(Metadata.class), any(Boolean.class)))
        .thenAnswer(invocation -> {
          ByteArrayOutputStream outputStream = outputStreamCaptor.getValue();
          outputStream.write(new SnapshotIndexSerde().toBytes(snapshotIndex));
          return CompletableFuture.completedFuture(null);
        });

    ArgumentCaptor<String> removeTTLBlobIdCaptor = ArgumentCaptor.forClass(String.class);
    when(blobStoreManager.removeTTL(removeTTLBlobIdCaptor.capture(), any(Metadata.class)))
        .thenAnswer(invocation -> CompletableFuture.completedFuture(null));

    ArgumentCaptor<String> deleteBlobIdCaptor = ArgumentCaptor.forClass(String.class);
    when(blobStoreManager.delete(deleteBlobIdCaptor.capture(), any(Metadata.class)))
        .thenAnswer(invocation -> CompletableFuture.completedFuture(null));

    CompletableFuture<Map<TaskName, Checkpoint>> updatedTaskCheckpointsFuture =
        ContainerStorageManagerRestoreUtil.initAndRestoreTaskInstances(ImmutableMap.of(taskName, factoryToTaskRestoreManager),
            samzaContainerMetrics, checkpointManager, jobContext, containerModel, taskCheckpoints,
            taskBackendFactoryToStoreNames, config, executor, new HashMap<>(), null,
            ImmutableMap.of(storeName, systemConsumer));

    // verify close is not called until init and restore futures are complete
    verify(taskRestoreManager, never()).close();
    Map<TaskName, Checkpoint> updatedTaskCheckpoints = updatedTaskCheckpointsFuture.get();
    // verify close is called only once after restore future is complete
    verify(taskRestoreManager, times(1)).close();

    // verify taskCheckpoint is updated
    assertNotEquals(((CheckpointV2) taskCheckpoints.get(taskName)).getCheckpointId(),
        ((CheckpointV2) updatedTaskCheckpoints.get(taskName)).getCheckpointId());

    // verify init is not retried with getDeleted
    verify(taskRestoreManager, times(0)).init(any(Checkpoint.class), anyBoolean());

    // verify restore is call twice - once without getDeleted flag, once with getDeleted flag
    verify(taskRestoreManager, times(1)).restore();
    verify(taskRestoreManager, times(1)).restore(true);

    // verify the GET and removeTTL was called on the new SnapshotIndex
    assertEquals(expectedBlobId, getBlobIdCaptor.getAllValues().get(0));
    assertEquals(expectedBlobId, removeTTLBlobIdCaptor.getAllValues().get(0));

    // verify that GET and delete was called on the old SnapshotIndex
    assertEquals(expectedOldBlobId, getBlobIdCaptor.getAllValues().get(1));
    assertEquals(expectedOldBlobId, deleteBlobIdCaptor.getValue());

    // verify the order of operations in taskRestoreManager and blobStoreManager
    // Verifies that close is called after restore(true)
    inOrder.verify(taskRestoreManager).init(any(Checkpoint.class));
    inOrder.verify(blobStoreManager).init(); // init called on blobStoreManager passed to taskRestoreManager
    inOrder.verify(taskRestoreManager).restore();
    inOrder.verify(blobStoreManager).init(); // init called on blobStoreManager created in ContainerStorageManagerRestoreUtil#restoreDeletedSnapshot
    inOrder.verify(taskRestoreManager).restore(true);
    inOrder.verify(blobStoreManager).close(); // close called on blobStoreManager created in ContainerStorageManagerRestoreUtil#restoreDeletedSnapshot
    inOrder.verify(blobStoreManager).close(); // close called on blobStoreManager passed to taskRestoreManager
  }
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