Java 类org.elasticsearch.client.IndicesAdminClient 实例源码

@Test
public void testCreateIndexWithoutSettings() {
    // Test data
    final String indexName = "Index_Name";
    final AdminClient adminClient = createMock(AdminClient.class);
    final IndicesAdminClient indicesAdminClient = createMock(IndicesAdminClient.class);
    final CreateIndexRequestBuilder createIndexRequestBuilder = createMock(CreateIndexRequestBuilder.class);
    final CreateIndexResponse createIndexResponse = createMock(CreateIndexResponse.class);
    // Reset
    resetAll();
    // Expectations
    expect(esClient.admin()).andReturn(adminClient);
    expect(adminClient.indices()).andReturn(indicesAdminClient);
    expect(indicesAdminClient.prepareCreate(indexName)).andReturn(createIndexRequestBuilder);
    expect(createIndexRequestBuilder.get()).andReturn(createIndexResponse);
    // Replay
    replayAll();
    // Run test scenario
    final boolean result = elasticsearchClientWrapper.createIndex(indexName);
    // Verify
    verifyAll();
    assertEquals(createIndexResponse.isAcknowledged(), result);
}

@Test
public void testCreateIndex() {
    // Test data
    final String indexName = "Index_Name";
    final Map<String, Object> settings = Collections.singletonMap("settings_key", "settings_value");
    final AdminClient adminClient = createMock(AdminClient.class);
    final IndicesAdminClient indicesAdminClient = createMock(IndicesAdminClient.class);
    final CreateIndexRequestBuilder createIndexRequestBuilder = createMock(CreateIndexRequestBuilder.class);
    final CreateIndexResponse createIndexResponse = createMock(CreateIndexResponse.class);
    // Reset
    resetAll();
    // Expectations
    expect(esClient.admin()).andReturn(adminClient);
    expect(adminClient.indices()).andReturn(indicesAdminClient);
    expect(indicesAdminClient.prepareCreate(indexName)).andReturn(createIndexRequestBuilder);
    expect(createIndexRequestBuilder.setSettings(settings)).andReturn(createIndexRequestBuilder);
    expect(createIndexRequestBuilder.get()).andReturn(createIndexResponse);
    // Replay
    replayAll();
    // Run test scenario
    final boolean result = elasticsearchClientWrapper.createIndex(indexName, settings);
    // Verify
    verifyAll();
    assertEquals(createIndexResponse.isAcknowledged(), result);
}

@Test
public void testRefreshIndex() {
    // Test data
    final String indexName = "Index_Name";
    final AdminClient adminClient = createMock(AdminClient.class);
    final IndicesAdminClient indicesAdminClient = createMock(IndicesAdminClient.class);
    final RefreshRequestBuilder refreshRequestBuilder = createMock(RefreshRequestBuilder.class);
    final PutMappingResponse putMappingResponse = createMock(PutMappingResponse.class);
    // Reset
    resetAll();
    // Expectations
    expect(esClient.admin()).andReturn(adminClient);
    expect(adminClient.indices()).andReturn(indicesAdminClient);
    expect(indicesAdminClient.prepareRefresh(indexName)).andReturn(refreshRequestBuilder);
    expect(refreshRequestBuilder.get()).andReturn(createMock(RefreshResponse.class));
    // Replay
    replayAll();
    // Run test scenario
    elasticsearchClientWrapper.refreshIndex(indexName);
    // Verify
    verifyAll();
}

@Test
public void testAddAlias() {
    //Test data
    final String indexName = "index";
    final String aliasName = "alias";
    final AdminClient adminClient = createMock(AdminClient.class);
    final IndicesAdminClient indicesAdminClient = createMock(IndicesAdminClient.class);
    final IndicesAliasesRequestBuilder indicesAliasesRequestBuilder = createMock(IndicesAliasesRequestBuilder.class);
    final IndicesAliasesResponse indicesAliasesResponse = createMock(IndicesAliasesResponse.class);
    //Reset
    resetAll();
    //Expectations
    expect(esClient.admin()).andReturn(adminClient);
    expect(adminClient.indices()).andReturn(indicesAdminClient);
    expect(indicesAdminClient.prepareAliases()).andReturn(indicesAliasesRequestBuilder);
    expect(indicesAliasesRequestBuilder.addAlias(indexName, aliasName)).andReturn(indicesAliasesRequestBuilder);
    expect(indicesAliasesRequestBuilder.get()).andReturn(indicesAliasesResponse);
    //Replay
    replayAll();
    //Run test scenario
    final boolean result = elasticsearchClientWrapper.addAlias(indexName, aliasName);
    //Verify
    verifyAll();
    assertEquals(indicesAliasesResponse.isAcknowledged(), result);
}

