Chunked X-Content Serialisation for ClusterState Customs (#92095)

Adds chunking to all CS metadata customs. Added a few utilities to make this a little less painful to write and read.
Admittedly didn't add tests to all the chunking implementations since I don't think they really matter all that much size wise with the exception of `DataStreamMetadata` which I'd like to take one step further in a follow-up anyway (the index name lists can become very long and should be individually chunked, but that would make this PR even more complicated than it already is).

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/get/GetRepositoriesResponse.java

@@ -13,6 +13,7 @@ import org.elasticsearch.cluster.metadata.RepositoriesMetadata;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ChunkedToXContent;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
@@ -55,7 +56,8 @@ public class GetRepositoriesResponse extends ActionResponse implements ToXConten
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
-        repositories.toXContent(builder, new DelegatingMapParams(Map.of(RepositoriesMetadata.HIDE_GENERATIONS_PARAM, "true"), params));
+        ChunkedToXContent.wrapAsXContentObject(repositories)
+            .toXContent(builder, new DelegatingMapParams(Map.of(RepositoriesMetadata.HIDE_GENERATIONS_PARAM, "true"), params));
         builder.endObject();
         return builder;
     }

server/src/main/java/org/elasticsearch/cluster/coordination/ElasticsearchNodeCommand.java

@@ -28,6 +28,7 @@ import org.elasticsearch.cluster.metadata.ComposableIndexTemplateMetadata;
 import org.elasticsearch.cluster.metadata.DataStreamMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.cli.EnvironmentAwareCommand;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
@@ -37,6 +38,7 @@ import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.env.NodeMetadata;
 import org.elasticsearch.gateway.PersistedClusterStateService;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 
@@ -45,6 +47,7 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 
@@ -229,8 +232,8 @@ public abstract class ElasticsearchNodeCommand extends EnvironmentAwareCommand {
         }
 
         @Override
-        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-            return builder.mapContents(contents);
+        public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+            return Iterators.single(((builder, params) -> builder.mapContents(contents)));
         }
     }
 

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

@@ -15,14 +15,16 @@ import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.common.Strings;
 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.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 
@@ -98,13 +100,8 @@ public class ComponentTemplateMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.startObject(COMPONENT_TEMPLATE.getPreferredName());
-        for (Map.Entry<String, ComponentTemplate> template : componentTemplates.entrySet()) {
-            builder.field(template.getKey(), template.getValue(), params);
-        }
-        builder.endObject();
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.xContentValuesMap(COMPONENT_TEMPLATE.getPreferredName(), componentTemplates);
     }
 
     @Override

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

@@ -15,14 +15,16 @@ import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.common.Strings;
 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.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 
@@ -99,13 +101,8 @@ public class ComposableIndexTemplateMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.startObject(INDEX_TEMPLATE.getPreferredName());
-        for (Map.Entry<String, ComposableIndexTemplate> template : indexTemplates.entrySet()) {
-            builder.field(template.getKey(), template.getValue(), params);
-        }
-        builder.endObject();
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.xContentValuesMap(INDEX_TEMPLATE.getPreferredName(), indexTemplates);
     }
 
     @Override

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

@@ -15,12 +15,14 @@ import org.elasticsearch.cluster.DiffableUtils;
 import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentFactory;
 import org.elasticsearch.xcontent.XContentParser;
 
@@ -28,6 +30,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.EnumSet;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -222,14 +225,13 @@ public class DataStreamMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.xContentValuesMap(DATA_STREAM.getPreferredName(), dataStreams);
-        builder.startObject(DATA_STREAM_ALIASES.getPreferredName());
-        for (Map.Entry<String, DataStreamAlias> dataStream : dataStreamAliases.entrySet()) {
-            dataStream.getValue().toXContent(builder, params);
-        }
-        builder.endObject();
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return Iterators.concat(
+            ChunkedToXContentHelper.xContentValuesMap(DATA_STREAM.getPreferredName(), dataStreams),
+            ChunkedToXContentHelper.startObject(DATA_STREAM_ALIASES.getPreferredName()),
+            dataStreamAliases.values().iterator(),
+            ChunkedToXContentHelper.endObject()
+        );
     }
 
     @Override

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

