public BulkProcessor getBulkProcessor(Client client, BulkProcessor.Listener listener) {
    BulkProcessor.Builder builder = BulkProcessor.builder(client, listener);

    // Concurrent requests set to 0 to ensure ordering of documents is maintained in batches.
    // This also means BulkProcessor#flush() is blocking as is also required.
    builder.setConcurrentRequests(0);

    config.getBulkFlushMaxActions().ifPresent(builder::setBulkActions);
    config.getBulkFlushMaxSizeMB().ifPresent(size ->
      builder.setBulkSize(new ByteSizeValue(size, ByteSizeUnit.MB))
    );
    config.getBulkFlushIntervalMS().ifPresent(interval ->
      builder.setFlushInterval(TimeValue.timeValueMillis(interval))
    );

    return builder.build();
  }
Creates Elasticsearch {@link BulkProcessor} instances based on properties from the Samza job.
samza-elasticsearch/src/main/java/org/apache/samza/system/elasticsearch/BulkProcessorFactory.java