@Test
public void testDeleteIndex() {
    //Test data
    final String indexName = "index";
    final AdminClient adminClient = createMock(AdminClient.class);
    final IndicesAdminClient indicesAdminClient = createMock(IndicesAdminClient.class);
    final DeleteIndexRequestBuilder deleteIndexRequestBuilder = createMock(DeleteIndexRequestBuilder.class);
    final DeleteIndexResponse deleteIndexResponse = createMock(DeleteIndexResponse.class);
    //Reset
    resetAll();
    //Expectations
    expect(esClient.admin()).andReturn(adminClient);
    expect(adminClient.indices()).andReturn(indicesAdminClient);
    expect(indicesAdminClient.prepareDelete(indexName)).andReturn(deleteIndexRequestBuilder);
    expect(deleteIndexRequestBuilder.get()).andReturn(deleteIndexResponse);
    //Replay
    replayAll();
    //Run test scenario
    final boolean result = elasticsearchClientWrapper.deleteIndex(indexName);
    //Verify
    verifyAll();
    assertEquals(deleteIndexResponse.isAcknowledged(), result);
}

@Test
public void testIndexExists() {
    //Test data
    final String indexName = "index";
    final AdminClient adminClient = createMock(AdminClient.class);
    final IndicesAdminClient indicesAdminClient = createMock(IndicesAdminClient.class);
    final IndicesExistsRequestBuilder indicesExistsRequestBuilder = createMock(IndicesExistsRequestBuilder.class);
    final IndicesExistsResponse indicesExistsResponse = createMock(IndicesExistsResponse.class);
    final boolean exists = new Random().nextBoolean();
    //Reset
    resetAll();
    //Expectations
    expect(esClient.admin()).andReturn(adminClient);
    expect(adminClient.indices()).andReturn(indicesAdminClient);
    expect(indicesAdminClient.prepareExists(indexName)).andReturn(indicesExistsRequestBuilder);
    expect(indicesExistsRequestBuilder.get()).andReturn(indicesExistsResponse);
    expect(indicesExistsResponse.isExists()).andReturn(exists);
    //Replay
    replayAll();
    //Run test scenario
    final boolean result = elasticsearchClientWrapper.indexExists(indexName);
    //Verify
    verifyAll();
    assertEquals(exists, result);
}

@Before
public void before() throws IOException {
  final IndicesAdminClient indices = client.admin().indices();
  if(!indices.prepareExists(INDEX).execute().actionGet().isExists()) {
    indices.prepareCreate(INDEX).execute().actionGet();
  }
  final JsonSerializer<Person> serializer = mapper.serializer(Person.class);
  final BulkRequestBuilder bulk = client.prepareBulk();
  for (int i = 0; i < SIZE; i++) {

    final String name = UUID.randomUUID().toString();
    final IndexRequest request = new IndexRequest(INDEX, TYPE);
    request.source(serializer.apply(Person.builder().id("").firstname(name).lastname(name).build()), JSON);
    bulk.add(request);
  }

  client.bulk(bulk.request()).actionGet();
  flush(INDEX);
}

private static void update() {
    try {
        IndicesAdminClient indicesAdminClient = client.admin().indices();
        if (indicesAdminClient.prepareExists(INDEX_NAME_v2).execute().actionGet().isExists()) {
            indicesAdminClient.delete(new DeleteIndexRequest(INDEX_NAME_v2)).actionGet();
        }
        indicesAdminClient.prepareCreate(INDEX_NAME_v2).addMapping(INDEX_TYPE,getItemInfoMapping()).execute().actionGet();
        //等待集群shard,防止No shard available for 异常
        ClusterAdminClient clusterAdminClient = client.admin().cluster();
        clusterAdminClient.prepareHealth().setWaitForYellowStatus().execute().actionGet(5000);
        //0、更新mapping
        updateMapping();
        //1、更新数据
        reindexData(indicesAdminClient);

        //2、realias 重新建立连接
       indicesAdminClient.prepareAliases().removeAlias(INDEX_NAME_v1, ALIX_NAME).addAlias(INDEX_NAME_v2, ALIX_NAME).execute().actionGet();
    }catch (Exception e){
        log.error("beforeUpdate error:{}"+e.getLocalizedMessage());
    }

}

private static void prepareData(IndicesAdminClient indicesAdminClient) throws InterruptedException {
    BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
    //批量添加
    for (int i = 0; i < 1000; i++) {
        ItemInfo iteminfo=new ItemInfo(i,"商品"+i, new Random().nextFloat(),new Date());
        // 当opType是Index时,如果文档id存在,更新文档,否则创建文档 当opType是Create,如果文档id存在,抛出文档存在的错误 *
        IndexRequestBuilder indexRequestBuilder = client.prepareIndex(INDEX_NAME_v1, INDEX_TYPE).setId(i+"").setSource(JSONObject.toJSONString(iteminfo)).setOpType(IndexRequest.OpType.INDEX);
        bulkRequestBuilder.add(indexRequestBuilder);
        //数据日期不一样
    }
    BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
    if (bulkResponse.hasFailures()) {
     log.error("bulk index error:{}",bulkResponse.buildFailureMessage());
    } else {
        log.info("index docs : {}" , bulkResponse);
    }
}

