import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
/**
 * Perform phase two of the recovery process.
 * <p>
 * Phase two uses a snapshot of the current translog *without* acquiring the write lock (however, the translog snapshot is
 * point-in-time view of the translog). It then sends each translog operation to the target node so it can be replayed into the new
 * shard.
 *
 * @param startingSeqNo the sequence number to start recovery from, or {@link SequenceNumbersService#UNASSIGNED_SEQ_NO} if all
 *                      ops should be sent
 * @param snapshot      a snapshot of the translog
 */
void phase2(final long startingSeqNo, final Translog.Snapshot snapshot) throws IOException {
    if (shard.state() == IndexShardState.CLOSED) {
        throw new IndexShardClosedException(request.shardId());
    }
    cancellableThreads.checkForCancel();

    final StopWatch stopWatch = new StopWatch().start();

    logger.trace("recovery [phase2]: sending transaction log operations");

    // send all the snapshot's translog operations to the target
    final int totalOperations = sendSnapshot(startingSeqNo, snapshot);

    stopWatch.stop();
    logger.trace("recovery [phase2]: took [{}]", stopWatch.totalTime());
    response.phase2Time = stopWatch.totalTime().millis();
    response.phase2Operations = totalOperations;
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
protected void prepareTargetForTranslog() {
    StopWatch stopWatch = new StopWatch().start();
    logger.trace("{} recovery [phase1] to {}: prepare remote engine for translog", request.shardId(), request.targetNode());
    final long startEngineStart = stopWatch.totalTime().millis();
    cancellableThreads.execute(new Interruptable() {
        @Override
        public void run() throws InterruptedException {
            // Send a request preparing the new shard's translog to receive
            // operations. This ensures the shard engine is started and disables
            // garbage collection (not the JVM's GC!) of tombstone deletes
            transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.PREPARE_TRANSLOG,
                    new RecoveryPrepareForTranslogOperationsRequest(request.recoveryId(), request.shardId(), 0),
                    TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(), EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
        }
    });

    stopWatch.stop();

    response.startTime = stopWatch.totalTime().millis() - startEngineStart;
    logger.trace("{} recovery [phase1] to {}: remote engine start took [{}]",
            request.shardId(), request.targetNode(), stopWatch.totalTime());
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
protected void prepareTargetForTranslog(final Translog.View translogView) {
    StopWatch stopWatch = new StopWatch().start();
    logger.trace("{} recovery [phase1] to {}: prepare remote engine for translog", request.shardId(), request.targetNode());
    final long startEngineStart = stopWatch.totalTime().millis();
    cancellableThreads.execute(new Interruptable() {
        @Override
        public void run() throws InterruptedException {
            // Send a request preparing the new shard's translog to receive
            // operations. This ensures the shard engine is started and disables
            // garbage collection (not the JVM's GC!) of tombstone deletes
            transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.PREPARE_TRANSLOG,
                    new RecoveryPrepareForTranslogOperationsRequest(request.recoveryId(), request.shardId(), translogView.totalOperations()),
                    TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(), EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
        }
    });

    stopWatch.stop();

    response.startTime = stopWatch.totalTime().millis() - startEngineStart;
    logger.trace("{} recovery [phase1] to {}: remote engine start took [{}]",
            request.shardId(), request.targetNode(), stopWatch.totalTime());
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
/**
 * Perform phase2 of the recovery process
 * <p/>
 * Phase2 takes a snapshot of the current translog *without* acquiring the
 * write lock (however, the translog snapshot is a point-in-time view of
 * the translog). It then sends each translog operation to the target node
 * so it can be replayed into the new shard.
 */
public void phase2(Translog.Snapshot snapshot) {
    if (shard.state() == IndexShardState.CLOSED) {
        throw new IndexShardClosedException(request.shardId());
    }
    cancellableThreads.checkForCancel();

    StopWatch stopWatch = new StopWatch().start();

    logger.trace("{} recovery [phase2] to {}: sending transaction log operations", request.shardId(), request.targetNode());
    // Send all the snapshot's translog operations to the target
    int totalOperations = sendSnapshot(snapshot);
    stopWatch.stop();
    logger.trace("{} recovery [phase2] to {}: took [{}]", request.shardId(), request.targetNode(), stopWatch.totalTime());
    response.phase2Time = stopWatch.totalTime().millis();
    response.phase2Operations = totalOperations;
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
/**
 * Perform phase2 of the recovery process
 * <p>
 * Phase2 takes a snapshot of the current translog *without* acquiring the
 * write lock (however, the translog snapshot is a point-in-time view of
 * the translog). It then sends each translog operation to the target node
 * so it can be replayed into the new shard.
 */
public void phase2(Translog.Snapshot snapshot) {
    if (shard.state() == IndexShardState.CLOSED) {
        throw new IndexShardClosedException(request.shardId());
    }
    cancellableThreads.checkForCancel();

    StopWatch stopWatch = new StopWatch().start();

    logger.trace("{} recovery [phase2] to {}: sending transaction log operations", request.shardId(), request.targetNode());
    // Send all the snapshot's translog operations to the target
    int totalOperations = sendSnapshot(snapshot);
    stopWatch.stop();
    logger.trace("{} recovery [phase2] to {}: took [{}]", request.shardId(), request.targetNode(), stopWatch.totalTime());
    response.phase2Time = stopWatch.totalTime().millis();
    response.phase2Operations = totalOperations;
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
protected OpenNlpService start() {
    StopWatch sw = new StopWatch("models-loading");
    Map<String, String> settingsMap = IngestOpenNlpPlugin.MODEL_FILE_SETTINGS.getAsMap(settings);
    for (Map.Entry<String, String> entry : settingsMap.entrySet()) {
        String name = entry.getKey();
        sw.start(name);
        Path path = configDirectory.resolve(entry.getValue());
        try (InputStream is = Files.newInputStream(path)) {
            nameFinderModels.put(name, new TokenNameFinderModel(is));
        } catch (IOException e) {
            logger.error((Supplier<?>) () -> new ParameterizedMessage("Could not load model [{}] with path [{}]", name, path), e);
        }
        sw.stop();
    }

    if (settingsMap.keySet().size() == 0) {
        logger.error("Did not load any models for ingest-opennlp plugin, none configured");
    } else {
        logger.info("Read models in [{}] for {}", sw.totalTime(), settingsMap.keySet());
    }

    return this;
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
private void check(Traversal traversal) {
    StopWatch sw = new StopWatch();
    int count = 0;
    sw.start();

    System.out.println("pre-strategy:" + traversal);
    traversal.hasNext();
    System.out.println("post-strategy:" + traversal);

    //traversal.profile().cap(TraversalMetrics.METRICS_KEY);

    while(traversal.hasNext()) {
        count ++;
        System.out.println(traversal.next());
    }

    sw.stop();
    System.out.println(sw.toString());
    System.out.println(count);
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
void prepareTargetForTranslog(final int totalTranslogOps, final long maxUnsafeAutoIdTimestamp) throws IOException {
    StopWatch stopWatch = new StopWatch().start();
    logger.trace("recovery [phase1]: prepare remote engine for translog");
    final long startEngineStart = stopWatch.totalTime().millis();
    // Send a request preparing the new shard's translog to receive operations. This ensures the shard engine is started and disables
    // garbage collection (not the JVM's GC!) of tombstone deletes.
    cancellableThreads.executeIO(() -> recoveryTarget.prepareForTranslogOperations(totalTranslogOps, maxUnsafeAutoIdTimestamp));
    stopWatch.stop();

    response.startTime = stopWatch.totalTime().millis() - startEngineStart;
    logger.trace("recovery [phase1]: remote engine start took [{}]", stopWatch.totalTime());
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
public void finalizeRecovery() {
    if (shard.state() == IndexShardState.CLOSED) {
        throw new IndexShardClosedException(request.shardId());
    }
    cancellableThreads.checkForCancel();
    StopWatch stopWatch = new StopWatch().start();
    logger.trace("finalizing recovery");
    cancellableThreads.execute(() -> {
        shard.markAllocationIdAsInSync(recoveryTarget.getTargetAllocationId());
        recoveryTarget.finalizeRecovery(shard.getGlobalCheckpoint());
    });

    if (request.isPrimaryRelocation()) {
        // in case of primary relocation we have to ensure that the cluster state on the primary relocation target has all
        // replica shards that have recovered or are still recovering from the current primary, otherwise replication actions
        // will not be send to these replicas. To accomplish this, first block new recoveries, then take version of latest cluster
        // state. This means that no new recovery can be completed based on information of a newer cluster state than the current one.
        try (Releasable ignored = delayNewRecoveries.apply("primary relocation hand-off in progress or completed for " + shardId)) {
            final long currentClusterStateVersion = currentClusterStateVersionSupplier.get();
            logger.trace("waiting on remote node to have cluster state with version [{}]", currentClusterStateVersion);
            cancellableThreads.execute(() -> recoveryTarget.ensureClusterStateVersion(currentClusterStateVersion));

            logger.trace("performing relocation hand-off");
            cancellableThreads.execute(() -> shard.relocated("to " + request.targetNode()));
        }
        /*
         * if the recovery process fails after setting the shard state to RELOCATED, both relocation source and
         * target are failed (see {@link IndexShard#updateRoutingEntry}).
         */
    }
    stopWatch.stop();
    logger.trace("finalizing recovery took [{}]", stopWatch.totalTime());
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
public void testSameAlias() throws Exception {
    logger.info("--> creating index [test]");
    assertAcked(prepareCreate("test").addMapping("type", "name", "type=text"));
    ensureGreen();

    logger.info("--> creating alias1 ");
    assertAcked((admin().indices().prepareAliases().addAlias("test", "alias1")));
    TimeValue timeout = TimeValue.timeValueSeconds(2);
    logger.info("--> recreating alias1 ");
    StopWatch stopWatch = new StopWatch();
    stopWatch.start();
    assertAcked((admin().indices().prepareAliases().addAlias("test", "alias1").setTimeout(timeout)));
    assertThat(stopWatch.stop().lastTaskTime().millis(), lessThan(timeout.millis()));

    logger.info("--> modifying alias1 to have a filter");
    stopWatch.start();
    assertAcked((admin().indices().prepareAliases().addAlias("test", "alias1", termQuery("name", "foo")).setTimeout(timeout)));
    assertThat(stopWatch.stop().lastTaskTime().millis(), lessThan(timeout.millis()));

    logger.info("--> recreating alias1 with the same filter");
    stopWatch.start();
    assertAcked((admin().indices().prepareAliases().addAlias("test", "alias1", termQuery("name", "foo")).setTimeout(timeout)));
    assertThat(stopWatch.stop().lastTaskTime().millis(), lessThan(timeout.millis()));

    logger.info("--> recreating alias1 with a different filter");
    stopWatch.start();
    assertAcked((admin().indices().prepareAliases().addAlias("test", "alias1", termQuery("name", "bar")).setTimeout(timeout)));
    assertThat(stopWatch.stop().lastTaskTime().millis(), lessThan(timeout.millis()));

    logger.info("--> verify that filter was updated");
    AliasMetaData aliasMetaData = ((AliasOrIndex.Alias) internalCluster().clusterService().state().metaData().getAliasAndIndexLookup().get("alias1")).getFirstAliasMetaData();
    assertThat(aliasMetaData.getFilter().toString(), equalTo("{\"term\":{\"name\":{\"value\":\"bar\",\"boost\":1.0}}}"));

    logger.info("--> deleting alias1");
    stopWatch.start();
    assertAcked((admin().indices().prepareAliases().removeAlias("test", "alias1").setTimeout(timeout)));
    assertThat(stopWatch.stop().lastTaskTime().millis(), lessThan(timeout.millis()));


}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
public void finalizeRecovery() {
    if (shard.state() == IndexShardState.CLOSED) {
        throw new IndexShardClosedException(request.shardId());
    }
    cancellableThreads.checkForCancel();
    StopWatch stopWatch = new StopWatch().start();
    logger.trace("[{}][{}] finalizing recovery to {}", indexName, shardId, request.targetNode());


    cancellableThreads.execute(new Interruptable() {
        @Override
        public void run() throws InterruptedException {
            // Send the FINALIZE request to the target node. The finalize request
            // clears unreferenced translog files, refreshes the engine now that
            // new segments are available, and enables garbage collection of
            // tombstone files. The shard is also moved to the POST_RECOVERY phase
            // during this time
            transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.FINALIZE,
                    new RecoveryFinalizeRecoveryRequest(request.recoveryId(), request.shardId()),
                    TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionLongTimeout()).build(),
                    EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
        }
    });


    if (request.markAsRelocated()) {
        // TODO what happens if the recovery process fails afterwards, we need to mark this back to started
        try {
            shard.relocated("to " + request.targetNode());
        } catch (IllegalIndexShardStateException e) {
            // we can ignore this exception since, on the other node, when it moved to phase3
            // it will also send shard started, which might cause the index shard we work against
            // to move be closed by the time we get to the the relocated method
        }
    }
    stopWatch.stop();
    logger.trace("[{}][{}] finalizing recovery to {}: took [{}]",
            indexName, shardId, request.targetNode(), stopWatch.totalTime());
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
/**
 * finalizes the recovery process
 */
public void finalizeRecovery() {
    if (shard.state() == IndexShardState.CLOSED) {
        throw new IndexShardClosedException(request.shardId());
    }
    cancellableThreads.checkForCancel();
    StopWatch stopWatch = new StopWatch().start();
    logger.trace("[{}][{}] finalizing recovery to {}", indexName, shardId, request.targetNode());


    cancellableThreads.execute(new Interruptable() {
        @Override
        public void run() throws InterruptedException {
            // Send the FINALIZE request to the target node. The finalize request
            // clears unreferenced translog files, refreshes the engine now that
            // new segments are available, and enables garbage collection of
            // tombstone files. The shard is also moved to the POST_RECOVERY phase
            // during this time
            transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.FINALIZE,
                    new RecoveryFinalizeRecoveryRequest(request.recoveryId(), request.shardId()),
                    TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionLongTimeout()).build(),
                    EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
        }
    });


    if (request.markAsRelocated()) {
        // TODO what happens if the recovery process fails afterwards, we need to mark this back to started
        try {
            shard.relocated("to " + request.targetNode());
        } catch (IllegalIndexShardStateException e) {
            // we can ignore this exception since, on the other node, when it moved to phase3
            // it will also send shard started, which might cause the index shard we work against
            // to move be closed by the time we get to the the relocated method
        }
    }
    stopWatch.stop();
    logger.trace("[{}][{}] finalizing recovery to {}: took [{}]",
            indexName, shardId, request.targetNode(), stopWatch.totalTime());
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
private void measureCollectTime() {
    final StopWatch stopWatch = new StopWatch(collectPhase.executionPhaseId() + ": " + collectPhase.name());
    stopWatch.start("starting collectors");
    listenableRowReceiver.finishFuture().addListener(new Runnable() {
        @Override
        public void run() {
            stopWatch.stop();
            logger.trace("Collectors finished: {}", stopWatch.shortSummary());
        }
    }, MoreExecutors.directExecutor());
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
@Override
public void run() {
    String filePath = settings.get(configParameter, "");
    if (filePath.length() == 0) {
        logger.error("OpenNLP property [{}] is not set.", configParameter);
        return;
    }

    File modelFile = new File(filePath);
    if (!modelFile.exists() || !modelFile.canRead()) {
        logger.error("Model file {} does not exist.", modelFile);
        return;
    }

    StopWatch sw = new StopWatch("Loading model " + filePath).start();
    try {
        finders.put(type, 
                new PooledTokenNameFinderModel(
                        new FileInputStream(modelFile)));
    } catch (IOException e) {
        logger.error("Error loading model file {}: {}", e, modelFile, e.getMessage());
    } finally {
        sw.stop();
    }
    logger.info("Loaded file {} in {}", modelFile, sw.totalTime());
    countDownLatch.countDown();
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
public void loadFinders() throws Exception {
    finders = new NameFinderME[names.length];
    StopWatch sw = new StopWatch("Loading models").start();
    for (int mi = 0; mi < names.length; mi++) {
        finders[mi] = new NameFinderME(
            new PooledTokenNameFinderModel(
                new FileInputStream(
                    new File("src/test/resources/models", "en-ner-"
                    + names[mi] + ".bin"))));
    }
    sw.stop();
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
public void phase1() throws Exception {
    logger.debug("[{}][{}] recovery [phase1] to {}: start",
        request.shardId().index().name(), request.shardId().id(), request.targetNode().getName());
    StopWatch stopWatch = new StopWatch().start();
    blobTransferTarget.startRecovery();
    blobTransferTarget.createActiveTransfersSnapshot();
    sendStartRecoveryRequest();

    final AtomicReference<Exception> lastException = new AtomicReference<Exception>();
    try {
        syncVarFiles(lastException);
    } catch (InterruptedException ex) {
        throw new ElasticsearchException("blob recovery phase1 failed", ex);
    }

    Exception exception = lastException.get();
    if (exception != null) {
        throw exception;
    }

    /**
     * as soon as the recovery starts the target node will receive PutChunkReplicaRequests
     * the target node will then request the bytes it is missing from the source node
     * (it is missing bytes from PutChunk/StartBlob requests that happened before the recovery)
     * here we need to block so that the target node has enough time to request the head chunks
     *
     * e.g.
     *      Target Node receives Chunk X with bytes 10-19
     *      Target Node requests bytes 0-9 from Source Node
     *      Source Node sends bytes 0-9
     *      Source Node sets transferTakenOver
     */

    blobTransferTarget.waitForGetHeadRequests(GET_HEAD_TIMEOUT, TimeUnit.SECONDS);
    blobTransferTarget.createActivePutHeadChunkTransfersSnapshot();

    /**
     * After receiving a getHeadRequest the source node starts to send HeadChunks to the target
     * wait for all PutHeadChunk-Runnables to finish before ending the recovery.
     */
    blobTransferTarget.waitUntilPutHeadChunksAreFinished();
    sendFinalizeRecoveryRequest();

    blobTransferTarget.stopRecovery();
    stopWatch.stop();
    logger.debug("[{}][{}] recovery [phase1] to {}: took [{}]",
        request.shardId().index().name(), request.shardId().id(), request.targetNode().getName(),
        stopWatch.totalTime());
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
public void setupIndex() {
    log("==== INDEX SETUP ====");
    try {
      client.admin().indices().create(createIndexRequest(PARENT_INDEX).mapping(PARENT_TYPE,
              "id", "type=string,index=not_analyzed,doc_values=true",
              "num", "type=integer,doc_values=true")).actionGet();
      client.admin().indices().create(createIndexRequest(CHILD_INDEX).mapping(CHILD_TYPE,
              "id", "type=string,index=not_analyzed,doc_values=true",
              "pid", "type=string,index=not_analyzed,doc_values=true",
              "num", "type=integer,doc_values=true")).actionGet();
        Thread.sleep(5000);

        StopWatch stopWatch = new StopWatch().start();

        log("Indexing [" + NUM_PARENTS + "] parent documents into [" + PARENT_INDEX + "]");
        log("Indexing [" + (NUM_PARENTS * NUM_CHILDREN_PER_PARENT) + "] child documents into [" + CHILD_INDEX + "]");
        int ITERS = NUM_PARENTS / BATCH_SIZE;
        int i = 1;
        int counter = 0;
        for (; i <= ITERS; i++) {
            BulkRequestBuilder request = client.prepareBulk();
            for (int j = 0; j < BATCH_SIZE; j++) {
                String parentId = Integer.toString(counter);
                counter++;
                request.add(Requests.indexRequest(PARENT_INDEX)
                        .type(PARENT_TYPE)
                        .id(parentId)
                        .source(parentSource(counter, "test" + counter)));

                for (int k = 0; k < NUM_CHILDREN_PER_PARENT; k++) {
                    String childId = parentId + "_" + k;
                    request.add(Requests.indexRequest(CHILD_INDEX)
                            .type(CHILD_TYPE)
                            .id(childId)
                            .source(childSource(childId, counter, "tag" + k)));
                }
            }

            BulkResponse response = request.execute().actionGet();
            if (response.hasFailures()) {
                log("Index Failures...");
            }

            if (((i * BATCH_SIZE) % 10000) == 0) {
                log("Indexed [" + (i * BATCH_SIZE) * (1 + NUM_CHILDREN_PER_PARENT) + "] took [" + stopWatch.stop().lastTaskTime() + "]");
                stopWatch.start();
            }
        }

        log("Indexing took [" + stopWatch.totalTime() + "]");
        log("TPS [" + (((double) (NUM_PARENTS * (1 + NUM_CHILDREN_PER_PARENT))) / stopWatch.totalTime().secondsFrac()) + "]");
    } catch (Exception e) {
        log("Indices exist, wait for green");
        waitForGreen();
    }

    client.admin().indices().prepareRefresh().execute().actionGet();
    log("Number of docs in index: " + client.prepareCount(PARENT_INDEX, CHILD_INDEX).setQuery(matchAllQuery()).execute().actionGet().getCount());
    log("");
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
public void setupIndex() {
    log("==== INDEX SETUP ====");
    try {
        client.admin().indices().create(createIndexRequest(PARENT_INDEX)).actionGet();
        client.admin().indices().create(createIndexRequest(CHILD_INDEX)).actionGet();
        Thread.sleep(5000);

        StopWatch stopWatch = new StopWatch().start();

        log("Indexing [" + NUM_PARENTS + "] parent documents into [" + PARENT_INDEX + "]");
        log("Indexing [" + (NUM_PARENTS * NUM_CHILDREN_PER_PARENT) + "] child documents into [" + CHILD_INDEX + "]");
        int ITERS = NUM_PARENTS / BATCH_SIZE;
        int i = 1;
        int counter = 0;
        for (; i <= ITERS; i++) {
            BulkRequestBuilder request = client.prepareBulk();
            for (int j = 0; j < BATCH_SIZE; j++) {
                String parentId = Integer.toString(counter);
                counter++;
                request.add(Requests.indexRequest(PARENT_INDEX)
                        .type(PARENT_TYPE)
                        .id(parentId)
                        .source(parentSource(counter, "test" + counter)));

                for (int k = 0; k < NUM_CHILDREN_PER_PARENT; k++) {
                    String childId = parentId + "_" + k;
                    request.add(Requests.indexRequest(CHILD_INDEX)
                            .type(CHILD_TYPE)
                            .id(childId)
                            .source(childSource(childId, counter, "tag" + k)));
                }
            }

            BulkResponse response = request.execute().actionGet();
            if (response.hasFailures()) {
                log("Index Failures...");
            }

            if (((i * BATCH_SIZE) % 10000) == 0) {
                log("Indexed [" + (i * BATCH_SIZE) * (1 + NUM_CHILDREN_PER_PARENT) + "] took [" + stopWatch.stop().lastTaskTime() + "]");
                stopWatch.start();
            }
        }

        log("Indexing took [" + stopWatch.totalTime() + "]");
        log("TPS [" + (((double) (NUM_PARENTS * (1 + NUM_CHILDREN_PER_PARENT))) / stopWatch.totalTime().secondsFrac()) + "]");
    } catch (Exception e) {
        log("Indices exist, wait for green");
        waitForGreen();
    }

    client.admin().indices().prepareRefresh().execute().actionGet();
    log("Number of docs in index: " + client.prepareCount(PARENT_INDEX, CHILD_INDEX).setQuery(matchAllQuery()).execute().actionGet().getCount());
    log("");
}
 

import org.elasticsearch.common.StopWatch; //导入依赖的package包/类
private void reset() {
    sw = new StopWatch().start();
    startTimestamp = System.currentTimeMillis() / 1000;
    bulk = client.prepareBulk();
}