@@ -12,16 +12,18 @@ import org.elasticsearch.Version;
 import org.elasticsearch.cluster.AbstractNamedDiffable;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.NamedDiff;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.Objects;
 
 public class DesiredNodesMetadata extends AbstractNamedDiffable<Metadata.Custom> implements Metadata.Custom {
@@ -67,9 +69,8 @@ public class DesiredNodesMetadata extends AbstractNamedDiffable<Metadata.Custom>
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.field(LATEST_FIELD.getPreferredName(), latestDesiredNodes, params);
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return Iterators.single((builder, params) -> builder.field(LATEST_FIELD.getPreferredName(), latestDesiredNodes, params));
     }
 
     public static DesiredNodesMetadata fromClusterState(ClusterState clusterState) {

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

@@ -17,10 +17,12 @@ import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.time.DateFormatter;
+import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.xcontent.ContextParser;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
@@ -32,6 +34,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
 
@@ -123,12 +126,8 @@ public final class IndexGraveyard implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
-        builder.startArray(TOMBSTONES_FIELD.getPreferredName());
-        for (Tombstone tombstone : tombstones) {
-            tombstone.toXContent(builder, params);
-        }
-        return builder.endArray();
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.array(TOMBSTONES_FIELD.getPreferredName(), tombstones);
     }
 
     public static IndexGraveyard fromXContent(final XContentParser parser) throws IOException {

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

@@ -37,6 +37,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.ArrayUtils;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.common.xcontent.ChunkedToXContent;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentParserUtils;
 import org.elasticsearch.core.Nullable;
@@ -137,7 +138,7 @@ public class Metadata extends AbstractCollection<IndexMetadata> implements Diffa
      * Custom metadata that persists (via XContent) across restarts. The deserialization method for each implementation must be registered
      * with the {@link NamedXContentRegistry}.
      */
-    public interface Custom extends NamedDiffable<Custom>, ToXContentFragment {
+    public interface Custom extends NamedDiffable<Custom>, ChunkedToXContent {
 
         EnumSet<XContentContext> context();
 
@@ -2508,7 +2509,7 @@ public class Metadata extends AbstractCollection<IndexMetadata> implements Diffa
             for (Map.Entry<String, Custom> cursor : metadata.customs().entrySet()) {
                 if (cursor.getValue().context().contains(context)) {
                     builder.startObject(cursor.getKey());
-                    cursor.getValue().toXContent(builder, params);
+                    ChunkedToXContent.wrapAsXContentObject(cursor.getValue()).toXContent(builder, params);
                     builder.endObject();
                 }
             }

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

@@ -16,14 +16,16 @@ import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.SimpleDiffable;
 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.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -167,9 +169,8 @@ public class NodesShutdownMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.field(NODES_FIELD.getPreferredName(), nodes);
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.xContentValuesMap(NODES_FIELD.getPreferredName(), nodes);
     }
 
     /**

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

@@ -27,6 +27,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.function.UnaryOperator;
 
@@ -250,15 +251,11 @@ public class RepositoriesMetadata extends AbstractNamedDiffable<Custom> implemen
         return new RepositoriesMetadata(repository);
     }
 
-    /**
-     * {@inheritDoc}
-     */
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
-        for (RepositoryMetadata repository : repositories) {
-            toXContent(repository, builder, params);
-        }
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return repositories.stream()
+            .map(repository -> (ToXContent) (builder, params) -> toXContent(repository, builder, params))
+            .iterator();
     }
 
     @Override
@@ -273,7 +270,8 @@ public class RepositoriesMetadata extends AbstractNamedDiffable<Custom> implemen
      * @param builder    XContent builder
      * @param params     serialization parameters
      */
-    public static void toXContent(RepositoryMetadata repository, XContentBuilder builder, ToXContent.Params params) throws IOException {
+    public static XContentBuilder toXContent(RepositoryMetadata repository, XContentBuilder builder, ToXContent.Params params)
+        throws IOException {
         builder.startObject(repository.name());
         builder.field("type", repository.type());
         if (repository.uuid().equals(RepositoryData.MISSING_UUID) == false) {
@@ -288,6 +286,7 @@ public class RepositoriesMetadata extends AbstractNamedDiffable<Custom> implemen
             builder.field("pending_generation", repository.pendingGeneration());
         }
         builder.endObject();
+        return builder;
     }
 
     @Override

server/src/main/java/org/elasticsearch/common/xcontent/ChunkedToXContentHelper.java

@@ -0,0 +1,72 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.common.xcontent;
+
+import org.elasticsearch.common.collect.Iterators;
+import org.elasticsearch.xcontent.ToXContent;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.function.Function;
+
+public enum ChunkedToXContentHelper {
+    ;
+
+    public static Iterator<ToXContent> startObject() {
+        return Iterators.single(((builder, params) -> builder.startObject()));
+    }
+
+    public static Iterator<ToXContent> startObject(String name) {
+        return Iterators.single(((builder, params) -> builder.startObject(name)));
+    }
+
+    public static Iterator<ToXContent> endObject() {
+        return Iterators.single(((builder, params) -> builder.endObject()));
+    }
+
+    public static Iterator<ToXContent> startArray(String name) {
+        return Iterators.single(((builder, params) -> builder.startArray(name)));
+    }
+
+    public static Iterator<ToXContent> endArray() {
+        return Iterators.single(((builder, params) -> builder.endArray()));
+    }
+
+    public static Iterator<ToXContent> map(String name, Map<String, ?> map) {
+        return map(name, map, entry -> (ToXContent) (builder, params) -> builder.field(entry.getKey(), entry.getValue()));
+    }
+
+    public static Iterator<ToXContent> xContentFragmentValuesMap(String name, Map<String, ? extends ToXContent> map) {
+        return map(
+            name,
+            map,
+            entry -> (ToXContent) (builder, params) -> entry.getValue().toXContent(builder.startObject(entry.getKey()), params).endObject()
+        );
+    }
+
+    public static Iterator<ToXContent> xContentValuesMap(String name, Map<String, ? extends ToXContent> map) {
+        return map(
+            name,
+            map,
+            entry -> (ToXContent) (builder, params) -> entry.getValue().toXContent(builder.field(entry.getKey()), params)
+        );
+    }
+
+    public static Iterator<ToXContent> field(String name, boolean value) {
+        return Iterators.single(((builder, params) -> builder.field(name, value)));
+    }
+
+    public static Iterator<ToXContent> array(String name, Iterable<? extends ToXContent> iterable) {
+        return Iterators.concat(ChunkedToXContentHelper.startArray(name), iterable.iterator(), ChunkedToXContentHelper.endArray());
+    }
+
+    private static <T> Iterator<ToXContent> map(String name, Map<String, T> map, Function<Map.Entry<String, T>, ToXContent> toXContent) {
+        return Iterators.concat(startObject(name), map.entrySet().stream().map(toXContent).iterator(), endObject());
+    }
+}

server/src/main/java/org/elasticsearch/ingest/IngestMetadata.java

@@ -16,9 +16,10 @@ import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.util.Maps;
+import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
@@ -26,6 +27,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
@@ -95,13 +97,8 @@ public final class IngestMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.startArray(PIPELINES_FIELD.getPreferredName());
-        for (PipelineConfiguration pipeline : pipelines.values()) {
-            pipeline.toXContent(builder, params);
-        }
-        builder.endArray();
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.array(PIPELINES_FIELD.getPreferredName(), pipelines.values());
     }
 
     @Override

server/src/main/java/org/elasticsearch/persistent/PersistentTasksCustomMetadata.java

@@ -15,10 +15,12 @@ import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.VersionedNamedWriteable;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ObjectParser;
@@ -34,6 +36,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -552,16 +555,11 @@ public final class PersistentTasksCustomMetadata extends AbstractNamedDiffable<M
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.field("last_allocation_id", lastAllocationId);
-        builder.startArray("tasks");
-        {
-            for (PersistentTask<?> entry : tasks.values()) {
-                entry.toXContent(builder, params);
-            }
-        }
-        builder.endArray();
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return Iterators.concat(
+            Iterators.single((builder, params) -> builder.field("last_allocation_id", lastAllocationId)),
+            ChunkedToXContentHelper.array("tasks", tasks.values())
+        );
     }
 
     public static Builder builder() {

server/src/main/java/org/elasticsearch/script/ScriptMetadata.java

@@ -20,8 +20,7 @@ import org.elasticsearch.common.ParsingException;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.xcontent.ToXContentFragment;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParser.Token;
 
@@ -29,13 +28,14 @@ import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 
 /**
  * {@link ScriptMetadata} is used to store user-defined scripts
  * as part of the {@link ClusterState} using only an id as the key.
  */
-public final class ScriptMetadata implements Metadata.Custom, Writeable, ToXContentFragment {
+public final class ScriptMetadata implements Metadata.Custom, Writeable {
 
     /**
      * Standard logger used to warn about dropped scripts.
@@ -260,25 +260,12 @@ public final class ScriptMetadata implements Metadata.Custom, Writeable, ToXCont
         out.writeMap(scripts, StreamOutput::writeString, (o, v) -> v.writeTo(o));
     }
 
-    /**
-     * This will write XContent from {@link ScriptMetadata}.  The following format will be written:
-     *
-     * {@code
-     * {
-     *     "<id>" : "<{@link StoredScriptSource#toXContent(XContentBuilder, Params)}>",
-     *     "<id>" : "<{@link StoredScriptSource#toXContent(XContentBuilder, Params)}>",
-     *     ...
-     * }
-     * }
-     */
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        for (Map.Entry<String, StoredScriptSource> entry : scripts.entrySet()) {
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return scripts.entrySet().stream().map(entry -> (ToXContent) (builder, params) -> {
             builder.field(entry.getKey());
-            entry.getValue().toXContent(builder, params);
-        }
-
-        return builder;
+            return entry.getValue().toXContent(builder, params);
+        }).iterator();
     }
 
     @Override

server/src/main/java/org/elasticsearch/upgrades/FeatureMigrationResults.java

@@ -16,16 +16,18 @@ import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -78,9 +80,8 @@ public class FeatureMigrationResults implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.field(RESULTS_FIELD.getPreferredName(), featureStatuses);
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.xContentValuesMap(RESULTS_FIELD.getPreferredName(), featureStatuses);
     }
 
     public static FeatureMigrationResults fromXContent(XContentParser parser) {

server/src/test/java/org/elasticsearch/action/admin/cluster/snapshots/get/GetSnapshotsResponseTests.java

@@ -165,9 +165,8 @@ public class GetSnapshotsResponseTests extends ESTestCase {
             .asMatchPredicate()
             .or(Pattern.compile("snapshots\\.\\d+\\.index_details").asMatchPredicate())
             .or(Pattern.compile("failures\\.*").asMatchPredicate());
-        chunkedXContentTester(this::createParser, (XContentType t) -> createTestInstance(), params, this::doParseInstance).numberOfTestRuns(
-            1
-        )
+        chunkedXContentTester(this::createParser, (XContentType t) -> createTestInstance(), params, this::doParseInstance, false)
+            .numberOfTestRuns(1)
             .supportsUnknownFields(true)
             .shuffleFieldsExceptions(Strings.EMPTY_ARRAY)
             .randomFieldsExcludeFilter(predicate)

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

@@ -10,8 +10,9 @@ package org.elasticsearch.cluster.metadata;
 
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
-import org.elasticsearch.test.AbstractNamedWriteableTestCase;
-import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractChunkedSerializingTestCase;
+import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -20,15 +21,7 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
-import static org.elasticsearch.test.AbstractXContentTestCase.xContentTester;
-
-public class DataStreamMetadataTests extends AbstractNamedWriteableTestCase<DataStreamMetadata> {
-
-    public void testFromXContent() throws IOException {
-        xContentTester(this::createParser, this::createTestInstance, ToXContent.EMPTY_PARAMS, DataStreamMetadata::fromXContent)
-            .assertEqualsConsumer(this::assertEqualInstances)
-            .test();
-    }
+public class DataStreamMetadataTests extends AbstractChunkedSerializingTestCase<DataStreamMetadata> {
 
     @Override
     protected DataStreamMetadata createTestInstance() {
@@ -61,7 +54,17 @@ public class DataStreamMetadataTests extends AbstractNamedWriteableTestCase<Data
     }
 
     @Override
-    protected Class<DataStreamMetadata> categoryClass() {
-        return DataStreamMetadata.class;
+    protected DataStreamMetadata doParseInstance(XContentParser parser) throws IOException {
+        return DataStreamMetadata.fromXContent(parser);
+    }
+
+    @Override
+    protected Writeable.Reader<DataStreamMetadata> instanceReader() {
+        return DataStreamMetadata::new;
+    }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
     }
 }

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

@@ -11,7 +11,7 @@ package org.elasticsearch.cluster.metadata;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.test.SimpleDiffableSerializationTestCase;
+import org.elasticsearch.test.ChunkedToXContentDiffableSerializationTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
@@ -20,7 +20,7 @@ import java.util.Collections;
 import static org.elasticsearch.cluster.metadata.DesiredNodesSerializationTests.mutateDesiredNodes;
 import static org.elasticsearch.cluster.metadata.DesiredNodesTestCase.randomDesiredNodes;
 
-public class DesiredNodesMetadataSerializationTests extends SimpleDiffableSerializationTestCase<Metadata.Custom> {
+public class DesiredNodesMetadataSerializationTests extends ChunkedToXContentDiffableSerializationTestCase<Metadata.Custom> {
     @Override
     protected Metadata.Custom makeTestChanges(Metadata.Custom testInstance) {
         if (randomBoolean()) {
@@ -65,4 +65,9 @@ public class DesiredNodesMetadataSerializationTests extends SimpleDiffableSerial
     private DesiredNodesMetadata mutate(DesiredNodesMetadata base) {
         return new DesiredNodesMetadata(mutateDesiredNodes(base.getLatestDesiredNodes()));
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

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

@@ -11,8 +11,10 @@ package org.elasticsearch.cluster.metadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.common.xcontent.XContentElasticsearchExtension;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.test.ESTestCase;
@@ -57,12 +59,26 @@ public class IndexGraveyardTests extends ESTestCase {
         final IndexGraveyard graveyard = createRandom();
         final XContentBuilder builder = JsonXContent.contentBuilder();
         builder.startObject();
-        graveyard.toXContent(builder, ToXContent.EMPTY_PARAMS);
+        final var iterator = graveyard.toXContentChunked(ToXContent.EMPTY_PARAMS);
+        int chunks = 0;
+        while (iterator.hasNext()) {
+            ++chunks;
+            iterator.next().toXContent(builder, ToXContent.EMPTY_PARAMS);
+        }
+        assertEquals(2 + graveyard.getTombstones().size(), chunks);
         builder.endObject();
         if (graveyard.getTombstones().size() > 0) {
             // check that date properly printed
             assertThat(
-                Strings.toString(graveyard, false, true),
+                Strings.toString(
+                    ignored -> Iterators.concat(
+                        ChunkedToXContentHelper.startObject(),
+                        graveyard.toXContentChunked(ToXContent.EMPTY_PARAMS),
+                        ChunkedToXContentHelper.endObject()
+                    ),
+                    false,
+                    true
+                ),
                 containsString(
                     XContentElasticsearchExtension.DEFAULT_FORMATTER.format(
                         Instant.ofEpochMilli(graveyard.getTombstones().get(0).getDeleteDateInMillis())

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

@@ -44,9 +44,11 @@ import org.elasticsearch.xcontent.json.JsonXContent;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -2348,9 +2350,10 @@ public class MetadataTests extends ESTestCase {
     }
 
     private static class TestCustomMetadata implements Metadata.Custom {
+
         @Override
-        public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
-            return null;
+        public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params) {
+            return Collections.emptyIterator();
         }
 
         @Override

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

@@ -17,7 +17,7 @@ import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.test.SimpleDiffableSerializationTestCase;
+import org.elasticsearch.test.ChunkedToXContentDiffableSerializationTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
@@ -35,7 +35,7 @@ import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
 
-public class NodesShutdownMetadataTests extends SimpleDiffableSerializationTestCase<Metadata.Custom> {
+public class NodesShutdownMetadataTests extends ChunkedToXContentDiffableSerializationTestCase<Metadata.Custom> {
 
     public void testInsertNewNodeShutdownMetadata() {
         NodesShutdownMetadata nodesShutdownMetadata = new NodesShutdownMetadata(new HashMap<>());
@@ -142,4 +142,9 @@ public class NodesShutdownMetadataTests extends SimpleDiffableSerializationTestC
     protected Metadata.Custom mutateInstance(Metadata.Custom instance) throws IOException {
         return makeTestChanges(instance);
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

server/src/test/java/org/elasticsearch/ingest/IngestMetadataTests.java

@@ -12,7 +12,9 @@ import org.elasticsearch.cluster.DiffableUtils;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.util.Maps;
+import org.elasticsearch.common.xcontent.ChunkedToXContent;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -23,6 +25,7 @@ import org.elasticsearch.xcontent.XContentType;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 
 import static org.hamcrest.Matchers.aMapWithSize;
 import static org.hamcrest.Matchers.allOf;
@@ -46,7 +49,7 @@ public class IngestMetadataTests extends ESTestCase {
         XContentBuilder builder = XContentFactory.contentBuilder(randomFrom(XContentType.values()));
         builder.prettyPrint();
         builder.startObject();
-        ingestMetadata.toXContent(builder, ToXContent.EMPTY_PARAMS);
+        ChunkedToXContent.wrapAsXContentObject(ingestMetadata).toXContent(builder, ToXContent.EMPTY_PARAMS);
         builder.endObject();
         XContentBuilder shuffled = shuffleXContent(builder);
         try (XContentParser parser = createParser(shuffled)) {
@@ -121,4 +124,25 @@ public class IngestMetadataTests extends ESTestCase {
             equalTo(new PipelineConfiguration("2", new BytesArray("{\"key\" : \"value\"}"), XContentType.JSON))
         );
     }
+
+    public void testChunkedToXContent() throws IOException {
+        final BytesReference pipelineConfig = new BytesArray("{}");
+        final int pipelines = randomInt(10);
+        final Map<String, PipelineConfiguration> pipelineConfigurations = new HashMap<>();
+        for (int i = 0; i < pipelines; i++) {
+            final String id = Integer.toString(i);
+            pipelineConfigurations.put(id, new PipelineConfiguration(id, pipelineConfig, XContentType.JSON));
+        }
+        int chunksSeen = 0;
+        try (XContentBuilder builder = new XContentBuilder(randomFrom(XContentType.values()), Streams.NULL_OUTPUT_STREAM, Set.of())) {
+            final var iterator = new IngestMetadata(pipelineConfigurations).toXContentChunked(ToXContent.EMPTY_PARAMS);
+            builder.startObject();
+            while (iterator.hasNext()) {
+                iterator.next().toXContent(builder, ToXContent.EMPTY_PARAMS);
+                chunksSeen++;
+            }
+            builder.endObject();
+        }
+        assertEquals(2 + pipelines, chunksSeen);
+    }
 }

server/src/test/java/org/elasticsearch/persistent/PersistentTasksCustomMetadataTests.java

@@ -32,7 +32,7 @@ import org.elasticsearch.persistent.PersistentTasksCustomMetadata.PersistentTask
 import org.elasticsearch.persistent.TestPersistentTasksPlugin.State;
 import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestParams;
 import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestPersistentTasksExecutor;
-import org.elasticsearch.test.SimpleDiffableSerializationTestCase;
+import org.elasticsearch.test.ChunkedToXContentDiffableSerializationTestCase;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContent;
@@ -58,7 +58,7 @@ import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.sameInstance;
 
-public class PersistentTasksCustomMetadataTests extends SimpleDiffableSerializationTestCase<Custom> {
+public class PersistentTasksCustomMetadataTests extends ChunkedToXContentDiffableSerializationTestCase<Custom> {
 
     @Override
     protected PersistentTasksCustomMetadata createTestInstance() {
@@ -174,7 +174,7 @@ public class PersistentTasksCustomMetadataTests extends SimpleDiffableSerializat
         );
 
         XContentType xContentType = randomFrom(XContentType.values());
-        BytesReference shuffled = toShuffledXContent(testInstance, xContentType, params, false);
+        BytesReference shuffled = toShuffledXContent(asXContent(testInstance), xContentType, params, false);
 
         PersistentTasksCustomMetadata newInstance;
         try (XContentParser parser = createParser(XContentFactory.xContent(xContentType), shuffled)) {
@@ -396,4 +396,9 @@ public class PersistentTasksCustomMetadataTests extends SimpleDiffableSerializat
         }
         return new Assignment(randomAlphaOfLength(10), randomAlphaOfLength(10));
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

server/src/test/java/org/elasticsearch/script/ScriptMetadataTests.java

@@ -12,7 +12,7 @@ import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.Maps;
-import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.test.AbstractChunkedSerializingTestCase;
 import org.elasticsearch.xcontent.DeprecationHandler;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -28,7 +28,7 @@ import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.hasKey;
 
-public class ScriptMetadataTests extends AbstractXContentSerializingTestCase<ScriptMetadata> {
+public class ScriptMetadataTests extends AbstractChunkedSerializingTestCase<ScriptMetadata> {
 
     public void testGetScript() throws Exception {
         ScriptMetadata.Builder builder = new ScriptMetadata.Builder(null);
@@ -170,4 +170,9 @@ public class ScriptMetadataTests extends AbstractXContentSerializingTestCase<Scr
             throw new UncheckedIOException(ioe);
         }
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

server/src/test/java/org/elasticsearch/snapshots/RepositoriesMetadataSerializationTests.java

@@ -16,7 +16,7 @@ import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.test.SimpleDiffableSerializationTestCase;
+import org.elasticsearch.test.ChunkedToXContentDiffableSerializationTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
@@ -24,7 +24,7 @@ import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.List;
 
-public class RepositoriesMetadataSerializationTests extends SimpleDiffableSerializationTestCase<Custom> {
+public class RepositoriesMetadataSerializationTests extends ChunkedToXContentDiffableSerializationTestCase<Custom> {
 
     @Override
     protected Custom createTestInstance() {
@@ -117,4 +117,8 @@ public class RepositoriesMetadataSerializationTests extends SimpleDiffableSerial
         return new RepositoriesMetadata(repos);
     }
 
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

server/src/test/java/org/elasticsearch/upgrades/FeatureMigrationResultsTests.java

@@ -12,12 +12,12 @@ import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.core.Tuple;
-import org.elasticsearch.test.SimpleDiffableSerializationTestCase;
+import org.elasticsearch.test.ChunkedToXContentDiffableSerializationTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
 
-public class FeatureMigrationResultsTests extends SimpleDiffableSerializationTestCase<Metadata.Custom> {
+public class FeatureMigrationResultsTests extends ChunkedToXContentDiffableSerializationTestCase<Metadata.Custom> {
 
     @Override
     protected FeatureMigrationResults createTestInstance() {
@@ -72,4 +72,9 @@ public class FeatureMigrationResultsTests extends SimpleDiffableSerializationTes
     protected Writeable.Reader<Diff<Metadata.Custom>> diffReader() {
         return FeatureMigrationResults.ResultsDiff::new;
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

test/framework/src/main/java/org/elasticsearch/test/AbstractChunkedSerializingTestCase.java

@@ -11,6 +11,7 @@ package org.elasticsearch.test;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.ChunkedToXContent;
 import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentType;
 
@@ -24,14 +25,22 @@ public abstract class AbstractChunkedSerializingTestCase<T extends ChunkedToXCon
     protected AbstractXContentTestCase.XContentTester<T> createXContentTester() {
         return chunkedXContentTester(
             this::createParser,
-            xContentType -> createTestInstance(),
+            this::createXContextTestInstance,
             getToXContentParams(),
-            this::doParseInstance
+            this::doParseInstance,
+            isFragment()
         );
     }
 
     @Override
     protected ToXContent asXContent(T instance) {
+        if (isFragment()) {
+            return (ToXContentObject) ((builder, params) -> {
+                builder.startObject();
+                ChunkedToXContent.wrapAsXContentObject(instance).toXContent(builder, params);
+                return builder.endObject();
+            });
+        }
         return ChunkedToXContent.wrapAsXContentObject(instance);
     }
 
@@ -44,4 +53,8 @@ public abstract class AbstractChunkedSerializingTestCase<T extends ChunkedToXCon
      * Parses to a new instance using the provided {@link XContentParser}
      */
     protected abstract T doParseInstance(XContentParser parser) throws IOException;
+
+    protected boolean isFragment() {
+        return false;
+    }
 }

test/framework/src/main/java/org/elasticsearch/test/AbstractXContentTestCase.java

@@ -87,14 +87,21 @@ public abstract class AbstractXContentTestCase<T extends ToXContent> extends EST
         CheckedBiFunction<XContent, BytesReference, XContentParser, IOException> createParser,
         Function<XContentType, T> instanceSupplier,
         ToXContent.Params toXContentParams,
-        CheckedFunction<XContentParser, T, IOException> fromXContent
+        CheckedFunction<XContentParser, T, IOException> fromXContent,
+        boolean fragment
     ) {
         return new XContentTester<>(createParser, instanceSupplier, (testInstance, xContentType) -> {
             try (XContentBuilder builder = XContentBuilder.builder(xContentType.xContent())) {
                 var serialization = testInstance.toXContentChunked(toXContentParams);
+                if (fragment) {
+                    builder.startObject();
+                }
                 while (serialization.hasNext()) {
                     serialization.next().toXContent(builder, toXContentParams);
                 }
+                if (fragment) {
+                    builder.endObject();
+                }
                 return BytesReference.bytes(builder);
             }
         }, fromXContent);

test/framework/src/main/java/org/elasticsearch/test/ChunkedToXContentDiffableSerializationTestCase.java

@@ -0,0 +1,43 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test;
+
+import org.elasticsearch.cluster.Diff;
+import org.elasticsearch.cluster.Diffable;
+import org.elasticsearch.common.io.stream.Writeable.Reader;
+import org.elasticsearch.common.xcontent.ChunkedToXContent;
+
+import java.io.IOException;
+
+/**
+ * An abstract test case to ensure correct behavior of Diffable.
+ *
+ * This class can be used as a based class for tests of {@link org.elasticsearch.cluster.ClusterState.Custom} classes and other classes
+ * that support, Writable serialization, chunked XContent-based serialization and are diffable.
+ */
+public abstract class ChunkedToXContentDiffableSerializationTestCase<T extends Diffable<T> & ChunkedToXContent> extends
+    AbstractChunkedSerializingTestCase<T> {
+
+    /**
+     *  Introduces random changes into the test object
+     */
+    protected abstract T makeTestChanges(T testInstance);
+
+    protected abstract Reader<Diff<T>> diffReader();
+
+    public final void testDiffableSerialization() throws IOException {
+        DiffableTestUtils.testDiffableSerialization(
+            this::createTestInstance,
+            this::makeTestChanges,
+            getNamedWriteableRegistry(),
+            instanceReader(),
+            diffReader()
+        );
+    }
+}

test/framework/src/main/java/org/elasticsearch/test/TestCustomMetadata.java

@@ -12,12 +12,14 @@ import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.cluster.AbstractNamedDiffable;
 import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
+import java.util.Iterator;
 import java.util.function.Function;
 
 public abstract class TestCustomMetadata extends AbstractNamedDiffable<Metadata.Custom> implements Metadata.Custom {
@@ -91,9 +93,8 @@ public abstract class TestCustomMetadata extends AbstractNamedDiffable<Metadata.
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.field("data", getData());
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return Iterators.single((builder, params) -> builder.field("data", getData()));
     }
 
     @Override

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/AutoscalingMetadata.java

@@ -15,15 +15,17 @@ import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.cluster.metadata.Metadata;
 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.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicyMetadata;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -115,9 +117,8 @@ public class AutoscalingMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.field(POLICIES_FIELD.getPreferredName(), policies);
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.xContentValuesMap(POLICIES_FIELD.getPreferredName(), policies);
     }
 
     @Override

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/AutoscalingMetadataDiffableSerializationTests.java

@@ -11,13 +11,12 @@ import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.test.SimpleDiffableSerializationTestCase;
+import org.elasticsearch.test.ChunkedToXContentDiffableSerializationTestCase;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicy;
 import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicyMetadata;
 
-import java.io.IOException;
 import java.util.SortedMap;
 import java.util.TreeMap;
 
@@ -25,7 +24,7 @@ import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.mutateAuto
 import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.randomAutoscalingMetadata;
 import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.randomAutoscalingPolicy;
 
-public class AutoscalingMetadataDiffableSerializationTests extends SimpleDiffableSerializationTestCase<Metadata.Custom> {
+public class AutoscalingMetadataDiffableSerializationTests extends ChunkedToXContentDiffableSerializationTestCase<Metadata.Custom> {
 
     @Override
     protected NamedWriteableRegistry getNamedWriteableRegistry() {
@@ -38,7 +37,7 @@ public class AutoscalingMetadataDiffableSerializationTests extends SimpleDiffabl
     }
 
     @Override
-    protected AutoscalingMetadata doParseInstance(final XContentParser parser) throws IOException {
+    protected AutoscalingMetadata doParseInstance(final XContentParser parser) {
         return AutoscalingMetadata.parse(parser);
     }
 
@@ -79,4 +78,8 @@ public class AutoscalingMetadataDiffableSerializationTests extends SimpleDiffabl
         return AutoscalingMetadata.AutoscalingMetadataDiff::new;
     }
 
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/AutoFollowMetadataTests.java

@@ -12,7 +12,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.test.AbstractChunkedSerializingTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata;
 
@@ -23,7 +23,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.function.Predicate;
 
-public class AutoFollowMetadataTests extends AbstractXContentSerializingTestCase<AutoFollowMetadata> {
+public class AutoFollowMetadataTests extends AbstractChunkedSerializingTestCase<AutoFollowMetadata> {
 
     @Override
     protected Predicate<String> getRandomFieldsExcludeFilter() {
@@ -80,4 +80,9 @@ public class AutoFollowMetadataTests extends AbstractXContentSerializingTestCase
     protected Writeable.Reader<AutoFollowMetadata> instanceReader() {
         return AutoFollowMetadata::new;
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/license/LicensesMetadata.java

@@ -10,14 +10,16 @@ import org.elasticsearch.Version;
 import org.elasticsearch.cluster.AbstractNamedDiffable;
 import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.Nullable;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.Objects;
 
 /**
@@ -138,18 +140,20 @@ public class LicensesMetadata extends AbstractNamedDiffable<Metadata.Custom> imp
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        if (license == LICENSE_TOMBSTONE) {
-            builder.nullField(Fields.LICENSE);
-        } else {
-            builder.startObject(Fields.LICENSE);
-            license.toInnerXContent(builder, params);
-            builder.endObject();
-        }
-        if (trialVersion != null) {
-            builder.field(Fields.TRIAL_LICENSE, trialVersion.toString());
-        }
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return Iterators.single(((builder, params) -> {
+            if (license == LICENSE_TOMBSTONE) {
+                builder.nullField(Fields.LICENSE);
+            } else {
+                builder.startObject(Fields.LICENSE);
+                license.toInnerXContent(builder, params);
+                builder.endObject();
+            }
+            if (trialVersion != null) {
+                builder.field(Fields.TRIAL_LICENSE, trialVersion.toString());
+            }
+            return builder;
+        }));
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ccr/AutoFollowMetadata.java

@@ -11,14 +11,17 @@ import org.elasticsearch.Version;
 import org.elasticsearch.cluster.AbstractNamedDiffable;
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.ToXContentFragment;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
@@ -28,6 +31,7 @@ import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -147,31 +151,12 @@ public class AutoFollowMetadata extends AbstractNamedDiffable<Metadata.Custom> i
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.startObject(PATTERNS_FIELD.getPreferredName());
-        for (Map.Entry<String, AutoFollowPattern> entry : patterns.entrySet()) {
-            builder.startObject(entry.getKey());
-            builder.value(entry.getValue());
-            builder.endObject();
-        }
-        builder.endObject();
-
-        builder.startObject(FOLLOWED_LEADER_INDICES_FIELD.getPreferredName());
-        for (Map.Entry<String, List<String>> entry : followedLeaderIndexUUIDs.entrySet()) {
-            builder.field(entry.getKey(), entry.getValue());
-        }
-        builder.endObject();
-        builder.startObject(HEADERS.getPreferredName());
-        for (Map.Entry<String, Map<String, String>> entry : headers.entrySet()) {
-            builder.field(entry.getKey(), entry.getValue());
-        }
-        builder.endObject();
-        return builder;
-    }
-
-    @Override
-    public boolean isFragment() {
-        return true;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return Iterators.concat(
+            ChunkedToXContentHelper.xContentFragmentValuesMap(PATTERNS_FIELD.getPreferredName(), patterns),
+            ChunkedToXContentHelper.map(FOLLOWED_LEADER_INDICES_FIELD.getPreferredName(), followedLeaderIndexUUIDs),
+            ChunkedToXContentHelper.map(HEADERS.getPreferredName(), headers)
+        );
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/IndexLifecycleMetadata.java

@@ -14,15 +14,18 @@ import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.Metadata.Custom;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.Iterators;
 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.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -100,10 +103,11 @@ public class IndexLifecycleMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.xContentValuesMap(POLICIES_FIELD.getPreferredName(), policyMetadatas);
-        builder.field(OPERATION_MODE_FIELD.getPreferredName(), operationMode);
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return Iterators.concat(
+            ChunkedToXContentHelper.xContentValuesMap(POLICIES_FIELD.getPreferredName(), policyMetadatas),
+            Iterators.single((builder, params) -> builder.field(OPERATION_MODE_FIELD.getPreferredName(), operationMode))
+        );
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlMetadata.java

@@ -14,19 +14,21 @@ import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 import java.util.SortedMap;
@@ -118,10 +120,11 @@ public class MlMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.field(UPGRADE_MODE.getPreferredName(), upgradeMode);
-        builder.field(RESET_MODE.getPreferredName(), resetMode);
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return Iterators.single(
+            ((builder, params) -> builder.field(UPGRADE_MODE.getPreferredName(), upgradeMode)
+                .field(RESET_MODE.getPreferredName(), resetMode))
+        );
     }
 
     public static class MlMetadataDiff implements NamedDiff<Metadata.Custom> {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/slm/SnapshotLifecycleMetadata.java

@@ -14,17 +14,20 @@ import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.Iterators;
 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.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xpack.core.ilm.OperationMode;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -132,11 +135,15 @@ public class SnapshotLifecycleMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.field(POLICIES_FIELD.getPreferredName(), this.snapshotConfigurations);
-        builder.field(OPERATION_MODE_FIELD.getPreferredName(), operationMode);
-        builder.field(STATS_FIELD.getPreferredName(), this.slmStats);
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return Iterators.concat(
+            ChunkedToXContentHelper.xContentValuesMap(POLICIES_FIELD.getPreferredName(), this.snapshotConfigurations),
+            Iterators.single((builder, params) -> {
+                builder.field(OPERATION_MODE_FIELD.getPreferredName(), operationMode);
+                builder.field(STATS_FIELD.getPreferredName(), this.slmStats);
+                return builder;
+            })
+        );
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/TransformMetadata.java

@@ -15,14 +15,15 @@ import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContent;
-import org.elasticsearch.xcontent.XContentBuilder;
 
 import java.io.IOException;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.Objects;
 
 public class TransformMetadata implements Metadata.Custom {
@@ -81,9 +82,8 @@ public class TransformMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
-        builder.field(RESET_MODE.getPreferredName(), resetMode);
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.field(RESET_MODE.getPreferredName(), resetMode);
     }
 
     public static class TransformMetadataDiff implements NamedDiff<Metadata.Custom> {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/WatcherMetadata.java

@@ -12,12 +12,14 @@ import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.xcontent.ParseField;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.Objects;
 
 public class WatcherMetadata extends AbstractNamedDiffable<Metadata.Custom> implements Metadata.Custom {
@@ -105,9 +107,8 @@ public class WatcherMetadata extends AbstractNamedDiffable<Metadata.Custom> impl
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.field(Field.MANUALLY_STOPPED.getPreferredName(), manuallyStopped);
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.field(Field.MANUALLY_STOPPED.getPreferredName(), manuallyStopped);
     }
 
     interface Field {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/transport/actions/stats/WatcherStatsResponse.java

@@ -13,6 +13,7 @@ import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ChunkedToXContent;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -28,7 +29,7 @@ import java.util.Locale;
 
 public class WatcherStatsResponse extends BaseNodesResponse<WatcherStatsResponse.Node> implements ToXContentObject {
 
-    private WatcherMetadata watcherMetadata;
+    private final WatcherMetadata watcherMetadata;
 
     public WatcherStatsResponse(StreamInput in) throws IOException {
         super(in);
@@ -63,7 +64,7 @@ public class WatcherStatsResponse extends BaseNodesResponse<WatcherStatsResponse
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        watcherMetadata.toXContent(builder, params);
+        ChunkedToXContent.wrapAsXContentObject(watcherMetadata).toXContent(builder, params);
         builder.startArray("stats");
         for (Node node : getNodes()) {
             node.toXContent(builder, params);

x-pack/plugin/core/src/test/java/org/elasticsearch/license/LicensesMetadataSerializationTests.java

@@ -12,6 +12,7 @@ import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.RepositoriesMetadata;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.ChunkedToXContent;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
@@ -38,7 +39,7 @@ public class LicensesMetadataSerializationTests extends ESTestCase {
         XContentBuilder builder = XContentFactory.jsonBuilder();
         builder.startObject();
         builder.startObject("licenses");
-        licensesMetadata.toXContent(builder, ToXContent.EMPTY_PARAMS);
+        ChunkedToXContent.wrapAsXContentObject(licensesMetadata).toXContent(builder, ToXContent.EMPTY_PARAMS);
         builder.endObject();
         builder.endObject();
         LicensesMetadata licensesMetadataFromXContent = getLicensesMetadataFromXContent(createParser(builder));
@@ -52,7 +53,7 @@ public class LicensesMetadataSerializationTests extends ESTestCase {
         XContentBuilder builder = XContentFactory.jsonBuilder();
         builder.startObject();
         builder.startObject("licenses");
-        licensesMetadata.toXContent(builder, ToXContent.EMPTY_PARAMS);
+        ChunkedToXContent.wrapAsXContentObject(licensesMetadata).toXContent(builder, ToXContent.EMPTY_PARAMS);
         builder.endObject();
         builder.endObject();
         LicensesMetadata licensesMetadataFromXContent = getLicensesMetadataFromXContent(createParser(builder));
@@ -101,7 +102,7 @@ public class LicensesMetadataSerializationTests extends ESTestCase {
         XContentBuilder builder = XContentFactory.jsonBuilder();
         builder.startObject();
         builder.startObject("licenses");
-        licensesMetadata.toXContent(builder, ToXContent.EMPTY_PARAMS);
+        ChunkedToXContent.wrapAsXContentObject(licensesMetadata).toXContent(builder, ToXContent.EMPTY_PARAMS);
         builder.endObject();
         builder.endObject();
         LicensesMetadata licensesMetadataFromXContent = getLicensesMetadataFromXContent(createParser(builder));

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/slm/SnapshotLifecycleMetadataTests.java

@@ -9,14 +9,14 @@ package org.elasticsearch.xpack.core.slm;
 
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.Maps;
-import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.test.AbstractChunkedSerializingTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.core.ilm.OperationMode;
 
 import java.io.IOException;
 import java.util.Map;
 
-public class SnapshotLifecycleMetadataTests extends AbstractXContentSerializingTestCase<SnapshotLifecycleMetadata> {
+public class SnapshotLifecycleMetadataTests extends AbstractChunkedSerializingTestCase<SnapshotLifecycleMetadata> {
     @Override
     protected SnapshotLifecycleMetadata doParseInstance(XContentParser parser) throws IOException {
         return SnapshotLifecycleMetadata.PARSER.apply(parser, null);
@@ -41,4 +41,9 @@ public class SnapshotLifecycleMetadataTests extends AbstractXContentSerializingT
     protected Writeable.Reader<SnapshotLifecycleMetadata> instanceReader() {
         return SnapshotLifecycleMetadata::new;
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichMetadata.java

@@ -12,9 +12,10 @@ import org.elasticsearch.cluster.AbstractNamedDiffable;
 import org.elasticsearch.cluster.metadata.Metadata;
 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.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.core.enrich.EnrichPolicy;
 
@@ -22,6 +23,7 @@ import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 
@@ -96,15 +98,8 @@ public final class EnrichMetadata extends AbstractNamedDiffable<Metadata.Custom>
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.startObject(POLICIES.getPreferredName());
-        for (Map.Entry<String, EnrichPolicy> entry : policies.entrySet()) {
-            builder.startObject(entry.getKey());
-            builder.value(entry.getValue());
-            builder.endObject();
-        }
-        builder.endObject();
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.xContentFragmentValuesMap(POLICIES.getPreferredName(), policies);
     }
 
     @Override

x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/EnrichMetadataTests.java

@@ -8,7 +8,7 @@ package org.elasticsearch.xpack.enrich;
 
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.Maps;
-import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.test.AbstractChunkedSerializingTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xpack.core.enrich.EnrichPolicy;
@@ -19,7 +19,7 @@ import java.util.Map;
 import static org.elasticsearch.xpack.enrich.EnrichPolicyTests.randomEnrichPolicy;
 import static org.hamcrest.Matchers.equalTo;
 
-public class EnrichMetadataTests extends AbstractXContentSerializingTestCase<EnrichMetadata> {
+public class EnrichMetadataTests extends AbstractChunkedSerializingTestCase<EnrichMetadata> {
 
     @Override
     protected EnrichMetadata doParseInstance(XContentParser parser) throws IOException {
@@ -61,4 +61,9 @@ public class EnrichMetadataTests extends AbstractXContentSerializingTestCase<Enr
             EnrichPolicyTests.assertEqualPolicies(expected, actual);
         }
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/ilm/IndexLifecycleMetadataTests.java

@@ -15,7 +15,7 @@ import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable.Reader;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.test.SimpleDiffableSerializationTestCase;
+import org.elasticsearch.test.ChunkedToXContentDiffableSerializationTestCase;
 import org.elasticsearch.test.VersionUtils;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.ParseField;
@@ -55,7 +55,7 @@ import java.util.TreeMap;
 import static org.elasticsearch.xpack.ilm.LifecyclePolicyTestsUtils.newTestLifecyclePolicy;
 import static org.elasticsearch.xpack.ilm.LifecyclePolicyTestsUtils.randomTimeseriesLifecyclePolicy;
 
-public class IndexLifecycleMetadataTests extends SimpleDiffableSerializationTestCase<Custom> {
+public class IndexLifecycleMetadataTests extends ChunkedToXContentDiffableSerializationTestCase<Custom> {
 
     @Override
     protected IndexLifecycleMetadata createTestInstance() {
@@ -212,4 +212,9 @@ public class IndexLifecycleMetadataTests extends SimpleDiffableSerializationTest
         }
         return new IndexLifecycleMetadata(policies, mode);
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/inference/ModelAliasMetadata.java

@@ -16,8 +16,10 @@ import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.cluster.metadata.Metadata;
 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.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
@@ -26,6 +28,7 @@ import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 
@@ -88,13 +91,8 @@ public class ModelAliasMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.startObject(MODEL_ALIASES.getPreferredName());
-        for (Map.Entry<String, ModelAliasEntry> modelAliasEntry : modelAliases.entrySet()) {
-            builder.field(modelAliasEntry.getKey(), modelAliasEntry.getValue());
-        }
-        builder.endObject();
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return ChunkedToXContentHelper.xContentValuesMap(MODEL_ALIASES.getPreferredName(), modelAliases);
     }
 
     @Override

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/inference/assignment/TrainedModelAssignmentMetadata.java

@@ -19,7 +19,7 @@ import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.core.ml.inference.assignment.TrainedModelAssignment;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
@@ -27,6 +27,7 @@ import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -107,9 +108,11 @@ public class TrainedModelAssignmentMetadata implements Metadata.Custom {
     }
 
     @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.mapContents(modelRoutingEntries);
-        return builder;
+    public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
+        return modelRoutingEntries.entrySet()
+            .stream()
+            .map(entry -> (ToXContent) (builder, params) -> entry.getValue().toXContent(builder.field(entry.getKey()), params))
+            .iterator();
     }
 
     @Override

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlMetadataTests.java

@@ -10,7 +10,7 @@ import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.search.SearchModule;
-import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.test.AbstractChunkedSerializingTestCase;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.core.ml.MlMetadata;
@@ -19,7 +19,7 @@ import java.util.Collections;
 
 import static org.hamcrest.Matchers.equalTo;
 
-public class MlMetadataTests extends AbstractXContentSerializingTestCase<MlMetadata> {
+public class MlMetadataTests extends AbstractChunkedSerializingTestCase<MlMetadata> {
 
     @Override
     protected MlMetadata createTestInstance() {
@@ -71,4 +71,9 @@ public class MlMetadataTests extends AbstractXContentSerializingTestCase<MlMetad
 
         return metadataBuilder.build();
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/inference/assignment/TrainedModelAssignmentMetadataTests.java

@@ -9,7 +9,7 @@ package org.elasticsearch.xpack.ml.inference.assignment;
 
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.test.AbstractChunkedSerializingTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.core.ml.action.StartTrainedModelDeploymentAction;
 import org.elasticsearch.xpack.core.ml.inference.assignment.Priority;
@@ -25,7 +25,7 @@ import java.util.stream.Stream;
 
 import static org.hamcrest.Matchers.is;
 
-public class TrainedModelAssignmentMetadataTests extends AbstractXContentSerializingTestCase<TrainedModelAssignmentMetadata> {
+public class TrainedModelAssignmentMetadataTests extends AbstractChunkedSerializingTestCase<TrainedModelAssignmentMetadata> {
 
     public static TrainedModelAssignmentMetadata randomInstance() {
         LinkedHashMap<String, TrainedModelAssignment> map = Stream.generate(() -> randomAlphaOfLength(10))
@@ -72,4 +72,8 @@ public class TrainedModelAssignmentMetadataTests extends AbstractXContentSeriali
         );
     }
 
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/TransformMetadataTests.java

@@ -8,11 +8,11 @@
 package org.elasticsearch.xpack.transform;
 
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.test.AbstractChunkedSerializingTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.core.transform.TransformMetadata;
 
-public class TransformMetadataTests extends AbstractXContentSerializingTestCase<TransformMetadata> {
+public class TransformMetadataTests extends AbstractChunkedSerializingTestCase<TransformMetadata> {
 
     @Override
     protected TransformMetadata createTestInstance() {
@@ -33,4 +33,9 @@ public class TransformMetadataTests extends AbstractXContentSerializingTestCase<
     protected TransformMetadata mutateInstance(TransformMetadata instance) {
         return new TransformMetadata.Builder().isResetMode(instance.isResetMode() == false).build();
     }
+
+    @Override
+    protected boolean isFragment() {
+        return true;
+    }
 }

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/WatcherMetadataSerializationTests.java

@@ -11,6 +11,7 @@ import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.RepositoriesMetadata;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.ChunkedToXContent;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.ToXContent;
@@ -35,7 +36,7 @@ public class WatcherMetadataSerializationTests extends ESTestCase {
         XContentBuilder builder = XContentFactory.jsonBuilder();
         builder.startObject();
         builder.startObject("watcher");
-        watcherMetadata.toXContent(builder, ToXContent.EMPTY_PARAMS);
+        ChunkedToXContent.wrapAsXContentObject(watcherMetadata).toXContent(builder, ToXContent.EMPTY_PARAMS);
         builder.endObject();
         builder.endObject();
         WatcherMetadata watchersMetadataFromXContent = getWatcherMetadataFromXContent(createParser(builder));