private boolean createTemplate(CollectionMetadata collection) throws Exception {
    String searchAlias = collection.getId() + "-search";
    String indexTemplateName = collection.getId() + "-template";
    String indexTemplatePattern = collection.getId() + "-*";

    log.debug("Attempting to initialize template for collection '" + collection.getId() + "'");

    PutIndexTemplateRequest request = new PutIndexTemplateRequest(indexTemplateName)
            .patterns(Arrays.asList(indexTemplatePattern))
            .alias(new Alias(searchAlias))
            .mapping("_doc", buildMappings(collection));

    client.indices().putTemplate(request, RequestOptions.DEFAULT);

    log.info("Template '" + indexTemplateName + "' for collection '" + collection.getId() + "' initialized");
    return true;
}
 

private static void putGarmadonTemplate(RestHighLevelClient esClient, ElasticsearchConfiguration elasticsearch)
    throws IOException, GarmadonEsException {
    PutIndexTemplateRequest indexRequest = new PutIndexTemplateRequest("garmadon");
    indexRequest.patterns(Collections.singletonList(elasticsearch.getIndexPrefix() + "*"));

    // Create template settings with mandatory one
    Settings.Builder templateSettings = Settings.builder()
        .put("sort.field", "timestamp")
        .put("sort.order", "desc")
        .put("analysis.analyzer.path_analyzer.tokenizer", "path_tokenizer")
        .put("analysis.tokenizer.path_tokenizer.type", "path_hierarchy")
        .put("analysis.tokenizer.path_tokenizer.delimiter", "/")
        .put("index.lifecycle.name", "garmadon")
        .put("index.lifecycle.rollover_alias", "garmadon");

    // Add settings from config
    elasticsearch.getSettings().forEach(templateSettings::put);

    indexRequest.settings(templateSettings);

    // Add garmadon alias
    Alias alias = new Alias("garmadon");
    indexRequest.alias(alias);

    String template = IOUtils.toString(Objects.requireNonNull(ElasticSearchReader.class.getClassLoader()
        .getResourceAsStream("template.json")), "UTF-8");

    indexRequest.mapping(ES_TYPE, template, XContentType.JSON);

    if (!esClient.indices().putTemplate(indexRequest, RequestOptions.DEFAULT).isAcknowledged()) {
        throw new GarmadonEsException("Failed to insert garmadon template");
    }
}
 

private void updateTemplate(DocIndexMetaData md,
                            TransportPutIndexTemplateAction transportPutIndexTemplateAction,
                            Settings updateSettings) {
    String templateName = PartitionName.templateName(ident.schema(), ident.name());
    PutIndexTemplateRequest request = new PutIndexTemplateRequest(templateName)
            .mapping(Constants.DEFAULT_MAPPING_TYPE, md.defaultMappingMap)
            .create(false)
            .settings(updateSettings)
            .template(templateName + "*");
    for (String alias : md.aliases()) {
        request = request.alias(new Alias(alias));
    }
    transportPutIndexTemplateAction.execute(request);
}
 

private ListenableFuture<Long> updateTemplate(Map<String, Object> newMappings,
                                              Settings newSettings,
                                              TableIdent tableIdent,
                                              AbstractDDLAnalyzedStatement statement) {
    String templateName = PartitionName.templateName(tableIdent.schema(), tableIdent.name());
    IndexTemplateMetaData indexTemplateMetaData =
            clusterService.state().metaData().templates().get(templateName);
    if (indexTemplateMetaData == null) {
        return Futures.immediateFailedFuture(new RuntimeException("Template for partitioned table is missing"));
    }

    // merge mappings
    Map<String, Object> mapping = mergeTemplateMapping(indexTemplateMetaData, newMappings);

    // merge settings
    Settings.Builder settingsBuilder = Settings.builder();
    settingsBuilder.put(indexTemplateMetaData.settings());
    settingsBuilder.put(newSettings);

    PutIndexTemplateRequest request = new PutIndexTemplateRequest(templateName)
            .create(false)
            .mapping(Constants.DEFAULT_MAPPING_TYPE, mapping)
            .order(indexTemplateMetaData.order())
            .settings(settingsBuilder.build())
            .template(indexTemplateMetaData.template());

    request.putHeader(LoginUserContext.USER_INFO_KEY, statement.getParameterContext().getLoginUserContext());
    for (ObjectObjectCursor<String, AliasMetaData> container : indexTemplateMetaData.aliases()) {
        Alias alias = new Alias(container.key);
        request.alias(alias);
    }

    SettableFuture<Long> result = SettableFuture.create();
    transportActionProvider.transportPutIndexTemplateAction().execute(request,
            new SettableFutureToNullActionListener<PutIndexTemplateResponse>(result));

    return result;
}
 

