Fix empty_value handling in CsvProcessor (#55649) (#55969)

* Fix empty_value handling in CsvProcessor

Due to bug in `CsvProcessor.Factory` it was impossible to specify `empty_value`.
This change fixes that and adds relevant test.

Closes #55643

* assert changed

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/CsvProcessor.java

@@ -45,13 +45,14 @@ public final class CsvProcessor extends AbstractProcessor {
 
     public static final String TYPE = "csv";
 
-    private final String field;
-    private final String[] headers;
-    private final boolean trim;
-    private final char quote;
-    private final char separator;
-    private final boolean ignoreMissing;
-    private final Object emptyValue;
+    //visible for testing
+    final String field;
+    final String[] headers;
+    final boolean trim;
+    final char quote;
+    final char separator;
+    final boolean ignoreMissing;
+    final Object emptyValue;
 
     CsvProcessor(String tag, String field, String[] headers, boolean trim, char separator, char quote, boolean ignoreMissing,
                  Object emptyValue) {
@@ -101,7 +102,7 @@ public CsvProcessor create(Map<String, org.elasticsearch.ingest.Processor.Factor
             }
             boolean trim = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "trim", false);
             Object emptyValue = null;
-            if(config.containsKey("emptyValue")){
+            if(config.containsKey("empty_value")){
                 emptyValue = ConfigurationUtils.readObject(TYPE, processorTag, config, "empty_value");
             }
             boolean ignoreMissing = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "ignore_missing", false);

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/CsvProcessorFactoryTests.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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.elasticsearch.ingest.common;
+
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.Collections;
+import java.util.HashMap;
+
+import static java.util.Collections.emptyMap;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+
+public class CsvProcessorFactoryTests extends ESTestCase {
+
+    public void testProcessorIsCreated() {
+        CsvProcessor.Factory factory = new CsvProcessor.Factory();
+        HashMap<String, Object> properties = new HashMap<>();
+        properties.put("field", "field");
+        properties.put("target_fields", Collections.singletonList("target"));
+        properties.put("quote", "|");
+        properties.put("separator", "/");
+        properties.put("empty_value", "empty");
+        properties.put("trim", true);
+        properties.put("ignore_missing", true);
+        CsvProcessor csv = factory.create(null, "csv", properties);
+        assertThat(csv, notNullValue());
+        assertThat(csv.field, equalTo("field"));
+        assertThat(csv.headers, equalTo(new String[]{"target"}));
+        assertThat(csv.quote, equalTo('|'));
+        assertThat(csv.separator, equalTo('/'));
+        assertThat(csv.emptyValue, equalTo("empty"));
+        assertThat(csv.trim, equalTo(true));
+        assertThat(csv.ignoreMissing, equalTo(true));
+        assertThat(properties, is(emptyMap()));
+    }
+}