private static void beforeUpdate()  {
    try {
        IndicesAdminClient indicesAdminClient = client.admin().indices();
        indicesAdminClient.delete(new DeleteIndexRequest(INDEX_NAME_v1)).actionGet();
        if (!indicesAdminClient.prepareExists(INDEX_NAME_v1).execute().actionGet().isExists()) {
            indicesAdminClient.prepareCreate(INDEX_NAME_v1).addMapping(INDEX_TYPE,getItemInfoMapping()).execute().actionGet();
        }
        //等待集群shard,防止No shard available for 异常
        ClusterAdminClient clusterAdminClient = client.admin().cluster();
        clusterAdminClient.prepareHealth().setWaitForYellowStatus().execute().actionGet(5000);
        //创建别名alias
        indicesAdminClient.prepareAliases().addAlias(INDEX_NAME_v1, ALIX_NAME).execute().actionGet();
        prepareData(indicesAdminClient);
    }catch (Exception e){
        log.error("beforeUpdate error:{}"+e.getLocalizedMessage());
    }
}

/**
 * ttp://localhost:9200/index/_analyze?text=做人如果没有梦想那和咸鱼有什么区别勒&analyzer=pinyin_analyzer
 * ttp://localhost:9200/index/_analyze?text=今天是个好天气啊&analyzer=ik_max_word
 */
private static void analyze() {
    IndicesAdminClient indicesAdminClient = client.admin().indices();
    String analyzerName="pinyin";
    String text="默认的拼音例子";
    showAnaylzerText(indicesAdminClient,analyzerName,text);
    //自定义分析器必须要指定indexName,插件的就不用了
     analyzerName="pinyin_analyzer";
    text="做人如果没有梦想那和咸鱼有什么区别勒";
    showAnaylzerText(indicesAdminClient,analyzerName,text);
    //官网的demo例子
    analyzerName="my_analyzer";
    text="The quick & brown fox is a dog";
    showAnaylzerText(indicesAdminClient,analyzerName,text);
    //验证ik分词
    analyzerName="ik_max_word";
    text="好好学习天天向上！";
    showAnaylzerText(indicesAdminClient,analyzerName,text);
}

/**
     * http://es.xiaoleilu.com/010_Intro/25_Tutorial_Indexing.html
     * http://es.xiaoleilu.com/070_Index_Mgmt/05_Create_Delete.html
     * 索引相关的
     */
    private static void index() throws IOException, InterruptedException {
        // cluster()，产生一个允许从集群中执行action或操作的client；
        IndicesAdminClient indicesAdminClient = client.admin().indices();
        //创建索引
        if (checkExistsIndex(indicesAdminClient, INDEX_NAME)) {
            deleteIndex(indicesAdminClient, INDEX_NAME);
        }
//        String settings = getIndexSetting();
//        CreateIndexResponse createIndexResponse = indicesAdminClient.prepareCreate(INDEX_NAME).setSettings(settings).execute().actionGet();

        CreateIndexResponse createIndexResponse = indicesAdminClient.prepareCreate(INDEX_NAME).setSettings().execute().actionGet();
//        log.info("创建索引{}：{}", INDEX_NAME, createIndexResponse.getContext());

        //索引的相关配置操作
        indexConfig(indicesAdminClient, INDEX_NAME);

//        indexMapping(indicesAdminClient, INDEX_NAME, TYPE_NAME);
    }

/**
 * 索引的相关操作
 *
 * @param indicesAdminClient
 * @param indexName
 * @throws IOException
 */
private static void indexConfig(IndicesAdminClient indicesAdminClient, String indexName) throws IOException {
    //settings 设置
    String settings = getIndexSetting();
    // PUT /my_temp_index/_settings updatesettings
    showIndexSettings(indicesAdminClient,indexName);
    UpdateSettingsResponse updateSettingsResponse = indicesAdminClient.prepareUpdateSettings(indexName).setSettings(settings).execute().actionGet();
    log.info("更新 index setting:{}", updateSettingsResponse);


    //更新索引settings之前要关闭索引
    indicesAdminClient.close(new CloseIndexRequest().indices(indexName)).actionGet();
    //配置拼音自定义分析器
    indicesAdminClient.prepareUpdateSettings(indexName).setSettings(getIndexPinYinSetting()).execute().actionGet();
    //自定义分析器
    indicesAdminClient.prepareUpdateSettings(indexName).setSettings(getIndexDemoSetting()).execute().actionGet();
    //打开索引
    indicesAdminClient.open(new OpenIndexRequest().indices(indexName)).actionGet();

    //索引别名映射
    createAliasIndex(indicesAdminClient);

    showIndexSettings(indicesAdminClient,indexName);
}