private PutIndexTemplateRequest createTemplateRequest(CreateTableAnalyzedStatement statement) {
    PutIndexTemplateRequest putIndexTemplateRequest = new PutIndexTemplateRequest(statement.templateName())
            .mapping(Constants.DEFAULT_MAPPING_TYPE, statement.mapping())
            .create(true)
            .settings(settings(statement))
            .template(statement.templatePrefix())
            .order(100)
            .alias(new Alias(statement.tableIdent().indexName()));
    putIndexTemplateRequest.putHeader(LoginUserContext.USER_INFO_KEY, statement.getParameterContext().getLoginUserContext());
    return putIndexTemplateRequest;
}
 

private PutIndexTemplateRequest buildRequest(ESCreateTemplateNode node) {
    PutIndexTemplateRequest templateRequest = new PutIndexTemplateRequest(node.templateName())
            .mapping(Constants.DEFAULT_MAPPING_TYPE, node.mapping())
            .create(true)
            .settings(node.indexSettings())
            .template(node.indexMatch());
    if (node.alias() != null) {
        templateRequest.alias(new Alias(node.alias()));
    }

    templateRequest.putHeader(LoginUserContext.USER_INFO_KEY, getParamContext().getLoginUserContext());
    return templateRequest;
}
 

@SuppressWarnings({"unchecked"})
@Override
public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) {
    PutIndexTemplateRequest putRequest = new PutIndexTemplateRequest(request.param("name"));
    putRequest.template(request.param("template", putRequest.template()));
    putRequest.order(request.paramAsInt("order", putRequest.order()));
    putRequest.masterNodeTimeout(request.paramAsTime("master_timeout", putRequest.masterNodeTimeout()));
    putRequest.create(request.paramAsBoolean("create", false));
    putRequest.cause(request.param("cause", ""));
    putRequest.source(request.content());
    client.admin().indices().putTemplate(putRequest, new AcknowledgedRestListener<PutIndexTemplateResponse>(channel));
}
 

@Override
public void execute(IndexTemplate indexTemplate) {
    try {
        createClient().admin().indices().putTemplate(
                new PutIndexTemplateRequest()
                        .name(indexTemplate.getName())
                        .source(indexTemplate.getSource(), XContentType.JSON)
        );
    } catch (Exception e) {
        throw new ConfigurationException(e.getMessage(), e);
    }
}
 

@Override
public void execute(IndexTemplate indexTemplate) {
    try {
        createClient().admin().indices().putTemplate(
                new PutIndexTemplateRequest()
                        .name(indexTemplate.getName())
                        .source(indexTemplate.getSource())
        );
    } catch (Exception e) {
        throw new ConfigurationException(e);
    }
}
 

@Override
public void execute(IndexTemplate indexTemplate) {
    try {
        createClient().admin().indices().putTemplate(
                new PutIndexTemplateRequest()
                        .name(indexTemplate.getName())
                        .source(indexTemplate.getSource(), XContentType.JSON)
        );
    } catch (Exception e) {
        throw new ConfigurationException(e.getMessage(), e);
    }
}
 

@Before
public void before() throws IOException {
  repository = factory.create(Person.class);
  repository.refreshPolicy(IMMEDIATE);
  final IndicesAdminClient indices = client.admin().indices();

  final PutIndexTemplateRequest datas = indices.preparePutTemplate("datas")
    .setSource(toByteArray(getClass().getResourceAsStream("/datas.json")), JSON)
    .request();
  checkState(indices.putTemplate(datas).actionGet().isAcknowledged());
}
 

public static PutIndexTemplateRequest getClusterTemplateMapping() {
    try {
        return new PutIndexTemplateRequest().name("template_foxtrot_mappings")
                .patterns(Lists.newArrayList(String.format("%s-*", getTableNamePrefix())))
                .mapping(DOCUMENT_TYPE_NAME, getDocumentMapping());
    }
    catch (IOException ex) {
        logger.error("TEMPLATE_CREATION_FAILED", ex);
        return null;
    }
}
 

public static void initializeMappings(Client client) {
    PutIndexTemplateRequest templateRequest = getClusterTemplateMapping();
    client.admin()
            .indices()
            .putTemplate(templateRequest)
            .actionGet();
}
 

