Fix error in FieldCapabilitiesResponse serialization (#74509)

In version 7.13.0 we added failure serialization to the `_field_caps` API.
FieldCapabilitiesResponse currently has a bug here where the failures list is
serialized based on Version.CURRENT, which is different on different releases,
but instead it should be serialized/deserialized for all nodes where the version
is on or above 7.13.0.

Backport of #74504

server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesResponse.java

@@ -77,7 +77,7 @@ public FieldCapabilitiesResponse(StreamInput in) throws IOException {
         }
         this.responseMap = in.readMap(StreamInput::readString, FieldCapabilitiesResponse::readField);
         indexResponses = in.readList(FieldCapabilitiesIndexResponse::new);
-        if (in.getVersion().onOrAfter(Version.CURRENT)) {
+        if (in.getVersion().onOrAfter(Version.V_7_13_0)) {
             this.failures = in.readList(FieldCapabilitiesFailure::new);
         } else {
             this.failures = Collections.emptyList();
@@ -149,7 +149,7 @@ public void writeTo(StreamOutput out) throws IOException {
         }
         out.writeMap(responseMap, StreamOutput::writeString, FieldCapabilitiesResponse::writeField);
         out.writeList(indexResponses);
-        if (out.getVersion().onOrAfter(Version.CURRENT)) {
+        if (out.getVersion().onOrAfter(Version.V_7_13_0)) {
             out.writeList(failures);
         }
     }

server/src/test/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesResponseTests.java

@@ -9,12 +9,17 @@
 package org.elasticsearch.action.fieldcaps;
 
 import org.elasticsearch.ElasticsearchExceptionTests;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.io.stream.NamedWriteableAwareStreamInput;
+import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
+import org.elasticsearch.test.VersionUtils;
 import org.hamcrest.Matchers;
 
 import java.io.IOException;
@@ -120,6 +125,75 @@ public void testFailureSerialization() throws IOException {
         }
     }
 
+    /**
+     * check that failure serialization between different minor versions after 7.13 works
+     */
+    public void testMixedVersionFailureSerialization_post_7_13() throws IOException {
+        FieldCapabilitiesResponse original = createResponseWithFailures();
+        FieldCapabilitiesResponse deserialized;
+        Version outVersion = VersionUtils.randomVersionBetween(random(), Version.V_7_13_0, Version.CURRENT);
+        Version inVersion = VersionUtils.randomVersionBetween(random(), Version.V_7_13_0, Version.CURRENT);
+        try (BytesStreamOutput output = new BytesStreamOutput()) {
+            output.setVersion(outVersion);
+            original.writeTo(output);
+            try (StreamInput in = new NamedWriteableAwareStreamInput(output.bytes().streamInput(), getNamedWriteableRegistry())) {
+
+                in.setVersion(inVersion);
+                deserialized = new FieldCapabilitiesResponse(in);
+                assertEquals(-1, in.read());
+            }
+        }
+        assertThat(deserialized.getIndices(), Matchers.equalTo(original.getIndices()));
+
+        // only match size of failure list and indices, most exceptions don't support 'equals'
+        List<FieldCapabilitiesFailure> deserializedFailures = deserialized.getFailures();
+        assertEquals(deserializedFailures.size(), original.getFailures().size());
+        int i = 0;
+        for (FieldCapabilitiesFailure originalFailure : original.getFailures()) {
+            FieldCapabilitiesFailure deserializedFaliure = deserializedFailures.get(i);
+            assertThat(deserializedFaliure.getIndices(), Matchers.equalTo(originalFailure.getIndices()));
+            i++;
+        }
+    }
+
+    /**
+     * check that failure serialization to minor versions before 7.13 works without serializing the failures part
+     */
+    public void testSerialization_pre_7_13() throws IOException {
+        FieldCapabilitiesResponse original = createResponseWithFailures();
+        FieldCapabilitiesResponse deserialized;
+        Version version = VersionUtils.randomVersionBetween(random(), Version.V_7_6_0, VersionUtils.getPreviousVersion(Version.V_7_13_0));
+        try (BytesStreamOutput output = new BytesStreamOutput()) {
+            output.setVersion(version);
+            original.writeTo(output);
+            try (StreamInput in = new NamedWriteableAwareStreamInput(output.bytes().streamInput(), getNamedWriteableRegistry())) {
+                in.setVersion(version);
+                deserialized = new FieldCapabilitiesResponse(in);
+                assertEquals(-1, in.read());
+            }
+        }
+        assertThat(deserialized.getIndices(), Matchers.equalTo(original.getIndices()));
+
+        // only match size of failure list and indices, most exceptions don't support 'equals'
+        assertEquals(0, deserialized.getFailures().size());
+    }
+
+    public void testReadFromPre7_13() throws IOException {
+        String msg = "AQpzb21lLWluZGV4AgdmaWVsZC0xAQdrZXl3b3JkB2ZpZWxkLTEHa2V5d29yZAEBAQABAAEAAAdmaWVsZC0y"
+            + "AgdrZXl3b3JkB2ZpZWxkLTIHa2V5d29yZAEBAQABAAEAAARsb25nB2ZpZWxkLTIEbG9uZwEBAQABAAEAAAAAAAA=";
+        try (StreamInput in = StreamInput.wrap(java.util.Base64.getDecoder().decode(msg))) {
+            // minimum version set to 7.6 because the nested FieldCapabilities had another serialization change there
+            in.setVersion(VersionUtils.randomVersionBetween(random(), Version.V_7_6_0, VersionUtils.getPreviousVersion(Version.V_7_13_0)));
+            FieldCapabilitiesResponse deserialized = new FieldCapabilitiesResponse(in);
+            assertArrayEquals(new String[]{"some-index"}, deserialized.getIndices());
+            assertEquals(2, deserialized.get().size());
+            assertNotNull(deserialized.get().get("field-1").get("keyword"));
+            assertNotNull(deserialized.get().get("field-2").get("keyword"));
+            assertEquals(0, deserialized.getIndexResponses().size());
+            assertEquals(0, deserialized.getFailures().size());
+        }
+    }
+
     public void testFailureParsing() throws IOException {
         FieldCapabilitiesResponse randomResponse = createResponseWithFailures();
         boolean humanReadable = randomBoolean();