@Test
public void synchronizationAfterStaleHostRestart() throws Throwable {

    // This test verifies that if a stale host joins
    // a node-group after restart, synchronization updates
    // the local index with the latest state from peers
    this.isPeerSynchronizationEnabled = false;
    setUp(this.nodeCount);
    this.host.joinNodesAndVerifyConvergence(this.nodeCount);
    // create example services on one of the hosts, let replication do its thing
    VerificationHost peerHost = this.host.getPeerHost();
    Map<String, ExampleServiceState> exampleStatesPerSelfLink = createExampleServices(peerHost.getUri());
    // relax quorum, so we can perform updates after a host is stopped
    this.host.setNodeGroupQuorum(this.nodeCount - 1);
    // expire node that went away quickly to avoid a lot of log spam from gossip failures
    NodeGroupConfig cfg = new NodeGroupConfig();
    cfg.nodeRemovalDelayMicros = TimeUnit.SECONDS.toMicros(1);
    this.host.setNodeGroupConfig(cfg);
    // stop one of the hosts, preserve its index
    this.host.stopHostAndPreserveState(peerHost);
    // wait for stopped host to be removed from node group
    this.host.waitForNodeGroupConvergence(this.nodeCount - 1, this.nodeCount - 1);
    // update about half of the services
    VerificationHost existingHost = this.host.getPeerHost();
    Map<String, ExampleServiceState> servicesToUpdate = new HashMap<>();
    boolean update = false;
    for (Entry<String, ExampleServiceState> e : exampleStatesPerSelfLink.entrySet()) {
        update = !update;
        if (update) {
            servicesToUpdate.put(e.getKey(), e.getValue());
        }
    }
    doExampleServicePatch(servicesToUpdate, existingHost.getUri());
    // increase quorum on existing nodes, so they wait for restarted host to join
    this.host.setNodeGroupQuorum(this.nodeCount);
    // restart stopped host
    peerHost.setPort(0);
    peerHost.setSecurePort(0);
    peerHost.setPeerSynchronizationEnabled(false);
    assertTrue(VerificationHost.restartStatefulHost(peerHost, false));
    // join restarted host and wait for node group convergence
    this.host.addPeerNode(peerHost);
    this.host.joinNodesAndVerifyConvergence(this.nodeCount);
    // set quorum on new node as well
    this.host.setNodeGroupQuorum(this.nodeCount);
    this.host.waitForNodeGroupConvergence();
    // find example factory owner
    VerificationHost owner = this.host.getOwnerPeer(ExampleService.FACTORY_LINK);
    // wait for node selector availability before manually trigger a synchronization task
    this.host.waitFor("wait node selector available timeout", () -> {
        Operation op = Operation.createGet(owner, ServiceUriPaths.DEFAULT_NODE_SELECTOR);
        NodeSelectorState nss = this.host.getTestRequestSender().sendAndWait(op, NodeSelectorState.class);
        if (nss.status == NodeSelectorState.Status.AVAILABLE) {
            return true;
        }
        return false;
    });
    // kick-off synchronization and wait for it to finish
    startSynchronizationTaskAndWait(owner,
            ExampleService.FACTORY_LINK, ExampleServiceState.class, 1L);
    this.host.waitForReplicatedFactoryChildServiceConvergence(
            getFactoriesPerNodeGroup(ExampleService.FACTORY_LINK),
            exampleStatesPerSelfLink,
            this.exampleStateConvergenceChecker,
            exampleStatesPerSelfLink.size(),
            0,
            this.replicationFactor);
    // Verify all factories return exactly the same documentVersion and
    // documentEpoch for each service.
    List<String> nonComparedLinks = new ArrayList<>(this.serviceCount);
    HashMap<String, ServiceDocument> services = new HashMap<>(this.serviceCount);
    for (VerificationHost h : this.host.getInProcessHostMap().values()) {
        ServiceDocumentQueryResult r = this.host.getFactoryState(
                UriUtils.buildExpandLinksQueryUri(UriUtils.buildUri(h, ExampleService.FACTORY_LINK)));
        for (Entry<String, Object> e : r.documents.entrySet()) {
            ServiceDocument newDoc = Utils.fromJson(e.getValue(), ServiceDocument.class);
            ServiceDocument prevDoc = services.get(e.getKey());
            if (prevDoc == null) {
                services.put(e.getKey(), newDoc);
                nonComparedLinks.add(e.getKey());
                continue;
            }
            assertTrue(newDoc.documentVersion == prevDoc.documentVersion &&
                    newDoc.documentEpoch == prevDoc.documentEpoch);
            nonComparedLinks.remove(e.getKey());
        }
    }
    // This is to make sure that the deleted services reflect on all the hosts, including
    // the restarted host.
    assertTrue(nonComparedLinks.isEmpty());
    // Verify no duplicate document versions have been created
    // (see https://www.pivotaltracker.com/n/projects/1471320/stories/149126897)
    URI queryTaskUri = UriUtils.buildUri(peerHost, ServiceUriPaths.CORE_LOCAL_QUERY_TASKS);
    long actualTotalVersionCount = 0;
    for (String selfLink : services.keySet()) {
        Query query = Query.Builder.create()
                .addFieldClause(ServiceDocument.FIELD_NAME_SELF_LINK, selfLink)
                .build();
        QueryTask queryTask = QueryTask.Builder.createDirectTask()
                .setQuery(query)
                .addOption(QueryOption.INCLUDE_ALL_VERSIONS)
                .addOption(QueryOption.COUNT)
                .build();
        Operation postQuery = Operation.createPost(queryTaskUri)
                .setBody(queryTask);
        QueryTask queryResponse = peerHost.getTestRequestSender().sendAndWait(postQuery, QueryTask.class);
        actualTotalVersionCount += queryResponse.results.documentCount.longValue();
    }
    long expectedTotalVersionCount = exampleStatesPerSelfLink.size() // contributed version 0
            + servicesToUpdate.size(); // contributed version 1
    assertEquals(expectedTotalVersionCount, actualTotalVersionCount);
}