public Flux<StorageDescriptor> deleteMedia(final AuthenticatedBackupUser backupUser,
      final List<StorageDescriptor> storageDescriptors) {
    checkBackupLevel(backupUser, BackupLevel.MESSAGES);

    // Check for a cdn we don't know how to process
    if (storageDescriptors.stream().anyMatch(sd -> sd.cdn() != remoteStorageManager.cdnNumber())) {
      throw Status.INVALID_ARGUMENT
          .withDescription("unsupported media cdn provided")
          .asRuntimeException();
    }

    return Flux.usingWhen(

        // Gather usage updates into the UsageBatcher to apply during the cleanup operation
        Mono.just(new UsageBatcher()),

        // Deletes the objects, returning their former location. Tracks bytes removed so the quota can be updated on
        // completion
        batcher -> Flux.fromIterable(storageDescriptors)
            .flatMapSequential(sd -> Mono
                // Delete the object
                .fromCompletionStage(remoteStorageManager.delete(cdnMediaPath(backupUser, sd.key())))
                // Track how much the remote storage manager indicated was deleted as part of the operation
                .doOnNext(deletedBytes -> batcher.update(-deletedBytes))
                .thenReturn(sd), DELETION_CONCURRENCY),

        // On cleanup, update the quota using whatever updates were accumulated in the batcher
        batcher ->
            Mono.fromFuture(backupsDb.trackMedia(backupUser, batcher.countDelta.get(), batcher.usageDelta.get())));
  }
@Test
  public void deleteUnknownCdn() {
    final AuthenticatedBackupUser backupUser = backupUser(TestRandomUtil.nextBytes(16), BackupLevel.MEDIA);
    final BackupManager.StorageDescriptor sd = new BackupManager.StorageDescriptor(4, TestRandomUtil.nextBytes(15));
    when(remoteStorageManager.cdnNumber()).thenReturn(5);
    assertThatThrownBy(() ->
        backupManager.deleteMedia(backupUser, List.of(sd)).then().block())
        .isInstanceOf(StatusRuntimeException.class)
        .matches(e -> ((StatusRuntimeException) e).getStatus().getCode() == Status.INVALID_ARGUMENT.getCode());
  }