Github user nickwallen commented on a diff in the pull request:
https://github.com/apache/metron/pull/1254#discussion_r236843942
--- Diff:
metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/bulk/ElasticsearchBulkDocumentWriter.java
---
@@ -0,0 +1,184 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.metron.elasticsearch.bulk;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.metron.elasticsearch.client.ElasticsearchClient;
+import org.apache.metron.indexing.dao.update.Document;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * Writes documents to an Elasticsearch index in bulk.
+ *
+ * @param <D> The type of document to write.
+ */
+public class ElasticsearchBulkDocumentWriter<D extends Document>
implements BulkDocumentWriter<D> {
+
+ /**
+ * A {@link Document} along with the index it will be written to.
+ */
+ private class Indexable {
+ D document;
+ String index;
+
+ public Indexable(D document, String index) {
+ this.document = document;
+ this.index = index;
+ }
+ }
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ private Optional<SuccessListener> onSuccess;
+ private Optional<FailureListener> onFailure;
+ private ElasticsearchClient client;
+ private List<Indexable> documents;
+
+ public ElasticsearchBulkDocumentWriter(ElasticsearchClient client) {
+ this.client = client;
+ this.onSuccess = Optional.empty();
+ this.onFailure = Optional.empty();
+ this.documents = new ArrayList<>();
+ }
+
+ @Override
+ public void onSuccess(SuccessListener<D> onSuccess) {
+ this.onSuccess = Optional.of(onSuccess);
+ }
+
+ @Override
+ public void onFailure(FailureListener<D> onFailure) {
+ this.onFailure = Optional.of(onFailure);
+ }
+
+ @Override
+ public void addDocument(D document, String index) {
+ documents.add(new Indexable(document, index));
+ LOG.debug("Adding document to batch; document={}, index={}",
document, index);
+ }
+
+ @Override
+ public void write() {
+ try {
+ // create an index request for each document
+ List<DocWriteRequest> requests = documents
--- End diff --
It turned out to be much cleaner to get rid of the streams. So double-win.
Thanks.
---
