[7.x] Ensure system index upgrade mechanism can handle integer _meta.verion (#75383)

server/src/main/java/org/elasticsearch/indices/SystemIndexManager.java

@@ -269,7 +269,13 @@ private Version readMappingVersion(SystemIndexDescriptor descriptor, MappingMeta
                 throw new IllegalStateException("Cannot read version string in index " + indexName);
             }
 
-            final String versionString = (String) meta.get(descriptor.getVersionMetaKey());
+            final Object rawVersion = meta.get(descriptor.getVersionMetaKey());
+            if (rawVersion instanceof Integer) {
+                // This can happen with old system indices, such as .tasks, which were created before we used an Elasticsearch
+                // version here. We should just replace the template to be sure.
+                return Version.V_EMPTY;
+            }
+            final String versionString = rawVersion != null ? rawVersion.toString() : null;
             if (versionString == null) {
                 logger.warn("No value found in mappings for [_meta.{}]", descriptor.getVersionMetaKey());
                 // If we called `Version.fromString(null)`, it would return `Version.CURRENT` and we wouldn't update the mappings

server/src/test/java/org/elasticsearch/indices/SystemIndexManagerTests.java

@@ -252,6 +252,19 @@ public void testManagerSubmitsPutRequest() {
         verify(client, times(1)).execute(any(PutMappingAction.class), any(PutMappingRequest.class), any());
     }
 
+    /**
+     * Check that this
+     */
+    public void testCanHandleIntegerMetaVersion() {
+        SystemIndices systemIndices = new SystemIndices(Map.of("MyIndex", FEATURE));
+        SystemIndexManager manager = new SystemIndexManager(systemIndices, client);
+
+        final ClusterState.Builder clusterStateBuilder = createClusterState(Strings.toString(getMappings(3)));
+        markShardsAvailable(clusterStateBuilder);
+
+        assertThat(manager.getUpgradeStatus(clusterStateBuilder.build(), DESCRIPTOR), equalTo(UpgradeStatus.NEEDS_MAPPINGS_UPDATE));
+    }
+
     private static ClusterState.Builder createClusterState() {
         return createClusterState(SystemIndexManagerTests.DESCRIPTOR.getMappings());
     }
@@ -412,4 +425,32 @@ private static XContentBuilder getMappings(String version) {
             throw new UncheckedIOException("Failed to build " + SYSTEM_INDEX_NAME + " index mappings", e);
         }
     }
+
+    // Prior to 7.12.0, .tasks had _meta.version: 3 so we need to be sure we can handle that
+    private static XContentBuilder getMappings(int version) {
+        try {
+            final XContentBuilder builder = jsonBuilder();
+
+            builder.startObject();
+            {
+                builder.startObject("_meta");
+                builder.field("version", version);
+                builder.endObject();
+
+                builder.field("dynamic", "strict");
+                builder.startObject("properties");
+                {
+                    builder.startObject("completed");
+                    builder.field("type", "boolean");
+                    builder.endObject();
+                }
+                builder.endObject();
+            }
+
+            builder.endObject();
+            return builder;
+        } catch (IOException e) {
+            throw new UncheckedIOException("Failed to build " + SYSTEM_INDEX_NAME + " index mappings", e);
+        }
+    }
 }