@Override
protected void run(Bootstrap<FoxtrotServerConfiguration> bootstrap, Namespace namespace, FoxtrotServerConfiguration configuration)
        throws Exception {
    ElasticsearchConfig esConfig = configuration.getElasticsearch();
    ElasticsearchConnection connection = new ElasticsearchConnection(esConfig);
    connection.start();

    ClusterHealthResponse clusterHealth = connection.getClient()
            .admin()
            .cluster()
            .health(new ClusterHealthRequest())
            .actionGet();
    int numDataNodes = clusterHealth.getNumberOfDataNodes();
    int numReplicas = (numDataNodes < 2) ? 0 : 1;

    logger.info("# data nodes: {}, Setting replica count to: {}", numDataNodes, numReplicas);

    createMetaIndex(connection, ElasticsearchConsolePersistence.INDEX, numReplicas);
    createMetaIndex(connection, ElasticsearchConsolePersistence.INDEX_V2, numReplicas);
    createMetaIndex(connection, TableMapStore.TABLE_META_INDEX, numReplicas);
    createMetaIndex(connection, ElasticsearchConsolePersistence.INDEX_HISTORY, numReplicas);
    createMetaIndex(connection, FqlStoreServiceImpl.FQL_STORE_INDEX, numReplicas);

    logger.info("Creating mapping");
    PutIndexTemplateRequest putIndexTemplateRequest = ElasticsearchUtils.getClusterTemplateMapping();
    PutIndexTemplateResponse response = connection.getClient()
            .admin()
            .indices()
            .putTemplate(putIndexTemplateRequest)
            .actionGet();
    logger.info("Created mapping: {}", response.isAcknowledged());

    logger.info("Creating hbase table");
    HBaseUtil.createTable(configuration.getHbase(), configuration.getHbase()
            .getTableName());
}
 

@Override
protected XContentBuilder toXContent(PutIndexTemplateRequest request, PutIndexTemplateResponse response, XContentBuilder builder) throws IOException {
    builder.startObject()
            .field(Fields.OK, true)
            .field(Fields.ACKNOWLEDGED, response.isAcknowledged())
            .endObject();
    return builder;
}
 

@Override
protected ClusterBlockException checkBlock(CreateTableRequest request, ClusterState state) {
    if (request.getCreateIndexRequest() != null) {
        CreateIndexRequest createIndexRequest = request.getCreateIndexRequest();
        return transportCreateIndexAction.checkBlock(createIndexRequest, state);
    } else if (request.getPutIndexTemplateRequest() != null) {
        PutIndexTemplateRequest putIndexTemplateRequest = request.getPutIndexTemplateRequest();
        return transportPutIndexTemplateAction.checkBlock(putIndexTemplateRequest, state);
    } else {
        throw new IllegalStateException("Unknown table request");
    }
}
 

public CompletableFuture<Long> create(BoundCreateTable createTable) {
    var templateName = createTable.templateName();
    var relationName = createTable.tableIdent();
    var createTableRequest = templateName == null
        ? new CreateTableRequest(
            new CreateIndexRequest(
                relationName.indexNameOrAlias(),
                createTable.tableParameter().settings()
            ).mapping(Constants.DEFAULT_MAPPING_TYPE, createTable.mapping())
        )
        : new CreateTableRequest(
            new PutIndexTemplateRequest(templateName)
                .mapping(Constants.DEFAULT_MAPPING_TYPE, createTable.mapping())
                .create(true)
                .settings(createTable.tableParameter().settings())
                .patterns(Collections.singletonList(createTable.templatePrefix()))
                .order(100)
                .alias(new Alias(relationName.indexNameOrAlias()))
    );
    return Transports.execute(transportCreateTableAction, createTableRequest, resp -> {
        if (!resp.isAllShardsAcked() && LOGGER.isWarnEnabled()) {
            LOGGER.warn("CREATE TABLE `{}` was not acknowledged. This could lead to inconsistent state.", relationName.fqn());
        }
        return 1L;
    }).exceptionally(error -> {
        Throwable t = SQLExceptions.unwrap(error);
        String message = t.getMessage();
        Throwable cause = t.getCause();
        if ("mapping [default]".equals(message) && cause != null) {
            // this is a generic mapping parse exception,
            // the cause has usually a better more detailed error message
            return Exceptions.rethrowRuntimeException(cause);
        } else if (createTable.ifNotExists() && isTableExistsError(t, templateName)) {
            return 0L;
        } else {
            return Exceptions.rethrowRuntimeException(t);
        }
    });
}
 