/**
     * 索引 别名 就像一个快捷方式或软连接，可以指向一个或多个索引，也可以给任何需要索引名的 API 使用。别名带给我们极大的灵活性，
     * 比如 我们线上使用一个索引名为index_a的结构，里面有一个类型现在不满足要求，现在需要修改，因为elasticsearch不支持直接修改类型，所以我们必须要重新建立一个新的索引
     * 然后将久索引的数据拷贝过去。 但是如果让新的索引起作用，我们需要要修改引用代码，因为索引名称更换了，但是如果我们一开始创建索引的时候就给索引增加一个别名
     * 使用的时候都是使用index_alis 无论软连接的指向是什么索引，对外暴露的始终都是index_alis
     * @param indicesAdminClient
     */
    private static void createAliasIndex(IndicesAdminClient indicesAdminClient) {
//        1 创建索引 elsdb_alis_v1。
//        2 将别名 elsdb_alis 指向 elsdb_alis_v1
//        3 然后，我们决定修改索引elsdb_alis_v1中一个字段的映射。当然我们不能修改现存的映射，索引我们需要重新索引数据。首先，我们创建有新的映射的索引 elsdb_alis_v2。
//        4 然后我们从将数据从 elsdb_alis_v1 迁移到 elsdb_alis_v2，下面的过程在【重新索引】中描述过了。一旦我们认为数据已经被正确的索引了，我们就将别名指向新的索引。


        //创建索引
        if (checkExistsIndex(indicesAdminClient, INDEX_ALIAS_NAME_VERSION_ONE)) {
            deleteIndex(indicesAdminClient, INDEX_ALIAS_NAME_VERSION_ONE);
        }

        indicesAdminClient.prepareCreate(INDEX_ALIAS_NAME_VERSION_ONE).setSettings().execute().actionGet();
        //添加alias 所有别名
        indicesAdminClient.prepareAliases().addAlias(INDEX_ALIAS_NAME_VERSION_ONE,INDEX_ALIAS_NAME_ALIS).execute().actionGet();

        GetAliasesResponse getAliasesResponse = indicesAdminClient.getAliases(new GetAliasesRequest().indices(INDEX_ALIAS_NAME_ALIS)).actionGet();
        //log.info("getAliasesResponse index setting:{}", getAliasesResponse.getAliases());

        indicesAdminClient.prepareAliases().removeAlias(INDEX_ALIAS_NAME_VERSION_ONE,INDEX_ALIAS_NAME_ALIS).addAlias(INDEX_ALIAS_NAME_VERSION_TWO,INDEX_ALIAS_NAME_ALIS);
    }

public boolean createMapping(Class<?> docType) {
    Mapping mapping = getMappingFromClass(docType);
    IndicesAdminClient idc = client.admin().indices();
    GetMappingsResponse gmr = idc.getMappings(new GetMappingsRequest()).actionGet();
    ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> mappings = gmr.getMappings();
    if (mappings.containsKey(mapping.getType())) {
        log.info("Found mapping for class " + docType.getName() + ".");
        return false;
    }
    log.info("Mapping not found for class " + docType.getName() + ". Auto-create...");
    PutMappingResponse pmr = idc.preparePutMapping(index).setType(mapping.getType()).setSource(mapping.getSource())
            .get();
    if (!pmr.isAcknowledged()) {
        throw new RuntimeException("Failed to create mapping for class:" + docType.getName() + ".");
    }
    return true;
}

/**
 * Create index.
 * 
 * @return True if index create ok, false if index is already exist.
 */
public boolean createIndex() {
    // check index exist:
    IndicesAdminClient idc = client.admin().indices();
    IndicesExistsResponse ier = idc.exists(new IndicesExistsRequest(index)).actionGet();
    if (!ier.isExists()) {
        log.info("Index not found. Auto-create...");
        // create index:
        CreateIndexResponse cir = idc.create(new CreateIndexRequest(index)).actionGet();
        if (!cir.isAcknowledged()) {
            throw new RuntimeException("Failed to create index.");
        }
        return true;
    }
    return false;
}

public void createIndexIfNotExists(String index, String settingJson, Map<String, String> mappingJson) {
    String formattedIndex = formatIndex(index);
    IndicesAdminClient indicesClient = client.admin().indices();

    IndicesExistsResponse existsResponse = indicesClient.prepareExists(formattedIndex).get();
    if (existsResponse.isExists()) {
        return;
    }

    CreateIndexRequestBuilder builder = indicesClient.prepareCreate(formattedIndex)
            .setSettings(Settings.settingsBuilder().loadFromSource(settingJson));
    mappingJson.forEach((k, v) -> {
        builder.addMapping(k, v);
    });

    CreateIndexResponse indexResponse = builder.get();
    if (!indexResponse.isAcknowledged()) {
        throw new ElasticsearchException(String.format("index %s の作成に失敗しました", index));
    }
}

