docs: @Override
      public synchronized void init(DataGenerator dataGenerator) throws Exception {
          if (esConfig.getRestClientPort() == 443 && !esConfig.isHttps()) {
              throw new IllegalArgumentException(
                      "You must set the configuration property 'https' to true if you use the https default port");
          }
  
          Integer indexRollsPerHour = esConfig.getIndexRollsPerDay();
          if (indexRollsPerHour < 0 || indexRollsPerHour > MAX_INDEX_ROLLS_PER_HOUR) {
              throw new IllegalArgumentException(
                      "The configuration property 'indexRollsPerHour' must be > 0 and <= " + MAX_INDEX_ROLLS_PER_HOUR);
          }
  
          if (indexRollsPerHour > 0 && 60 % indexRollsPerHour != 0) {
              throw new IllegalArgumentException( "The configuration property 'indexRollsPerHour' must evenly divide 60");
          }
          if (esConfig.getBulkWriteBatchSize() < 0) {
              throw new IllegalArgumentException( "bulkWriteBatchSize can't be negative'");
          }
  
          RestClientBuilder.RequestConfigCallback callback = requestConfigBuilder -> {
              requestConfigBuilder.setConnectTimeout(esConfig.getConnectTimeoutSeconds() * 1000);
              requestConfigBuilder.setConnectionRequestTimeout(esConfig.getConnectionRequestTimeoutSeconds() * 1000);
              requestConfigBuilder.setSocketTimeout(esConfig.getSocketTimeoutSeconds()* 1000) ;
              return requestConfigBuilder;
          };
          List<HttpHost> endpoints = getEndpoints(discoverer, esConfig);
          HttpHost[] hosts = endpoints.toArray(new HttpHost[0]);
          restClient =
                  RestClient.builder(hosts).
                          setMaxRetryTimeoutMillis(esConfig.getMaxRetryTimeoutSeconds() * 1000).
                          setRequestConfigCallback( callback ).
                          build();
  
          String hostname = endpoints.get(0).getHostName();
          ES_HOST_PORT = String.format("%s://%s:%s", getScheme(), hostname, esConfig.getRestClientPort());
          ES_INDEX_TYPE_RESOURCE_PATH = String.format("/%s/%s", esConfig.getIndexName(), DEFAULT_DOC_TYPE);
          CONNECTION_INFO =
                  "Cluster: " + this.getClusterOrHostName() + "\n" +
                          "Test Index URL: " + ES_HOST_PORT + ES_INDEX_TYPE_RESOURCE_PATH;
  
          writer = new EsWriter(
                  esConfig.getIndexName(),
                  DEFAULT_DOC_TYPE,
                  esConfig.getBulkWriteBatchSize() > 0,
                  indexRollsPerHour,
                  esConfig.getBulkWriteBatchSize(), esConfig.isRandomizeStrings() ?
                  dataGenerator :
                  new FakeWordDictionaryBasedDataGenerator(dataGenerator, coreConfig.getDataSize()));
  
          if (coreConfig.isAutoTuneEnabled()) {
              this.autoTuner = new EsAutoTuner(
                      coreConfig.getAutoTuneRampPeriodMillisecs(),
                      coreConfig.getAutoTuneIncrementIntervalMillisecs(),
                      coreConfig.getWriteRateLimit(),
                      coreConfig.getAutoTuneFinalWriteRate(),
                      coreConfig.getAutoTuneWriteFailureRatioThreshold());
          } else {
              // OK if it is null because it will never be used if ! isAutoTuneEnabled
              // In fact if we initialized it when auto tune is not enabled, then we would enforce needless checks
              // on related parameters that would impose more hassle on the user to configure
              this.autoTuner = null;
          }
  
          this.randomizeKeys = randomizeKeys;
          logger.info("ES_REST plugin initialized: " + CONNECTION_INFO);
      }