import io.searchbox.core.BulkResult; //导入依赖的package包/类
@Override
public void accept(HTTPBulkLoader.BulkReport report) {
  long tsNowMs = clock.currentTimeMillis();
  BulkResult result = report.bulkResult;
  if (!result.isSucceeded()) {
    if (result.getItems().size() == 0) {
      throw new SamzaException("Elasticsearch API error: " + result.getErrorMessage());
    }
    //Ignore version conflicts
    List<BulkResult.BulkResultItem> fatal = result.getFailedItems().stream().filter(item -> item.status != STATUS_CONFLICT).collect(Collectors.toList());
    if (fatal.size() > 0) {
      fatal.forEach(item -> logger.error(String.format("Error: index %s/%s, id %s, status %s, error %s",
          item.index, item.type, item.id, item.status, item.error)));
      throw new SamzaException(String.format("Elasticsearch bulk result contained %s errors", fatal.size()));
    }
  }
  logger.debug(String.format("Wrote %s actions to Elasticsearch system %s", result.getItems().size(), systemName));
  updateSuccessMetrics(report, tsNowMs);
}
 

import io.searchbox.core.BulkResult; //导入依赖的package包/类
@Override
public BulkResponse execute(Bulk bulk) throws IOException {
  final BulkResult result = client.execute(bulk);

  if (result.isSucceeded()) {
    return BulkResponse.success();
  }

  boolean retriable = true;

  final List<Key> versionConflicts = new ArrayList<>();
  final List<String> errors = new ArrayList<>();

  for (BulkResult.BulkResultItem item : result.getItems()) {
    if (item.error != null) {
      final ObjectNode parsedError = (ObjectNode) OBJECT_MAPPER.readTree(item.error);
      final String errorType = parsedError.get("type").asText("");
      if ("version_conflict_engine_exception".equals(errorType)) {
        versionConflicts.add(new Key(item.index, item.type, item.id));
      } else if ("mapper_parse_exception".equals(errorType)) {
        retriable = false;
        errors.add(item.error);
      } else {
        errors.add(item.error);
      }
    }
  }

  if (!versionConflicts.isEmpty()) {
    LOG.debug("Ignoring version conflicts for items: {}", versionConflicts);
    if (errors.isEmpty()) {
      // The only errors were version conflicts
      return BulkResponse.success();
    }
  }

  final String errorInfo = errors.isEmpty() ? result.getErrorMessage() : errors.toString();

  return BulkResponse.failure(retriable, errorInfo);
}
 

import io.searchbox.core.BulkResult; //导入依赖的package包/类
public int executeBulk(Bulk bulk) throws CouldNotCompleteBulkOperationException {
    try {

        BulkResult result = client.execute(bulk);
        return result.getFailedItems().size();

    } catch (IOException e) {
        throw new CouldNotCompleteBulkOperationException(e);
    }
}
 

import io.searchbox.core.BulkResult; //导入依赖的package包/类
@Test
public void testMetrics() throws Exception {
  long esWaitMs = 200;
  long receivedMs = 100;
  long eventTsMs = 50;
  long tsNow = 300;
  List<HTTPBulkLoader.SourcedActionRequest> requests = new ArrayList<>();
  requests.add(getIndexRequest(100, 50));
  requests.add(getIndexRequest(100, 50));
  requests.add(getIndexRequest(100, 50));
  requests.add(getIndexRequest(100, 50));
  requests.add(getIndexRequest(100, 50));

  List<BulkResult.BulkResultItem> items = new ArrayList<>();
  items.add(getItemIndexInsert());
  items.add(getItemIndexUpdate());
  items.add(getItemIndexConflict());
  items.add(getItemUpdate());
  items.add(getItemDelete());

  BulkResult bulkResult = mock(BulkResult.class);
  when(bulkResult.getItems()).thenReturn(items);
  HTTPBulkLoader.BulkReport report = new HTTPBulkLoader.BulkReport(bulkResult, HTTPBulkLoader.TriggerType.MAX_ACTIONS, esWaitMs, requests);

  when(flushListener.clock.currentTimeMillis()).thenReturn(tsNow);
  flushListener.accept(report);
  assertEquals(1, metrics.bulkSendSuccess.getCount());
  assertEquals(5.0, metrics.bulkSendBatchSize.getSnapshot().getMean(), 0.0001);
  assertEquals(esWaitMs, metrics.bulkSendWaitMs.getSnapshot().getMean(), 0.0001);
  assertEquals(1, metrics.inserts.getCount());
  assertEquals(2, metrics.updates.getCount());
  assertEquals(1, metrics.conflicts.getCount());
  assertEquals(1, metrics.deletes.getCount());
  assertEquals(0, metrics.triggerFlushCmd.getCount());
  assertEquals(1, metrics.triggerMaxActions.getCount());
  assertEquals(0, metrics.triggerMaxInterval.getCount());
  assertEquals(tsNow - eventTsMs, metrics.lagFromOriginMs.getSnapshot().getMean(), 0.0001);
  assertEquals(tsNow - receivedMs, metrics.lagFromReceiveMs.getSnapshot().getMean(), 0.0001);
}
 

