相关环境
Elasticsearch 6.4.2

问题

服务需要对ES的rest-high-level-client的进行bulk代码的封装.

public Boolean batchInsertOrUpdate(BulkOpeRequest bulkOpeRequest) throws IOException {
        List<BatchRequest> docs = bulkOpeRequest.getDocs();
        if (CollectionUtil.isEmpty(docs)) {
            throw new RuntimeException("BatchRequest list is empty !");
        }
        BulkRequest bulkRequest = new BulkRequest();
        docs.forEach(batchRequest ->
                bulkRequest.add(
                        new IndexRequest(bulkOpeRequest.getIndex(), bulkOpeRequest.getType(), batchRequest.getId())
                                .source(XContentType.JSON, batchRequest.getData())
                )
        );
        BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        if (response.hasFailures()) {
            log.error(response.buildFailureMessage());
            return Boolean.FALSE;
        }
        return Boolean.TRUE;
    }

再进行调用的该方法插入数据的时候, 提示出错, 报错信息: The number of object passed must be even but was [1], 如图所示:

异常提示

解决步骤

首先此处根据source的源码, 可以看到, source需要传入的参数必须要成双, 这个是异常出现的关键.

public IndexRequest source(XContentType xContentType, Object... source) {
        if (source.length % 2 != 0) {
            throw new IllegalArgumentException("The number of object passed must be even but was [" + source.length + "]");
        }
        // other code
}

结合官方文档 给出的简单的例子来猜出, 应该是把需要的field和value平铺下来传入进去, 比如{"name": "zhangsan"}, 需要转化成 "name", "zhangsan" 这种形式. 于是对代码做了一个修改.

// other code...

  /**
    * 批量插入/更新文档
    * @return
  */
  public Boolean batchInsertOrUpdate(BulkOpeRequest bulkOpeRequest) throws IOException {
      List<BatchRequest> docs = bulkOpeRequest.getDocs();
      if (CollectionUtil.isEmpty(docs)) {
          throw new RuntimeException("BatchRequest list is empty !");
      }
      BulkRequest bulkRequest = new BulkRequest();
      docs.forEach(batchRequest ->
              bulkRequest.add(
                      new IndexRequest(bulkOpeRequest.getIndex(), bulkOpeRequest.getType(), batchRequest.getId())
                              .source(XContentType.JSON, toArgs(batchRequest.getData()))
              )
      );
      BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
      if (response.hasFailures()) {
          log.error(response.buildFailureMessage());
          return Boolean.FALSE;
      }
      return Boolean.TRUE;
  }

  /**
   * 将批量数据转换成es参数数组
   * @param data
   * @return
   */
  private Object[] toArgs(Map<String, Object> data) {
      List<Object> args = Lists.newArrayList();
      if (CollectionUtil.isEmpty(data)) {
          throw new RuntimeException("批量操作数据不可为空 !");
      }
      data.forEach((key, value) -> {
          args.add(key);
          args.add(value);
      });
      return args.toArray();
  }

// other code...

经过测试, 问题解决.