Configurable media_type for mustache template encoding on append processor

docs/reference/ingest/processors/append.asciidoc

@@ -15,10 +15,13 @@ Accepts a single value or an array of values.
 [options="header"]
 |======
 | Name      | Required  | Default  | Description
-| `field`  | yes       | -        | The field to be appended to. Supports <<template-snippets,template snippets>>.
-| `value`  | yes       | -        | The value to be appended. Supports <<template-snippets,template snippets>>.
+| `field`   | yes       | -        | The field to be appended to. Supports <<template-snippets,template snippets>>.
+| `value`   | yes       | -        | The value to be appended. Supports <<template-snippets,template snippets>>.
 | `allow_duplicates` | no | true  | If `false`, the processor does not append
 values already present in the field.
+| `media_type` | no | `application/json` | The media type for encoding `value`. Applies only when `value` is a
+<<template-snippets,template snippet>>. Must be one of `application/json`, `text/plain`, or
+`application/x-www-form-urlencoded`.
 include::common-options.asciidoc[]
 |======
 

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

@@ -13,6 +13,7 @@
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.ValueSource;
+import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.script.TemplateScript;
 
@@ -71,10 +72,15 @@ public AppendProcessor create(Map<String, Processor.Factory> registry, String pr
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             Object value = ConfigurationUtils.readObject(TYPE, processorTag, config, "value");
             boolean allowDuplicates = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "allow_duplicates", true);
-            TemplateScript.Factory compiledTemplate = ConfigurationUtils.compileTemplate(TYPE, processorTag,
-                "field", field, scriptService);
-            return new AppendProcessor(processorTag, description, compiledTemplate, ValueSource.wrap(value, scriptService),
-                allowDuplicates);
+            TemplateScript.Factory compiledTemplate = ConfigurationUtils.compileTemplate(TYPE, processorTag, "field", field, scriptService);
+            String mediaType = ConfigurationUtils.readMediaTypeProperty(TYPE, processorTag, config, "media_type", "application/json");
+            return new AppendProcessor(
+                processorTag,
+                description,
+                compiledTemplate,
+                ValueSource.wrap(value, scriptService, Map.of(Script.CONTENT_TYPE_OPTION, mediaType)),
+                allowDuplicates
+            );
         }
     }
 }

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

@@ -10,6 +10,7 @@
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.TestTemplateService;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
@@ -20,6 +21,7 @@
 import java.util.Map;
 
 import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.Matchers.containsString;
 
 public class AppendProcessorFactoryTests extends ESTestCase {
 
@@ -92,4 +94,26 @@ public void testInvalidMustacheTemplate() throws Exception {
         assertThat(exception.getMessage(), equalTo("java.lang.RuntimeException: could not compile script"));
         assertThat(exception.getMetadata("es.processor_tag").get(0), equalTo(processorTag));
     }
+
+    public void testMediaType() throws Exception {
+        // valid media type
+        String expectedMediaType = randomFrom(ConfigurationUtils.VALID_MEDIA_TYPES);
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", "field1");
+        config.put("value", "value1");
+        config.put("media_type", expectedMediaType);
+        String processorTag = randomAlphaOfLength(10);
+        AppendProcessor appendProcessor = factory.create(null, processorTag, null, config);
+        assertThat(appendProcessor.getTag(), equalTo(processorTag));
+
+        // invalid media type
+        expectedMediaType = randomValueOtherThanMany(m -> Arrays.asList(ConfigurationUtils.VALID_MEDIA_TYPES).contains(m),
+            () -> randomAlphaOfLengthBetween(5, 9));
+        final Map<String, Object> config2 = new HashMap<>();
+        config2.put("field", "field1");
+        config2.put("value", "value1");
+        config2.put("media_type", expectedMediaType);
+        ElasticsearchException e = expectThrows(ElasticsearchException.class, () -> factory.create(null, processorTag, null, config2));
+        assertThat(e.getMessage(), containsString("property does not contain a supported media type [" + expectedMediaType + "]"));
+    }
 }