@Test
  public void testGetSSIThrowsExceptionIfAnyNonIgnoredAsyncBlobStoreErrors() {
    String store = "storeName1";
    String otherStore = "storeName2";
    Set<String> storesToBackupOrRestore = new HashSet<>();
    storesToBackupOrRestore.add(store);
    storesToBackupOrRestore.add(otherStore);
    Checkpoint checkpoint = createCheckpointV2(BlobStoreStateBackendFactory.class.getName(),
        ImmutableMap.of(store, "snapshotIndexBlobId1", otherStore, "snapshotIndexBlobId2"));
    SnapshotIndex store1SnapshotIndex = mock(SnapshotIndex.class);
    BlobStoreUtil mockBlobStoreUtil = mock(BlobStoreUtil.class);
    when(mockBlobStoreUtil.getStoreSnapshotIndexes(anyString(), anyString(), anyString(),
        any(Checkpoint.class), anySetOf(String.class), anyBoolean())).thenCallRealMethod();
    RuntimeException nonIgnoredException = new RuntimeException();
    CompletableFuture<SnapshotIndex> failedFuture = FutureUtil.failedFuture(nonIgnoredException);
    when(mockBlobStoreUtil.getSnapshotIndex(eq("snapshotIndexBlobId1"), any(Metadata.class), anyBoolean())).thenReturn(
        FutureUtil.failedFuture(new DeletedException())); // should fail even if some errors are ignored
    when(mockBlobStoreUtil.getSnapshotIndex(eq("snapshotIndexBlobId2"), any(Metadata.class), anyBoolean())).thenReturn(failedFuture);

    try {
      mockBlobStoreUtil.getStoreSnapshotIndexes("testJobName", "testJobId", "taskName",
          checkpoint, storesToBackupOrRestore, false);
      fail("Should have thrown an exception");
    } catch (Exception e) {
      Throwable cause =
          FutureUtil.unwrapExceptions(CompletionException.class, FutureUtil.unwrapExceptions(SamzaException.class, e));
      assertEquals(nonIgnoredException, cause);
    }
  }
Tests related to {@link BlobStoreUtil#getStoreSnapshotIndexes}
samza-core/src/test/java/org/apache/samza/storage/blobstore/util/TestBlobStoreUtil.java