Fixing StreamsMetadata xcontent serialization

server/src/main/java/org/elasticsearch/cluster/ClusterModule.java

@@ -380,6 +380,13 @@ public static List<NamedXContentRegistry.Entry> getNamedXWriteables() {
                 DesiredNodesMetadata::fromXContent
             )
         );
+        entries.add(
+            new NamedXContentRegistry.Entry(
+                Metadata.ProjectCustom.class,
+                new ParseField(StreamsMetadata.TYPE),
+                StreamsMetadata::fromXContent
+            )
+        );
         return entries;
     }
 

server/src/main/java/org/elasticsearch/cluster/metadata/StreamsMetadata.java

@@ -17,7 +17,10 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
+import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
 import java.util.EnumSet;
@@ -32,6 +35,14 @@ public class StreamsMetadata extends AbstractNamedDiffable<Metadata.ProjectCusto
 
     public static final String TYPE = "streams";
     public static final StreamsMetadata EMPTY = new StreamsMetadata(false);
+    private static final ParseField LOGS_ENABLED = new ParseField("logs_enabled");
+    private static final ConstructingObjectParser<StreamsMetadata, Void> PARSER = new ConstructingObjectParser<>(TYPE, false, args -> {
+        boolean logsEnabled = (boolean) args[0];
+        return new StreamsMetadata(logsEnabled);
+    });
+    static {
+        PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), LOGS_ENABLED);
+    }
 
     public boolean logsEnabled;
 
@@ -79,7 +90,9 @@ public void writeTo(StreamOutput out) throws IOException {
 
     @Override
     public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params) {
-        return Iterators.concat(ChunkedToXContentHelper.chunk((builder, bParams) -> builder.field("logs_enabled", logsEnabled)));
+        return Iterators.concat(
+            ChunkedToXContentHelper.chunk((builder, bParams) -> builder.field(LOGS_ENABLED.getPreferredName(), logsEnabled))
+        );
     }
 
     @Override
@@ -95,4 +108,8 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Objects.hashCode(logsEnabled);
     }
+
+    public static StreamsMetadata fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
 }

server/src/test/java/org/elasticsearch/cluster/metadata/StreamsMetadataTests.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.cluster.metadata;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractChunkedSerializingTestCase;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+
+public class StreamsMetadataTests extends AbstractChunkedSerializingTestCase<StreamsMetadata> {
+    @Override
+    protected StreamsMetadata doParseInstance(XContentParser parser) throws IOException {
+        return StreamsMetadata.fromXContent(parser);
+    }
+
+    @Override
+    protected Writeable.Reader<StreamsMetadata> instanceReader() {
+        return StreamsMetadata::new;
+    }
+
+    @Override
+    protected StreamsMetadata createTestInstance() {
+        return new StreamsMetadata(randomBoolean());
+    }
+
+    @Override
+    protected StreamsMetadata mutateInstance(StreamsMetadata instance) throws IOException {
+        return new StreamsMetadata(instance.logsEnabled == false);
+    }
+}