Fix simulate pipeline API

The simulate pipeline API uses a special deep copy method to write out
its content via toXContent - this method does not support writing out
sets, resulting in an error as this processor writes out data using
sets.

This commit fixes this issue by replacing the set with a list, so that
the serialization works again, at the small price of creating more
objects when running.

Closes #27

Author: spinscale

src/main/java/de/spinscale/elasticsearch/ingest/opennlp/OpenNlpProcessor.java

@@ -22,9 +22,10 @@
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
 
-import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -34,15 +35,14 @@
 
 public class OpenNlpProcessor extends AbstractProcessor {
 
-    public static final String TYPE = "opennlp";
+    static final String TYPE = "opennlp";
 
     private final OpenNlpService openNlpService;
     private final String sourceField;
     private final String targetField;
     private final Set<String> fields;
 
-    OpenNlpProcessor(OpenNlpService openNlpService, String tag, String sourceField, String targetField, Set<String> fields) throws
-            IOException {
+    OpenNlpProcessor(OpenNlpService openNlpService, String tag, String sourceField, String targetField, Set<String> fields) {
         super(tag);
         this.openNlpService = openNlpService;
         this.sourceField = sourceField;
@@ -51,7 +51,7 @@ public class OpenNlpProcessor extends AbstractProcessor {
     }
 
     @Override
-    public void execute(IngestDocument ingestDocument) throws Exception {
+    public void execute(IngestDocument ingestDocument) {
         String content = ingestDocument.getFieldValue(sourceField, String.class);
 
         if (Strings.hasLength(content)) {
@@ -63,7 +63,15 @@ public void execute(IngestDocument ingestDocument) throws Exception {
                 merge(entities, field, data);
             }
 
-            ingestDocument.setFieldValue(targetField, entities);
+            // convert set to list, otherwise toXContent serialization in simulate pipeline fails
+            Map<String, List<String>> entitiesToStore = new HashMap<>();
+            Iterator<Map.Entry<String, Set<String>>> iterator = entities.entrySet().iterator();
+            while (iterator.hasNext()) {
+                Map.Entry<String, Set<String>> entry = iterator.next();
+                entitiesToStore.put(entry.getKey(), new ArrayList<>(entry.getValue()));
+            }
+
+            ingestDocument.setFieldValue(targetField, entitiesToStore);
         }
     }
 
@@ -81,8 +89,7 @@ public static final class Factory implements Processor.Factory {
         }
 
         @Override
-        public OpenNlpProcessor create(Map<String, Processor.Factory> registry, String processorTag, Map<String, Object> config)
-                throws Exception {
+        public OpenNlpProcessor create(Map<String, Processor.Factory> registry, String processorTag, Map<String, Object> config) {
             String field = readStringProperty(TYPE, processorTag, config, "field");
             String targetField = readStringProperty(TYPE, processorTag, config, "target_field", "entities");
             List<String> fields = readOptionalList(TYPE, processorTag, config, "fields");
@@ -104,8 +111,9 @@ private static void mergeExisting(Map<String, Set<String>> entities, IngestDocum
     private static void merge(Map<String, Set<String>> map, String key, Set<String> values) {
         if (values.size() == 0) return;
 
-        if (map.containsKey(key))
+        if (map.containsKey(key)) {
             values.addAll(map.get(key));
+        }
 
         map.put(key, values);
     }

src/test/java/de/spinscale/elasticsearch/ingest/opennlp/OpenNlpProcessorTests.java

@@ -17,39 +17,46 @@
 
 package de.spinscale.elasticsearch.ingest.opennlp;
 
+import org.elasticsearch.action.ingest.SimulateProcessorResult;
+import org.elasticsearch.common.io.PathUtils;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.RandomDocumentPicks;
 import org.elasticsearch.test.ESTestCase;
-import org.junit.Before;
+import org.junit.BeforeClass;
 
-import java.io.IOException;
+import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.not;
 
 public class OpenNlpProcessorTests extends ESTestCase {
 
-    private OpenNlpService service;
+    private static OpenNlpService service;
 
-    @Before
-    public void createOpenNlpService() throws IOException {
+    @BeforeClass
+    public static void createOpenNlpService() throws Exception {
         Settings settings = Settings.builder()
                 .put("ingest.opennlp.model.file.names", "en-ner-persons.bin")
                 .put("ingest.opennlp.model.file.locations", "en-ner-locations.bin")
                 .put("ingest.opennlp.model.file.dates", "en-ner-dates.bin")
                 .build();
 
-        service = new OpenNlpService(getDataPath("/models/en-ner-persons.bin").getParent(), settings).start();
+        Path path = PathUtils.get(OpenNlpProcessorTests.class.getResource("/models/en-ner-persons.bin").toURI());
+        service = new OpenNlpService(path.getParent(), settings).start();
     }
 
     public void testThatExtractionsWork() throws Exception {
@@ -110,7 +117,20 @@ public void testConstructorNoFieldsSpecified() throws Exception {
         assertThatHasElements(entityData, "names", "Kobe Bryant", "Michael Jordan");
         assertThatHasElements(entityData, "dates", "Yesterday");
         assertThatHasElements(entityData, "locations", "Munich", "New York");
+    }
+
+    public void testToXContent() throws Exception {
+        OpenNlpProcessor processor = new OpenNlpProcessor(service, randomAlphaOfLength(10), "source_field", "target_field",
+                new HashSet<>(Arrays.asList("names", "dates", "locations")));
+
+        IngestDocument ingestDocument = getIngestDocument();
+        processor.execute(ingestDocument);
+
+        SimulateProcessorResult result = new SimulateProcessorResult("tag", ingestDocument);
 
+        try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
+            result.toXContent(builder, ToXContent.EMPTY_PARAMS);
+        }
     }
 
     private Map<String, Object> getIngestDocumentData(OpenNlpProcessor processor) throws Exception {
@@ -138,15 +158,16 @@ private Map<String, Object> getIngestDocumentData(IngestDocument ingestDocument)
     }
 
     private void assertThatHasElements(Map<String, Object> entityData, String field, String ... items) {
-        Set<String> values = getValues(entityData, field);
+        List<String> values = getValues(entityData, field);
+        assertThat(values, hasSize(items.length));
         assertThat(values, containsInAnyOrder(items));
     }
 
-    private Set<String> getValues(Map<String, Object> entityData, String field) {
+    private List<String> getValues(Map<String, Object> entityData, String field) {
         assertThat(entityData, hasKey(field));
-        assertThat(entityData.get(field), instanceOf(Set.class));
+        assertThat(entityData.get(field), instanceOf(List.class));
         @SuppressWarnings("unchecked")
-        Set<String> values = (Set<String>) entityData.get(field);
+        List<String> values = (List<String>) entityData.get(field);
         return values;
     }
 }

src/test/resources/rest-api-spec/test/ingest_opennlp/30_opennlp_simulate_api.yml

@@ -0,0 +1,29 @@
+---
+"Test opennlp processor with defaults":
+  - do:
+      ingest.put_pipeline:
+        id: "my_pipeline"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "opennlp" : {
+                  "field" : "field1"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      ingest.simulate:
+        id: "my_pipeline"
+        body: {docs: [ { _source: { field1: "Kobe Bryant was one of the best basketball players of all times. Not even Michael Jordan has ever scored 81 points in one game. Munich is really an awesome city, but New York is as well. Yesterday has been the hottest day of the year." } } ] }
+
+  - match: { docs.0.doc._source.field1: "Kobe Bryant was one of the best basketball players of all times. Not even Michael Jordan has ever scored 81 points in one game. Munich is really an awesome city, but New York is as well. Yesterday has been the hottest day of the year." }
+  - length: { docs.0.doc._source.entities: 3 }
+  - match: { docs.0.doc._source.entities.dates: [ "Yesterday" ] }
+  - match: { docs.0.doc._source.entities.names: [ "Kobe Bryant", "Michael Jordan" ] }
+  - match: { docs.0.doc._source.entities.locations: [ "Munich", "New York" ] }
+