Github user cestella commented on a diff in the pull request:
https://github.com/apache/incubator-metron/pull/286#discussion_r81335456
--- Diff:
metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/writer/ElasticsearchWriter.java
---
@@ -166,14 +166,37 @@ public void write(String sensorType,
WriterConfiguration configurations, Iterabl
}
- BulkResponse resp = bulkRequest.execute().actionGet();
-
- if (resp.hasFailures()) {
+ BulkResponse bulkResponse = bulkRequest.execute().actionGet();
- throw new Exception(resp.buildFailureMessage());
+ return buildWriteReponse(tuples, bulkResponse);
+ }
+ protected BulkWriterResponse buildWriteReponse(Iterable<Tuple> tuples,
BulkResponse bulkResponse) throws Exception {
+ // Elasticsearch responses are in the same order as the request,
giving us an implicit mapping with Tuples
+ BulkWriterResponse writerResponse = new BulkWriterResponse();
+ if (bulkResponse.hasFailures()) {
+ Iterator<BulkItemResponse> respIter = bulkResponse.iterator();
+ Iterator<Tuple> tupleIter = tuples.iterator();
+ while (respIter.hasNext() && tupleIter.hasNext()) {
+ BulkItemResponse item = respIter.next();
+ Tuple tuple = tupleIter.next();
+
+ if (item.isFailed()) {
+ writerResponse.addError(item.getFailure().getCause(), tuple);
+ } else {
+ writerResponse.addSuccess(tuple);
+ }
+
+ // Should never happen, so fail the entire batch if it does.
+ if (respIter.hasNext() != tupleIter.hasNext()) {
+ throw new Exception(bulkResponse.buildFailureMessage());
--- End diff --
Perhaps a named exception? `IllegalStateException`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
