1 files changed, 36 insertions, 8 deletions
diff --git a/core/src/main/java/org/elasticsearch/ingest/PipelineConfiguration.java b/core/src/main/java/org/elasticsearch/ingest/PipelineConfiguration.java
index 5cb9f6111f..3e874a0fd1 100644
--- a/core/src/main/java/org/elasticsearch/ingest/PipelineConfiguration.java
+++ b/core/src/main/java/org/elasticsearch/ingest/PipelineConfiguration.java
@@ -19,6 +19,7 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.cluster.AbstractDiffable;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.common.ParseField;
@@ -29,10 +30,13 @@ import org.elasticsearch.common.xcontent.ContextParser;
 import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentType;
 
 import java.io.IOException;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * Encapsulates a pipeline's id and configuration as a blob
@@ -45,8 +49,9 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
         PARSER.declareField((parser, builder, aVoid) -> {
             XContentBuilder contentBuilder = XContentBuilder.builder(parser.contentType().xContent());
             XContentHelper.copyCurrentStructure(contentBuilder.generator(), parser);
-            builder.setConfig(contentBuilder.bytes());
+            builder.setConfig(contentBuilder.bytes(), contentBuilder.contentType());
         }, new ParseField("config"), ObjectParser.ValueType.OBJECT);
+
     }
 
     public static ContextParser<Void, PipelineConfiguration> getParser() {
@@ -56,17 +61,19 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
 
         private String id;
         private BytesReference config;
+        private XContentType xContentType;
 
         void setId(String id) {
             this.id = id;
         }
 
-        void setConfig(BytesReference config) {
+        void setConfig(BytesReference config, XContentType xContentType) {
             this.config = config;
+            this.xContentType = xContentType;
         }
 
         PipelineConfiguration build() {
-            return new PipelineConfiguration(id, config);
+            return new PipelineConfiguration(id, config, xContentType);
         }
     }
 
@@ -75,10 +82,12 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
     // and the way the map of maps config is read requires a deep copy (it removes instead of gets entries to check for unused options)
     // also the get pipeline api just directly returns this to the caller
     private final BytesReference config;
+    private final XContentType xContentType;
 
-    public PipelineConfiguration(String id, BytesReference config) {
-        this.id = id;
-        this.config = config;
+    public PipelineConfiguration(String id, BytesReference config, XContentType xContentType) {
+        this.id = Objects.requireNonNull(id);
+        this.config = Objects.requireNonNull(config);
+        this.xContentType = Objects.requireNonNull(xContentType);
     }
 
     public String getId() {
@@ -86,7 +95,17 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
     }
 
     public Map<String, Object> getConfigAsMap() {
-        return XContentHelper.convertToMap(config, true).v2();
+        return XContentHelper.convertToMap(config, true, xContentType).v2();
+    }
+
+    // pkg-private for tests
+    XContentType getXContentType() {
+        return xContentType;
+    }
+
+    // pkg-private for tests
+    BytesReference getConfig() {
+        return config;
     }
 
     @Override
@@ -99,7 +118,13 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
     }
 
     public static PipelineConfiguration readFrom(StreamInput in) throws IOException {
-        return new PipelineConfiguration(in.readString(), in.readBytesReference());
+        if (in.getVersion().after(Version.V_5_3_0_UNRELEASED)) { // TODO update to onOrAfter after backporting
+            return new PipelineConfiguration(in.readString(), in.readBytesReference(), XContentType.readFrom(in));
+        } else {
+            final String id = in.readString();
+            final BytesReference config = in.readBytesReference();
+            return new PipelineConfiguration(id, config, XContentFactory.xContentType(config));
+        }
     }
 
     public static Diff<PipelineConfiguration> readDiffFrom(StreamInput in) throws IOException {
@@ -110,6 +135,9 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(id);
         out.writeBytesReference(config);
+        if (out.getVersion().after(Version.V_5_3_0_UNRELEASED)) { // TODO update to onOrAfter after backporting
+            xContentType.writeTo(out);
+        }
     }
 
     @Override