@Test
public void successMigrateDocumentsWithAllVersionsWithDeleteAfter() throws Throwable {

    // this test dirties destination host, requires clean up
    clearSourceAndDestInProcessPeers = true;
    // disable idempotent post on destination
    for (VerificationHost host : destinationHost.getInProcessHostMap().values()) {
        host.toggleServiceOptions(UriUtils.buildUri(host, ExampleService.FACTORY_LINK),
                null, EnumSet.of(ServiceOption.IDEMPOTENT_POST));
    }
    // create docs in source host
    Map<String, DocumentVersionType> typeBySelfLink = createVersionedExampleDocuments(this.exampleSourceFactory, this.serviceCount);
    // create docs in destination with same self links
    // exclude DELETE
    List<String> selfLinks = typeBySelfLink.entrySet().stream()
            .filter(entry -> entry.getValue() != DocumentVersionType.POST_PUT_PATCH_DELETE)
            .map(Entry::getKey)
            .collect(toList());
    List<Operation> posts = new ArrayList<>();
    for (String selfLink : selfLinks) {
        ExampleServiceState state = new ExampleServiceState();
        state.name = "example-in-destination";
        state.documentSelfLink = selfLink;
        Operation post = Operation.createPost(getDestinationHost(), ExampleService.FACTORY_LINK).setBody(state);
        posts.add(post);
    }
    this.sender.sendAndWait(posts);
    // start migration. specify DELETE_AFTER option
    MigrationTaskService.State migrationState = validMigrationState(ExampleService.FACTORY_LINK);
    migrationState.migrationOptions = EnumSet.of(MigrationOption.ALL_VERSIONS, MigrationOption.DELETE_AFTER);
    Operation post = Operation.createPost(this.destinationFactoryUri).setBody(migrationState);
    ServiceDocument taskState = this.sender.sendAndWait(post, ServiceDocument.class);
    State finalState = waitForServiceCompletion(taskState.documentSelfLink, getDestinationHost());
    assertEquals(TaskStage.FINISHED, finalState.taskInfo.stage);
    validateVersionedExampleDocuments(typeBySelfLink);
    // num of processed doc should be same as successful migration.
    // see comments in "successMigrateDocumentsWithAllVersions()" for the detailed number
    long expectedProcessedDocs = this.serviceCount * (1 + 2 + 3 + 1 + 2 + 3);
    validateNumOfProcessedDoc(getDestinationHost(), taskState.documentSelfLink, expectedProcessedDocs);
}