@Override
public ActionFuture<PutIndexTemplateResponse> putTemplate(final PutIndexTemplateRequest request) {
    return execute(PutIndexTemplateAction.INSTANCE, request);
}
 

@Override
public void putTemplate(final PutIndexTemplateRequest request, final ActionListener<PutIndexTemplateResponse> listener) {
    execute(PutIndexTemplateAction.INSTANCE, request, listener);
}
 

private String extractPayload(PutIndexTemplateRequest putIndexTemplateRequest) throws IOException {
    BytesStreamOutput out = new BytesStreamOutput();
    putIndexTemplateRequest.writeTo(out);
    return new String(out.bytes().toBytesRef().bytes);
}
 

private String extractPayload(PutIndexTemplateRequest putIndexTemplateRequest) throws IOException {
    BytesStreamOutput out = new BytesStreamOutput();
    putIndexTemplateRequest.writeTo(out);
    return new String(out.bytes().toBytesRef().bytes);
}
 

private String extractPayload(PutIndexTemplateRequest putIndexTemplateRequest) throws IOException {
    BytesStreamOutput out = new BytesStreamOutput();
    putIndexTemplateRequest.writeTo(out);
    return new String(out.bytes().toBytesRef().bytes);
}
 

public PutIndexTemplateRequestBuilder(Client client, JsonToString<JsonInput> jsonToString, StringToJson<JsonOutput> stringToJson) {
    super(client, new PutIndexTemplateRequest(null), jsonToString, stringToJson);
}
 

@Override
protected ActionFuture<PutIndexTemplateResponse> doExecute(PutIndexTemplateRequest request) {
    return client.admin().indices().putTemplate(request);
}
 

@Override
public ActionFuture<AcknowledgedResponse> putTemplate(final PutIndexTemplateRequest request) {
    return execute(PutIndexTemplateAction.INSTANCE, request);
}
 

@Override
public void putTemplate(final PutIndexTemplateRequest request, final ActionListener<AcknowledgedResponse> listener) {
    execute(PutIndexTemplateAction.INSTANCE, request, listener);
}
 

public CreateTableRequest(PutIndexTemplateRequest putIndexTemplateRequest) {
    this.createIndexRequest = null;
    this.putIndexTemplateRequest = putIndexTemplateRequest;
}
 

@Nullable
public PutIndexTemplateRequest getPutIndexTemplateRequest() {
    return putIndexTemplateRequest;
}
 

@Test
public void passesIndexTemplateToClient() throws IOException {

    //given
    BulkProcessorObjectFactory factory = spy(createTestObjectFactoryBuilder().build());

    IndicesAdminClient indicesAdminClient = mockedIndicesAdminClient(factory);

    IndexTemplate indexTemplate = spy(IndexTemplate.newBuilder()
            .withPath("classpath:indexTemplate.json")
            .withName("testName")
            .build());

    String expectedPayload = indexTemplate.getSource().replaceAll("\\s+","");

    // when
    factory.execute(indexTemplate);

    // then
    ArgumentCaptor<PutIndexTemplateRequest> requestArgumentCaptor = ArgumentCaptor.forClass(PutIndexTemplateRequest.class);
    verify(indicesAdminClient).putTemplate(requestArgumentCaptor.capture());

    String actualPayload = extractPayload(requestArgumentCaptor.getValue());

    Assert.assertTrue(actualPayload.contains(new ObjectMapper().readTree(expectedPayload).get("mappings").toString()));

}
 

@Test
public void passesIndexTemplateToClient() throws IOException {

    //given
    BulkProcessorObjectFactory factory = spy(createTestObjectFactoryBuilder().build());

    IndicesAdminClient indicesAdminClient = mockedIndicesAdminClient(factory);

    IndexTemplate indexTemplate = spy(IndexTemplate.newBuilder()
            .withPath("classpath:indexTemplate.json")
            .withName("testName")
            .build());

    String expectedPayload = indexTemplate.getSource().replaceAll("\\s+","");

    // when
    factory.execute(indexTemplate);

    // then
    ArgumentCaptor<PutIndexTemplateRequest> requestArgumentCaptor = ArgumentCaptor.forClass(PutIndexTemplateRequest.class);
    verify(indicesAdminClient).putTemplate(requestArgumentCaptor.capture());

    String actualPayload = extractPayload(requestArgumentCaptor.getValue());

    Assert.assertTrue(actualPayload.contains(new ObjectMapper().readTree(expectedPayload).get("mappings").toString()));

}