import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@Override
@SuppressForbidden(reason = "debug")
protected void stopInternal() {
    Releasables.close(serverOpenChannels, () -> {
        final List<Tuple<String, Future<?>>> serverBootstrapCloseFutures = new ArrayList<>(serverBootstraps.size());
        for (final Map.Entry<String, ServerBootstrap> entry : serverBootstraps.entrySet()) {
            serverBootstrapCloseFutures.add(
                Tuple.tuple(entry.getKey(), entry.getValue().config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS)));
        }
        for (final Tuple<String, Future<?>> future : serverBootstrapCloseFutures) {
            future.v2().awaitUninterruptibly();
            if (!future.v2().isSuccess()) {
                logger.debug(
                    (Supplier<?>) () -> new ParameterizedMessage(
                        "Error closing server bootstrap for profile [{}]", future.v1()), future.v2().cause());
            }
        }
        serverBootstraps.clear();

        if (bootstrap != null) {
            bootstrap.config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS).awaitUninterruptibly();
            bootstrap = null;
        }
    });
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@SuppressForbidden(reason = "system out is ok for a command line tool")
public static void main(String[] args) throws Exception {
    String type = args[0];
    AbstractBenchmark<?> benchmark = null;
    switch (type) {
        case "transport":
            benchmark = new TransportClientBenchmark();
            break;
        case "rest":
            benchmark = new RestClientBenchmark();
            break;
        default:
            System.err.println("Unknown client type [" + type + "]");
            System.exit(1);
    }
    benchmark.run(Arrays.copyOfRange(args, 1, args.length));
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@SuppressForbidden(reason = "system out is ok for a command line tool")
public final void run(String[] args) throws Exception {
    if (args.length < 1) {
        System.err.println("usage: [search|bulk]");
        System.exit(1);
    }
    switch (args[0]) {
        case "search":
            runSearchBenchmark(args);
            break;
        case "bulk":
            runBulkIndexBenchmark(args);
            break;
        default:
            System.err.println("Unknown benchmark type [" + args[0] + "]");
            System.exit(1);

    }

}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@SuppressForbidden(reason = "system out is ok for a command line tool")
private void runTrials(Runnable runner) {
    int totalWarmupTrialRuns = 1;
    for (int run = 1; run <= totalWarmupTrialRuns; run++) {
        System.out.println("======================");
        System.out.println(" Warmup trial run " + run + "/" + totalWarmupTrialRuns);
        System.out.println("======================");
        runner.run();
    }

    int totalTrialRuns = 5;
    for (int run = 1; run <= totalTrialRuns; run++) {
        System.out.println("================");
        System.out.println(" Trial run " + run + "/" + totalTrialRuns);
        System.out.println("================");

        runner.run();
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
/**
 * Requests a full GC and checks whether the GC did actually run after a request. It retries up to 5 times in case the GC did not
 * run in time.
 */
@SuppressForbidden(reason = "we need to request a system GC for the benchmark")
private void runGc() {
    long previousCollections = getTotalGcCount();
    int attempts = 0;
    do {
        // request a full GC ...
        System.gc();
        // ... and give GC a chance to run
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return;
        }
        attempts++;
    } while (previousCollections == getTotalGcCount() || attempts < 5);
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
/**
 * Netty wants to do some unsafe things like use unsafe and replace a private field. This method disables these things by default, but
 * can be overridden by setting the corresponding system properties.
 */
@SuppressForbidden(reason = "set system properties to configure Netty")
private static void initializeNetty() {
    final String noUnsafeKey = "io.netty.noUnsafe";
    final String noUnsafe = System.getProperty(noUnsafeKey);
    if (noUnsafe == null) {
        // disable Netty from using unsafe
        // while permissions are needed to set this, if a security exception is thrown the permission needed can either be granted or
        // the system property can be set directly before starting the JVM; therefore, we do not catch a security exception here
        System.setProperty(noUnsafeKey, Boolean.toString(true));
    }

    final String noKeySetOptimizationKey = "io.netty.noKeySetOptimization";
    final String noKeySetOptimization = System.getProperty(noKeySetOptimizationKey);
    if (noKeySetOptimization == null) {
        // disable Netty from replacing the selector key set
        // while permissions are needed to set this, if a security exception is thrown the permission needed can either be granted or
        // the system property can be set directly before starting the JVM; therefore, we do not catch a security exception here
        System.setProperty(noKeySetOptimizationKey, Boolean.toString(true));
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@SuppressForbidden(reason = "manipulates system properties for testing")
public void testMaxNumShards() {
    IllegalArgumentException exception = expectThrows(IllegalArgumentException.class, () ->
        IndexMetaData.buildNumberOfShardsSetting()
            .get(Settings.builder().put("index.number_of_shards", 1025).build()));
    assertEquals("Failed to parse value [1025] for setting [index.number_of_shards] must be <= 1024", exception.getMessage());

    Integer numShards = IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.get(Settings.builder().put("index.number_of_shards", 100).build());
    assertEquals(100, numShards.intValue());
    int limit = randomIntBetween(1, 10);
    System.setProperty("es.index.max_number_of_shards", Integer.toString(limit));
    try {
        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () ->
            IndexMetaData.buildNumberOfShardsSetting()
                .get(Settings.builder().put("index.number_of_shards", 11).build()));
        assertEquals("Failed to parse value [11] for setting [index.number_of_shards] must be <= " + limit, e.getMessage());
    } finally {
        System.clearProperty("es.index.max_number_of_shards");
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
/**
 * Returns a new FileSystem to read REST resources, or null if they
 * are available from classpath.
 */
@SuppressForbidden(reason = "proper use of URL, hack around a JDK bug")
protected static FileSystem getFileSystem() throws IOException {
    // REST suite handling is currently complicated, with lots of filtering and so on
    // For now, to work embedded in a jar, return a ZipFileSystem over the jar contents.
    URL codeLocation = FileUtils.class.getProtectionDomain().getCodeSource().getLocation();
    boolean loadPackaged = RandomizedTest.systemPropertyAsBoolean(REST_LOAD_PACKAGED_TESTS, true);
    if (codeLocation.getFile().endsWith(".jar") && loadPackaged) {
        try {
            // hack around a bug in the zipfilesystem implementation before java 9,
            // its checkWritable was incorrect and it won't work without write permissions.
            // if we add the permission, it will open jars r/w, which is too scary! so copy to a safe r-w location.
            Path tmp = Files.createTempFile(null, ".jar");
            try (InputStream in = FileSystemUtils.openFileURLStream(codeLocation)) {
                Files.copy(in, tmp, StandardCopyOption.REPLACE_EXISTING);
            }
            return FileSystems.newFileSystem(new URI("jar:" + tmp.toUri()), Collections.emptyMap());
        } catch (URISyntaxException e) {
            throw new IOException("couldn't open zipfilesystem: ", e);
        }
    } else {
        return null;
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@Test
@SuppressForbidden(reason = "repeat is a feature here")
@Repeat(iterations = 10, useConstantSeed = true)
public void testReproducible() throws IOException {
    if (ITER++ == 0) {
        CLUSTER_SEED = cluster().seed();
        for (int i = 0; i < SEQUENCE.length; i++) {
            SEQUENCE[i] = randomLong();
        }
    } else {
        assertEquals(CLUSTER_SEED, Long.valueOf(cluster().seed()));
        for (int i = 0; i < SEQUENCE.length; i++) {
            assertThat(SEQUENCE[i], equalTo(randomLong()));
        }
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@SuppressForbidden(reason = "access temp directory for building index")
protected static synchronized FSDirectory getTempDir() {
    if (tmpBuildDir == null) {
        // Lazy init
        String tempDirPath = System.getProperty("java.io.tmpdir");
        if (tempDirPath == null)  {
            throw new RuntimeException("Java has no temporary folder property (java.io.tmpdir)?");
        }
        try {
            tmpBuildDir = FSDirectory.open(PathUtils.get(tempDirPath));
        } catch (IOException ioe) {
            throw new RuntimeException(ioe);
        }
    }
    return tmpBuildDir;
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
/** Downloads a zip from the url, as well as a SHA1 checksum, and checks the checksum. */
@SuppressForbidden(reason = "We use openStream to download plugins")
private Path downloadZipAndChecksum(Terminal terminal, String urlString, Path tmpDir) throws Exception {
    Path zip = downloadZip(terminal, urlString, tmpDir);
    pathsToDeleteOnShutdown.add(zip);
    URL checksumUrl = new URL(urlString + ".sha1");
    final String expectedChecksum;
    try (InputStream in = checksumUrl.openStream()) {
        BufferedReader checksumReader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8));
        expectedChecksum = checksumReader.readLine();
        if (checksumReader.readLine() != null) {
            throw new UserException(ExitCodes.IO_ERROR, "Invalid checksum file at " + checksumUrl);
        }
    }

    byte[] zipbytes = Files.readAllBytes(zip);
    String gotChecksum = MessageDigests.toHexString(MessageDigests.sha1().digest(zipbytes));
    if (expectedChecksum.equals(gotChecksum) == false) {
        throw new UserException(ExitCodes.IO_ERROR, "SHA1 mismatch, expected " + expectedChecksum + " but got " + gotChecksum);
    }

    return zip;
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@SuppressForbidden(reason = "System.out.*")
public void testSegmentInfosTracing() {
    // Defaults to not hooking up std out
    assertNull(SegmentInfos.getInfoStream());

    try {
        // False means don't hook up std out
        NodeEnvironment.applySegmentInfosTrace(
                Settings.builder().put(NodeEnvironment.ENABLE_LUCENE_SEGMENT_INFOS_TRACE_SETTING.getKey(), false).build());
        assertNull(SegmentInfos.getInfoStream());

        // But true means hook std out up statically
        NodeEnvironment.applySegmentInfosTrace(
                Settings.builder().put(NodeEnvironment.ENABLE_LUCENE_SEGMENT_INFOS_TRACE_SETTING.getKey(), true).build());
        assertEquals(System.out, SegmentInfos.getInfoStream());
    } finally {
        // Clean up after ourselves
        SegmentInfos.setInfoStream(null);
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
/**
 * @param type the ec2 hostname type to discover.
 * @return the appropriate host resolved from ec2 meta-data, or null if it cannot be obtained.
 * @see CustomNameResolver#resolveIfPossible(String)
 */
@SuppressForbidden(reason = "We call getInputStream in doPrivileged and provide SocketPermission")
public InetAddress[] resolve(Ec2HostnameType type) throws IOException {
    InputStream in = null;
    String metadataUrl = AwsEc2ServiceImpl.EC2_METADATA_URL + type.ec2Name;
    try {
        URL url = new URL(metadataUrl);
        logger.debug("obtaining ec2 hostname from ec2 meta-data url {}", url);
        URLConnection urlConnection = SocketAccess.doPrivilegedIOException(url::openConnection);
        urlConnection.setConnectTimeout(2000);
        in = SocketAccess.doPrivilegedIOException(urlConnection::getInputStream);
        BufferedReader urlReader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8));

        String metadataResult = urlReader.readLine();
        if (metadataResult == null || metadataResult.length() == 0) {
            throw new IOException("no gce metadata returned from [" + url + "] for [" + type.configName + "]");
        }
        // only one address: because we explicitly ask for only one via the Ec2HostnameType
        return new InetAddress[] { InetAddress.getByName(metadataResult) };
    } catch (IOException e) {
        throw new IOException("IOException caught when fetching InetAddress from [" + metadataUrl + "]", e);
    } finally {
        IOUtils.closeWhileHandlingException(in);
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
/**
 * Returns true if the path is writable.
 * Acts just like {@link Files#isWritable(Path)}, except won't
 * falsely return false for paths on SUBST'd drive letters
 * See https://bugs.openjdk.java.net/browse/JDK-8034057
 * Note this will set the file modification time (to its already-set value)
 * to test access.
 */
@SuppressForbidden(reason = "works around https://bugs.openjdk.java.net/browse/JDK-8034057")
public static boolean isWritable(Path path) throws IOException {
    boolean v = Files.isWritable(path);
    if (v || Constants.WINDOWS == false) {
        return v;
    }

    // isWritable returned false on windows, the hack begins!!!!!!
    // resetting the modification time is the least destructive/simplest
    // way to check for both files and directories, and fails early just
    // in getting the current value if file doesn't exist, etc
    try {
        Files.setLastModifiedTime(path, Files.getLastModifiedTime(path));
        return true;
    } catch (Throwable e) {
        return false;
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@SuppressForbidden(reason = "Math#abs is trappy")
private int generateShardId(ClusterState clusterState, String index, String type, String id, @Nullable String routing) {
    IndexMetaData indexMetaData = clusterState.metaData().index(index);
    if (indexMetaData == null) {
        throw new IndexNotFoundException(index);
    }
    final Version createdVersion = indexMetaData.getCreationVersion();
    final HashFunction hashFunction = indexMetaData.getRoutingHashFunction();
    final boolean useType = indexMetaData.getRoutingUseType();

    final int hash;
    if (routing == null) {
        if (!useType) {
            hash = hash(hashFunction, id);
        } else {
            hash = hash(hashFunction, type, id);
        }
    } else {
        hash = hash(hashFunction, routing);
    }
    if (createdVersion.onOrAfter(Version.V_2_0_0_beta1)) {
        return MathUtils.mod(hash, indexMetaData.getNumberOfShards());
    } else {
        return Math.abs(hash % indexMetaData.getNumberOfShards());
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
/**
 * Prepares the appender for use.
 */
@Override
@SuppressForbidden(reason = "System#out")
public void activateOptions() {
    if (follow) {
        if (target.equals(SYSTEM_ERR)) {
            setWriter(createWriter(new SystemErrStream()));
        } else {
            setWriter(createWriter(new SystemOutStream()));
        }
    } else {
        if (target.equals(SYSTEM_ERR)) {
            setWriter(createWriter(System.err));
        } else {
            setWriter(createWriter(System.out));
        }
    }

    super.activateOptions();
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
/** Adds access to classpath jars/classes for jar hell scan, etc */
@SuppressForbidden(reason = "accesses fully qualified URLs to configure security")
static void addClasspathPermissions(Permissions policy) throws IOException {
    // add permissions to everything in classpath
    // really it should be covered by lib/, but there could be e.g. agents or similar configured)
    for (URL url : JarHell.parseClassPath()) {
        Path path;
        try {
            path = PathUtils.get(url.toURI());
        } catch (URISyntaxException e) {
            throw new RuntimeException(e);
        }
        // resource itself
        policy.add(new FilePermission(path.toString(), "read,readlink"));
        // classes underneath
        if (Files.isDirectory(path)) {
            policy.add(new FilePermission(path.toString() + path.getFileSystem().getSeparator() + "-", "read,readlink"));
        }
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@SuppressWarnings("Convert2Lambda")
@Before
public void initializeLibraryAndDictionaryPaths() {
    String voikkoPath = System.getProperty("voikko.path");

    Path dictDirectory;
    String dictPath = System.getProperty("voikko.dict.path");
    if (dictPath != null) {
        dictDirectory = PathUtils.get(dictPath);
    } else if (voikkoPath != null) {
        dictDirectory = PathUtils.get(voikkoPath, "dicts");
    } else {
        assumeTrue("System property 'voikko.path' is not defined, add '-Dvoikko.path=/path/to/voikko'", false);
        return;
    }

    Path morphology = dictDirectory.resolve("2/mor-morpho/voikko-fi_FI.pro");

    AccessController.doPrivileged(new PrivilegedAction<Object>() {
        @Override
        @SuppressForbidden(reason = "We really want to access real file system")
        public Object run() {
            if (!morphology.toFile().isFile())
                fail("morphology file " + morphology + " does not exist");
            return null;
        }
    });

    settings.put("index.analysis.filter.myFilter.type", "voikko");

    if (voikkoPath != null)
        settings.put("index.analysis.filter.myFilter.libraryPath", voikkoPath);

    settings.put("index.analysis.filter.myFilter.dictionaryPath", dictDirectory.toAbsolutePath());
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@SuppressForbidden(reason = "We call connect in doPrivileged and provide SocketPermission")
private static InputStream getInputStream(URL url) throws IOException {
    try {
        return AccessController.doPrivileged((PrivilegedExceptionAction<InputStream>) url::openStream);
    } catch (PrivilegedActionException e) {
        throw (IOException) e.getCause();
    }
}
 

import org.elasticsearch.common.SuppressForbidden; //导入依赖的package包/类
@SuppressForbidden(reason = "system out is ok for a command line tool")
private void runBulkIndexBenchmark(String[] args) throws Exception {
    if (args.length != 7) {
        System.err.println(
            "usage: 'bulk' benchmarkTargetHostIp indexFilePath indexName typeName numberOfDocuments bulkSize");
        System.exit(1);
    }
    String benchmarkTargetHost = args[1];
    String indexFilePath = args[2];
    String indexName = args[3];
    String typeName = args[4];
    int totalDocs = Integer.valueOf(args[5]);
    int bulkSize = Integer.valueOf(args[6]);

    int totalIterationCount = (int) Math.floor(totalDocs / bulkSize);
    // consider 40% of all iterations as warmup iterations
    int warmupIterations = (int) (0.4d * totalIterationCount);
    int iterations = totalIterationCount - warmupIterations;

    T client = client(benchmarkTargetHost);

    BenchmarkRunner benchmark = new BenchmarkRunner(warmupIterations, iterations,
        new BulkBenchmarkTask(
            bulkRequestExecutor(client, indexName, typeName), indexFilePath, warmupIterations, iterations, bulkSize));

    try {
        runTrials(() -> {
            runGc();
            benchmark.run();
        });
    } finally {
        client.close();
    }

}