/**
 * 获取当前有效的索引名
 * 
 * @return
 */
public static String getCurrentValidIndex() {
    IndexType indexType = IndexType.RESOURCES;
    if (indexName != null) {
        if (reLoad) {
            synchronized (AnalyzeHelper.class) {
                if (reLoad) {
                    reLoad = false;
                    indexName = indexName.equals(indexType.index_type_1()) ? indexType
                            .index_type_2() : indexType.index_type_1();
                }
            }
        }
        return indexName;
    }
    IndicesAdminClient adminClient = ESClient.getClient().admin().indices();
    if (adminClient.prepareExists(indexType.index_type_1()).execute().actionGet().isExists()) {
        indexName = indexType.index_type_1();
    } else if (adminClient.prepareExists(indexType.index_type_2()).execute().actionGet()
            .isExists()) {
        indexName = indexType.index_type_2();
    }
    return indexName;
}

@Override
public boolean prepare(MappingConfiguration mapping) {
    if (!mapping.isIndexCreationEnabled()) {
        log.info("Index creation is disabled, stopping preparations!");
        return false;
    }

    try {
        IndicesAdminClient indices = getClient().admin().indices();
        String indexId = mapping.getIndex();
        if (indices.prepareExists(indexId).get().isExists()) {
            log.info("Index {} already exists, updating the mapping ...", indexId);
            return updateMapping(indexId, mapping);
        } else {
            log.info("Index {} does not exist, creating it ...", indexId);
            return createMapping(mapping, indexId);
        }
    } catch (RuntimeException e) {
        throw new SinkError(e, "Problem preparing sink: %s", e.getMessage());
    }
}

protected boolean createMapping(MappingConfiguration mapping, String indexId) {
    IndicesAdminClient indices = getClient().admin().indices();

    Map<String, Object> schema = schemaGenerator.generate(mapping);
    log.trace("Built schema creation request:\n{}", Arrays.toString(schema.entrySet().toArray()));

    // create metadata mapping and schema mapping
    CreateIndexRequestBuilder request = indices.prepareCreate(indexId)
            .addMapping(MetadataDataMapping.METADATA_TYPE_NAME, getMetadataSchema())
            .addMapping(mapping.getType(), schema);
    if (mapping.hasIndexCreationRequest()) {
        request.setSettings(mapping.getIndexCreationRequest());
    }

    CreateIndexResponse response = request.get();
    log.debug("Created indices: {}, acknowledged: {}", response, response.isAcknowledged());

    Map<String, Object> mdRecord = createMetadataRecord(mapping.getVersion(), mapping.getName());
    IndexResponse mdResponse = getClient().prepareIndex(indexId, MetadataDataMapping.METADATA_TYPE_NAME, MetadataDataMapping.METADATA_ROW_ID).setSource(mdRecord).get();
    log.debug("Saved mapping metadata '{}': {}", mdResponse.isCreated(), Arrays.toString(mdRecord.entrySet().toArray()));

    return (mdResponse.isCreated() && response.isAcknowledged());
}

/**
 * Removes all data associated with tenant.
 *
 * @param tenantId index
 */
public void clearTenant(String tenantId) {
    String index = getIndex(tenantId);

    synchronized (knownIndices) {
        IndicesAdminClient indices = client.admin().indices();

        boolean indexExists = indices.prepareExists(index)
                .execute()
                .actionGet()
                .isExists();

        if (indexExists) {
            indices.prepareDelete(index)
                    .execute()
                    .actionGet();
        }

        knownIndices.remove(index);
    }
}

@Override
public long getCount() {
    long count = 0;
    IndicesAdminClient admin = elastic.getClient().admin().indices();
    IndicesExistsRequestBuilder builder = admin.prepareExists(config.getMisc().getDindex().getName());
    if (builder.execute().actionGet().isExists()) {
        SearchResponse countResponse = elastic.getClient().prepareSearch(config.getMisc().getDindex().getName())
                .setTypes(config.getMisc().getCtype().getName())
                .setSearchType(SearchType.QUERY_THEN_FETCH)
                .setSize(0)
                .execute().actionGet();
        count = countResponse.getHits().getTotalHits();
    } else {
        logger.info("Index/Type does not exist or does not contain the record");
        System.exit(-1);
    }

    logger.info("Elastic Index/Type count: " + count);
    return count;
}

@Override
public void deleteIndex()
{
    final IndicesAdminClient indices = getClient().admin().indices();

    //
    // Create index if does not exist
    final String indexName = getIndexName();
    final boolean indexExists = indices
            .prepareExists(indexName)
            .get()
            .isExists();
    if (!indexExists)
    {
        // nothing to delete
        return;
    }

    final boolean acknowledged = indices.prepareDelete(indexName)
            .get()
            .isAcknowledged();
    if (!acknowledged)
    {
        throw new AdempiereException("Cannot delete index: " + indexName);
    }
}

