Deduplicate IngestStats and IngestStats.Stats identity records when deserializing (elastic#122496) (elastic#122516)

This commit makes sure we reuse the existing static instance when deserializing to avoid excessive heap usage.
# Conflicts:
#	server/src/main/java/org/elasticsearch/ingest/IngestStats.java

Author: iverase

docs/changelog/122496.yaml

@@ -0,0 +1,5 @@
+pr: 122496
+summary: Deduplicate `IngestStats` and `IngestStats.Stats` identity records when deserializing
+area: Ingest Node
+type: bug
+issues: []

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

@@ -57,14 +57,17 @@ public record IngestStats(Stats totalStats, List<PipelineStat> pipelineStats, Ma
      * Read from a stream.
      */
     public static IngestStats read(StreamInput in) throws IOException {
-        var stats = new Stats(in);
+        var stats = readStats(in);
         var size = in.readVInt();
+        if (stats == Stats.IDENTITY && size == 0) {
+            return IDENTITY;
+        }
         var pipelineStats = new ArrayList<PipelineStat>(size);
         var processorStats = Maps.<String, List<ProcessorStat>>newMapWithExpectedSize(size);
 
         for (var i = 0; i < size; i++) {
             var pipelineId = in.readString();
-            var pipelineStat = new Stats(in);
+            var pipelineStat = readStats(in);
             var byteStat = in.getTransportVersion().onOrAfter(TransportVersions.NODE_STATS_INGEST_BYTES)
                 ? new ByteStats(in)
                 : new ByteStats(0, 0);
@@ -74,7 +77,7 @@ public static IngestStats read(StreamInput in) throws IOException {
             for (var j = 0; j < processorsSize; j++) {
                 var processorName = in.readString();
                 var processorType = in.readString();
-                var processorStat = new Stats(in);
+                var processorStat = readStats(in);
                 processorStatsPerPipeline.add(new ProcessorStat(processorName, processorType, processorStat));
             }
             processorStats.put(pipelineId, Collections.unmodifiableList(processorStatsPerPipeline));
@@ -171,20 +174,28 @@ static Map<String, List<ProcessorStat>> merge(Map<String, List<ProcessorStat>> f
         return totalsPerPipelineProcessor;
     }
 
+    /**
+     * Read {@link Stats} from a stream.
+     */
+    private static Stats readStats(StreamInput in) throws IOException {
+        long ingestCount = in.readVLong();
+        long ingestTimeInMillis = in.readVLong();
+        long ingestCurrent = in.readVLong();
+        long ingestFailedCount = in.readVLong();
+        if (ingestCount == 0 && ingestTimeInMillis == 0 && ingestCurrent == 0 && ingestFailedCount == 0) {
+            return Stats.IDENTITY;
+        } else {
+            return new Stats(ingestCount, ingestTimeInMillis, ingestCurrent, ingestFailedCount);
+        }
+    }
+
     public record Stats(long ingestCount, long ingestTimeInMillis, long ingestCurrent, long ingestFailedCount)
         implements
             Writeable,
             ToXContentFragment {
 
         public static final Stats IDENTITY = new Stats(0, 0, 0, 0);
 
-        /**
-         * Read from a stream.
-         */
-        public Stats(StreamInput in) throws IOException {
-            this(in.readVLong(), in.readVLong(), in.readVLong(), in.readVLong());
-        }
-
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeVLong(ingestCount);

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

@@ -19,6 +19,7 @@
 import java.util.Map;
 
 import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.sameInstance;
 
 public class IngestStatsTests extends ESTestCase {
 
@@ -31,6 +32,11 @@ public void testSerialization() throws IOException {
         assertIngestStats(ingestStats, serializedStats);
     }
 
+    public void testIdentitySerialization() throws IOException {
+        IngestStats serializedStats = serialize(IngestStats.IDENTITY);
+        assertThat(serializedStats, sameInstance(IngestStats.IDENTITY));
+    }
+
     public void testStatsMerge() {
         var first = randomStats();
         var second = randomStats();