import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
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;
}
 

import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
@Override
public void onResponse(PutIndexTemplateResponse putIndexTemplateResponse) {
    if (putIndexTemplateResponse.isAcknowledged()) {
        future.set(TaskResult.ONE_ROW);
    } else {
        future.set(TaskResult.ZERO);
    }
}
 

import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
@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));
}
 

import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
public AsyncFuture<Void> configure() {
    final IndicesAdminClient indices = client.getClient().admin().indices();

    log.info("[{}] updating template for {}", templateName, index.template());

    final PutIndexTemplateRequestBuilder put = indices.preparePutTemplate(templateName);

    put.setSettings(type.getSettings());
    put.setTemplate(index.template());

    for (final Map.Entry<String, Map<String, Object>> mapping : type.getMappings().entrySet()) {
        put.addMapping(mapping.getKey(), mapping.getValue());
    }

    final ResolvableFuture<Void> future = async.future();

    final ListenableActionFuture<PutIndexTemplateResponse> target = put.execute();

    target.addListener(new ActionListener<PutIndexTemplateResponse>() {
        @Override
        public void onResponse(final PutIndexTemplateResponse response) {
            if (!response.isAcknowledged()) {
                future.fail(new Exception("request not acknowledged"));
                return;
            }

            future.resolve(null);
        }

        @Override
        public void onFailure(Exception e) {
            future.fail(e);
        }
    });

    future.onCancelled(() -> target.cancel(false));
    return future;
}
 

import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
@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 = new ClusterHealthRequestBuilder(connection.getClient().admin().cluster())
            .execute()
            .get();

    int numDataNodes = clusterHealth.getNumberOfDataNodes();
    int numReplicas = (numDataNodes < 2) ? 0 : 1;

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

    createMetaIndex(connection, "consoles", numDataNodes - 1);
    createMetaIndex(connection, "table-meta", numDataNodes - 1);

    PutIndexTemplateResponse response = new PutIndexTemplateRequestBuilder(connection.getClient().admin().indices(), "template_foxtrot_mappings")
            .setTemplate(String.format("%s-*",configuration.getElasticsearch().getTableNamePrefix()))
            .setSettings(
                    ImmutableSettings.builder()
                            .put("number_of_shards", 10)
                            .put("number_of_replicas", numReplicas)
            )
            .addMapping("document", ElasticsearchUtils.getDocumentMapping())
            .execute()
            .get();
    logger.info("Create mapping: {}", response.isAcknowledged());

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

}
 

import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
@Override
public void open(VDSConfiguration ctx) throws Exception {
    _logger.info("Initializing ElasticSearch target");
    _indexName = ctx.optString("index-name", "surf");
    _typeName = ctx.optString("type-name", "http");
    boolean useTemplate = ctx.optBoolean("use-template", true);
    if(!_indexName.startsWith("surf") && useTemplate){
        _logger.error("The index name *must* start with 'surf' if the default Surf template is used");
        _logger.error("Either use an index-name starting with 'surf' or set use-template to false and provide your own template");
        throw new IllegalArgumentException("Invalid index name for default template");
    }
    String cluster = ctx.optString("cluster-name", "elasticsearch");
    _node = nodeBuilder()
              .client(true)
              .clusterName(cluster)
              .node();
    _client = _node.client();
    _logger.info("ElasticSearch target initialized");
    if(useTemplate){
        InputStream stream = getClass().getResourceAsStream("/surf-template.json");
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        byte buf[] = new byte[1024];
        while(stream.read(buf) > 0){
            bos.write(buf);
        }
        stream.close();
        bos.close();
        byte [] template = bos.toByteArray();
        PutIndexTemplateRequestBuilder req = _client.admin().indices().preparePutTemplate("surf-template")
                .setSource(template);
        PutIndexTemplateResponse resp = req.execute().actionGet();
        if(resp.isAcknowledged()){
            _logger.info("Template loaded");
        }
    }
    else{
        _logger.info("Default Surf template not loaded - make sure you have configured ElasticSearch correctly");
    }
}
 

import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
@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;
}
 

import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
@Override
public ActionFuture<PutIndexTemplateResponse> putTemplate(final PutIndexTemplateRequest request) {
    return execute(PutIndexTemplateAction.INSTANCE, request);
}
 

import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
/**
 * Create a template if needed
 * 
 * @param template template name
 * @param force    force recreate template
 * @throws Exception
 */
private void createTemplate(String template, boolean force)
		throws Exception {
	if (logger.isTraceEnabled()){
		logger.trace("createTemplate(" + template + ")");
	}
	
	checkClient();

	// If template already exists and if we are in force mode, we delete the template
	if (force && isTemplateExist(template)) {
		if (logger.isDebugEnabled()){
			logger.debug("Force remove template [" + template + "]");
		}
		// Remove template in ElasticSearch !
		client.admin()
				.indices().prepareDeleteTemplate(template).execute()
				.actionGet();
	}

	// Read the template json file if exists and use it
	String source = readTemplate(template);
	if (source != null) {
		if (logger.isTraceEnabled()){
			logger.trace("Template [" + template + "]=" + source);
		}
		// Create template
		final PutIndexTemplateResponse response = client.admin().indices()
				.preparePutTemplate(template).setSource(source).execute()
				.actionGet();
		if (!response.isAcknowledged()) {
			throw new Exception("Could not define template [" + template
					+ "].");
		} else {
			if (logger.isDebugEnabled()) {
				logger.debug("Template [" + template
						+ "] successfully created.");
			}
		}
	} else {
		if (logger.isWarnEnabled()) {
			logger.warn("No template definition for [" + template
					+ "]. Ignoring.");
		}
	}

	if (logger.isTraceEnabled()){
		logger.trace("/createTemplate(" + template + ")");
	}
}
 

import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
@Override
protected ActionFuture<PutIndexTemplateResponse> doExecute(PutIndexTemplateRequest request) {
    return client.admin().indices().putTemplate(request);
}
 

import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse; //导入依赖的package包/类
/**
 * Puts an index template.
 */
ActionFuture<PutIndexTemplateResponse> putTemplate(PutIndexTemplateRequest request);