private void createChatsIndex(IndicesAdminClient indices) {
    CreateIndexRequest createBuilder = new CreateIndexRequest("chats");
    try {
        // @formatter:off
        XContentBuilder mappingBuilder = XContentFactory.jsonBuilder()
            .startObject()
                .startObject("chat")
                    .startObject("properties")
                        .startObject("date")
                            .field("type", "long")
                        .endObject()
                        .startObject("type")
                            .field("type", "string")
                            .field("index", "not_analyzed")
                        .endObject()
                    .endObject()
                .endObject()
            .endObject();
        createBuilder.mapping("chat", mappingBuilder);
        // @formatter:on
    } catch (IOException e) {
        e.printStackTrace();
    }
    indices.create(createBuilder);
}

/**
 * @param indices The names of all indices that will exist.
 * @param mappings The index mappings.
 * @return An object to test.
 */
public ElasticsearchColumnMetadataDao setup(
        String[] indices,
        ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> mappings) {

  AdminClient adminClient = mock(AdminClient.class);
  IndicesAdminClient indicesAdminClient = mock(IndicesAdminClient.class);
  GetIndexRequestBuilder getIndexRequestBuilder = mock(GetIndexRequestBuilder.class);
  GetIndexResponse getIndexResponse = mock(GetIndexResponse.class);
  ActionFuture getMappingsActionFuture = mock(ActionFuture.class);
  GetMappingsResponse getMappingsResponse = mock(GetMappingsResponse.class);

  // setup the mocks so that a set of indices are available to the DAO
  when(adminClient.indices()).thenReturn(indicesAdminClient);
  when(indicesAdminClient.prepareGetIndex()).thenReturn(getIndexRequestBuilder);
  when(getIndexRequestBuilder.setFeatures()).thenReturn(getIndexRequestBuilder);
  when(getIndexRequestBuilder.get()).thenReturn(getIndexResponse);
  when(getIndexResponse.getIndices()).thenReturn(indices);

  // setup the mocks so that a set of mappings are available to the DAO
  when(indicesAdminClient.getMappings(any())).thenReturn(getMappingsActionFuture);
  when(getMappingsActionFuture.actionGet()).thenReturn(getMappingsResponse);
  when(getMappingsResponse.getMappings()).thenReturn(mappings);

  return new ElasticsearchColumnMetadataDao(adminClient);
}

/**
 * Deletes an index and block until deletion is complete.
 *
 * @param index The index to delete
 * @param client The client which points to the Elasticsearch instance
 * @throws InterruptedException if blocking thread is interrupted or index existence check failed
 * @throws java.util.concurrent.ExecutionException if index existence check failed
 * @throws IOException if deletion failed
 */
static void deleteIndex(String index, Client client)
        throws InterruptedException, java.util.concurrent.ExecutionException, IOException {
    IndicesAdminClient indices = client.admin().indices();
    IndicesExistsResponse indicesExistsResponse =
            indices.exists(new IndicesExistsRequest(index)).get();
    if (indicesExistsResponse.isExists()) {
        indices.prepareClose(index).get();
        // delete index is an asynchronous request, neither refresh or upgrade
        // delete all docs before starting tests. WaitForYellow() and delete directory are too slow,
        // so block thread until it is done (make it synchronous!!!)
        AtomicBoolean indexDeleted = new AtomicBoolean(false);
        AtomicBoolean waitForIndexDeletion = new AtomicBoolean(true);
        indices.delete(
                Requests.deleteIndexRequest(index),
                new DeleteActionListener(indexDeleted, waitForIndexDeletion));
        while (waitForIndexDeletion.get()) {
            Thread.sleep(100);
        }
        if (!indexDeleted.get()) {
            throw new IOException("Failed to delete index " + index);
        }
    }
}

private void SetupMocks(Client client, IndicesAdminClient indicesAdminClient, ImmutableOpenMap<String, IndexMetaData> indicesMap) {
    AdminClient adminClient = mock(AdminClient.class);
    ClusterAdminClient clusterAdminClient = mock(ClusterAdminClient.class);

    @SuppressWarnings("unchecked")
    ActionFuture<ClusterStateResponse> actionFuture = ((ActionFuture<ClusterStateResponse>) mock(ActionFuture.class));
    ClusterStateResponse response = mock(ClusterStateResponse.class);
    ClusterState state = mock(ClusterState.class);
    MetaData metaData = mock(MetaData.class);

    when(client.admin()).thenReturn(adminClient);
    when(adminClient.indices()).thenReturn(indicesAdminClient);
    when(adminClient.cluster()).thenReturn(clusterAdminClient);
    when(clusterAdminClient.state(Mockito.any(ClusterStateRequest.class))).thenReturn(actionFuture);
    when(actionFuture.actionGet()).thenReturn(response);
    when(response.getState()).thenReturn(state);
    when(state.getMetaData()).thenReturn(metaData);
    when(metaData.getIndices()).thenReturn(indicesMap);
}