import io.searchbox.core.BulkResult; //导入依赖的package包/类
/**
  * Process the next item in the queue.
  */
 protected void processQueue() {
     try {
         Collection<RequestMetric> batch = new ArrayList<>(this.batchSize);
         RequestMetric rm = queue.take();
         batch.add(rm);
queue.drainTo(batch, this.batchSize - 1);

Builder builder = new Bulk.Builder();
for (RequestMetric metric : batch) {
          Index index = new Index.Builder(metric).refresh(false)
                  .index(getIndexName())
                  .type("request").build(); //$NON-NLS-1$
	builder.addAction(index);
}
         
         BulkResult result = getClient().execute(builder.build());
         if (!result.isSucceeded()) {
             System.err.println("Failed to add metric(s) to ES: " + result.getErrorMessage()); //$NON-NLS-1$
         }
     } catch (Exception e) {
         // TODO better logging of this unlikely error
         System.err.println("Error adding metric to ES:"); //$NON-NLS-1$
         e.printStackTrace();
         return;
     }
 }
 

import io.searchbox.core.BulkResult; //导入依赖的package包/类
protected void updateSuccessMetrics(HTTPBulkLoader.BulkReport report, long tsNowMs) {
  metrics.bulkSendSuccess.inc();
  metrics.bulkSendBatchSize.update(report.requests.size());
  metrics.bulkSendWaitMs.update(report.esWaitMs);
  switch (report.triggerType) {
    case MAX_ACTIONS:
      metrics.triggerMaxActions.inc();
      break;
    case MAX_INTERVAL:
      metrics.triggerMaxInterval.inc();
      break;
    case FLUSH_CMD:
      metrics.triggerFlushCmd.inc();
      break;
  }

  int i = 0;
  for (Iterator<BulkResult.BulkResultItem> it = report.bulkResult.getItems().iterator(); it.hasNext(); i++) {
    BulkResult.BulkResultItem item = it.next();
    if (item.status == STATUS_CONFLICT) {
      metrics.conflicts.inc();
    }
    else {
      switch (item.operation) {
        case "index":
          if (item.status == STATUS_INDEX_DOC_INSERTED) {
            metrics.inserts.inc();
          }
          else {
            metrics.updates.inc();
          }
          break;
        case "create":
          metrics.inserts.inc();
          break;
        case "update":
          metrics.updates.inc();
          break;
        case "delete":
          metrics.deletes.inc();
      }
    }
    metrics.lagFromReceiveMs.update(tsNowMs - report.requests.get(i).request.receivedTsMs);
    Long eventTsMs = report.requests.get(i).request.key.getEventTsUnixMs();
    if (eventTsMs != null) {
      metrics.lagFromOriginMs.update(tsNowMs - eventTsMs);
    }
  }
}
 

import io.searchbox.core.BulkResult; //导入依赖的package包/类
private BulkResult.BulkResultItem getItemIndexInsert() {
  return new BulkResult(new Gson()).new BulkResultItem("index", "test", "test", "test", 201, "");
}
 

import io.searchbox.core.BulkResult; //导入依赖的package包/类
private BulkResult.BulkResultItem getItemIndexUpdate() {
  return new BulkResult(new Gson()).new BulkResultItem("index", "test", "test", "test", 200, "");
}
 

import io.searchbox.core.BulkResult; //导入依赖的package包/类
private BulkResult.BulkResultItem getItemIndexConflict() {
  return new BulkResult(new Gson()).new BulkResultItem("index", "test", "test", "test", 409, "");
}
 

import io.searchbox.core.BulkResult; //导入依赖的package包/类
private BulkResult.BulkResultItem getItemUpdate() {
  return new BulkResult(new Gson()).new BulkResultItem("update", "test", "test", "test", 200, "");
}
 

import io.searchbox.core.BulkResult; //导入依赖的package包/类
private BulkResult.BulkResultItem getItemDelete() {
  return new BulkResult(new Gson()).new BulkResultItem("delete", "test", "test", "test", 200, "");
}