public void setup() throws IOException, NoSuchFieldException, IllegalAccessException {
    String key;
    CreateIndexResponse ciResp;

    Reflections reflections = new Reflections("org.apache.usergrid.chop.webapp.dao");
    Set<Class<? extends Dao>> daoClasses = reflections.getSubTypesOf(Dao.class);

    IndicesAdminClient client = elasticSearchClient.getClient().admin().indices();

    for (Class<? extends Dao> daoClass : daoClasses) {

        key = daoClass.getDeclaredField("DAO_INDEX_KEY").get(null).toString();

        if (!client.exists(new IndicesExistsRequest(key)).actionGet().isExists()) {
            ciResp = client.create(new CreateIndexRequest(key)).actionGet();
            if (ciResp.isAcknowledged()) {
                LOG.debug("Index for key {} didn't exist, now created", key);
            } else {
                LOG.debug("Could not create index for key: {}", key);
            }
        } else {
            LOG.debug("Key {} already exists", key);
        }
    }
}

@Override
public Table execute() throws MetaModelException {
    final MutableTable table = getTable();
    final Map<String, ?> source = ElasticSearchUtils.getMappingSource(table);

    final ElasticSearchDataContext dataContext = getUpdateCallback().getDataContext();
    final IndicesAdminClient indicesAdmin = dataContext.getElasticSearchClient().admin().indices();
    final String indexName = dataContext.getIndexName();

    final PutMappingRequestBuilder requestBuilder =
            new PutMappingRequestBuilder(indicesAdmin, PutMappingAction.INSTANCE).setIndices(indexName)
                    .setType(table.getName());
    requestBuilder.setSource(source);
    final PutMappingResponse result = requestBuilder.execute().actionGet();

    logger.debug("PutMapping response: acknowledged={}", result.isAcknowledged());

    dataContext.getElasticSearchClient().admin().indices().prepareRefresh(indexName).get();

    final MutableSchema schema = (MutableSchema) getSchema();
    schema.addTable(table);
    return table;
}

public void startup() throws Exception {
    ImmutableSettings.Builder settings = ImmutableSettings.settingsBuilder();
    settings.put("node.name", "testnode");
    settings.put("gateway.type", "none");
    settings.put("path.data", "target/search-data");
    settings.put("http.enabled", true);

    settings.put("http.port", HTTP_PORT);
    // settings.put("index.compound_format", false);
    settings.put("transport.tcp.port", TRANSPORT_PORT);
    _node = NodeBuilder.nodeBuilder().settings(settings).clusterName(CLUSTER_NAME).data(true).local(false).node();

    try (Client client = _node.client()) {
        IndicesAdminClient indicesAdmin = client.admin().indices();
        if (!indicesAdmin.exists(new IndicesExistsRequest(INDEX_NAME)).actionGet().isExists()) {
            indicesAdmin.create(new CreateIndexRequest(INDEX_NAME)).actionGet();
        }
    }

    System.out.println("--- ElasticSearchTestServer started ---");
}

@Override
protected void startUp() throws Exception {
    IndicesAdminClient indices = esClient.admin().indices();
    IndicesExistsResponse exists = get(indices.exists(
            Requests.indicesExistsRequest(indexName)
    ));
    if (!exists.isExists()) {
        log.info("Creating index {}", indexName);
        get(indices.create(Requests.createIndexRequest(indexName)));
        get(indices.putMapping(Requests.putMappingRequest(indexName)
                .type(EsTopic.TYPE_NAME).source(EsTopic.getMapping())
        ));
    } else {
        log.info("Index {} exists", indexName);
    }
}

@Test
public void testPutMapping() {
    // Test data
    final String indexName = "Index_Name";
    final String documentType = "documentType";
    final String mappings = "mappings";
    final AdminClient adminClient = createMock(AdminClient.class);
    final IndicesAdminClient indicesAdminClient = createMock(IndicesAdminClient.class);
    final PutMappingRequestBuilder putMappingRequestBuilder = createMock(PutMappingRequestBuilder.class);
    final PutMappingResponse putMappingResponse = createMock(PutMappingResponse.class);
    // Reset
    resetAll();
    // Expectations
    expect(esClient.admin()).andReturn(adminClient);
    expect(adminClient.indices()).andReturn(indicesAdminClient);
    expect(indicesAdminClient.preparePutMapping(indexName)).andReturn(putMappingRequestBuilder);
    expect(putMappingRequestBuilder.setType(documentType)).andReturn(putMappingRequestBuilder);
    expect(putMappingRequestBuilder.setSource(mappings)).andReturn(putMappingRequestBuilder);
    expect(putMappingRequestBuilder.get()).andReturn(putMappingResponse);
    // Replay
    replayAll();
    // Run test scenario
    final boolean result = elasticsearchClientWrapper.putMapping(indexName, documentType, mappings);
    // Verify
    verifyAll();
    assertEquals(putMappingResponse.isAcknowledged(), result);
}

/**
 * 初始化索引
 *
 * @throws Exception
 */
private static void initIndex() throws Exception {
    String indice = esprop.getIndice();
    IndicesAdminClient c = client.admin().indices();
    //创建一个空的
    boolean a = c.prepareExists(indice).get().isExists();
    LOGGER.info("index {} isExists  {}",indice, a);
    if (!c.prepareExists(indice).get().isExists()) {
        CreateIndexResponse createIndexResponse =c.prepareCreate(indice).get();
        LOGGER.info("create index {}", createIndexResponse);
    }
    for (IndexType type : IndexType.values()) {
        TypesExistsResponse typesExistsResponse = c.typesExists(new TypesExistsRequest(new String[]{indice}, type.getDataName())).get();
        if (typesExistsResponse.isExists()) {
            continue;
        }
        String esMapper = type.getMapper();
        InputStream in = EsClientManager.class.getResourceAsStream(esMapper);
        String mappingStr = IOUtils.toString(in).trim();
        IOUtils.closeQuietly(in);
        c.preparePutMapping(indice).setType(type.getDataName()).setSource(mappingStr).get();
    }
}

@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());
}

@After
public void after() {
  final IndicesAdminClient indices = client.admin().indices();
  if(indices.prepareExists(INDEX).execute().actionGet().isExists()) {
    indices.prepareDelete(INDEX).execute().actionGet();
  }
}

@After
public void after() {
  final IndicesAdminClient indices = client.admin().indices();
  if(indices.prepareExists(INDEX).execute().actionGet().isExists()) {
    indices.prepareDelete(INDEX).execute().actionGet();
  }
}

@After
public void after() {
  final IndicesAdminClient indices = client.admin().indices();
  if(indices.prepareExists(INDEX).execute().actionGet().isExists()) {
    indices.prepareDelete(INDEX).execute().actionGet();
  }
}

@Test
public void shouldAutowire() {
  assertNotNull(client);
  final IndicesAdminClient indices = client.admin().indices();
  indices.prepareCreate("test").execute().actionGet();
  flush("test");
  indices.prepareDelete("test").execute().actionGet();
}

/**
 * Checks if KPI's elasticsearch Index and Type exists
 */
public KPIDataLoader assertESTypesExists()
{
    final IndicesAdminClient admin = elasticsearchClient.admin()
            .indices();

    //
    // Check index exists
    final String esSearchIndex = kpi.getESSearchIndex();
    final GetIndexResponse indexResponse = admin.prepareGetIndex()
            .addIndices(esSearchIndex)
            .get();
    final List<String> indexesFound = Arrays.asList(indexResponse.getIndices());
    if (!indexesFound.contains(esSearchIndex))
    {
        throw new AdempiereException("ES index '" + esSearchIndex + "' not found in " + indexesFound);
    }
    logger.debug("Indexes found: {}", indexesFound);

    //
    // Check type exists
    final String esTypes = kpi.getESSearchTypes();
    final boolean esTypesExists = admin.prepareTypesExists(esSearchIndex)
            .setTypes(kpi.getESSearchTypes())
            .get()
            .isExists();
    if (!esTypesExists)
    {
        throw new AdempiereException("Elasticseatch types " + esTypes + " does not exist");
    }

    // All good
    return this;
}

@Override
public void afterPropertiesSet() throws Exception {
    Map<String, String> map = new HashMap();
    //基础名称
    map.put("cluster.name", "my-application-A");
    Settings.Builder settings = Settings.builder().put(map);
    try {
        transportClient = TransportClient.builder().settings(settings).build()
                .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));

        IndicesAdminClient indicesAdminClient = transportClient.admin().indices();
        //查看索引是否存在,不存在就创建索引
        if(!checkExistsIndex(indicesAdminClient,INDEX_NAME)){
            indicesAdminClient.prepareCreate(INDEX_NAME).setSettings().execute().actionGet();
        }
        //查询mapping是否存在，已存在就不创建了
        GetMappingsResponse getMappingsResponse = indicesAdminClient.getMappings(new GetMappingsRequest().indices(INDEX_NAME)).actionGet();
        ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> indexToMappings = getMappingsResponse.getMappings();
       if(indexToMappings.get(INDEX_NAME).get(TIEABA_CONTENT_TYPE)==null) {
           //创建zk分词mapping
           PutMappingRequest mapping = Requests.putMappingRequest(INDEX_NAME).type(TIEABA_CONTENT_TYPE).source(createIKMapping(TIEABA_CONTENT_TYPE, TIEABA_CONTENT_FIELD).string());
           mapping.updateAllTypes(true);
           indicesAdminClient.putMapping(mapping).actionGet();
       }
    } catch (Exception e) {
       log.error("初始化 elasticsearch cliet error"+e.getLocalizedMessage());
    }
}