diff --git a/server/src/main/java/org/elasticsearch/cluster/DiffableUtils.java b/server/src/main/java/org/elasticsearch/cluster/DiffableUtils.java
index 8749397323fdf..f5fb4e797aa58 100644
--- a/server/src/main/java/org/elasticsearch/cluster/DiffableUtils.java
+++ b/server/src/main/java/org/elasticsearch/cluster/DiffableUtils.java
@@ -21,13 +21,14 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
 
 public final class DiffableUtils {
     private DiffableUtils() {}
 
-    private static final MapDiff<?, ?, ?> EMPTY = new MapDiff<>(null, null, List.of(), List.of(), List.of()) {
+    private static final MapDiff<?, ?, ?> EMPTY = new MapDiff<>(null, null, List.of(), List.of(), List.of(), null) {
         @Override
-        public Object apply(Object part) {
+        public Map<Object, Object> apply(Map<Object, Object> part) {
             return part;
         }
     };
@@ -53,62 +54,31 @@ public static KeySerializer<Integer> getVIntKeySerializer() {
         return VIntKeySerializer.INSTANCE;
     }
 
-    /**
-     * Calculates diff between two ImmutableOpenMaps of Diffable objects
-     */
-    public static <K, T extends Diffable<T>> MapDiff<K, T, ImmutableOpenMap<K, T>> diff(
-        ImmutableOpenMap<K, T> before,
-        ImmutableOpenMap<K, T> after,
-        KeySerializer<K> keySerializer
-    ) {
-        assert after != null && before != null;
-        return before.equals(after)
-            ? emptyDiff()
-            : ImmutableOpenMapDiff.create(before, after, keySerializer, DiffableValueSerializer.getWriteOnlyInstance());
-    }
-
-    /**
-     * Calculates diff between two ImmutableOpenMaps of non-diffable objects
-     */
-    public static <K, T> MapDiff<K, T, ImmutableOpenMap<K, T>> diff(
-        ImmutableOpenMap<K, T> before,
-        ImmutableOpenMap<K, T> after,
-        KeySerializer<K> keySerializer,
-        ValueSerializer<K, T> valueSerializer
-    ) {
-        assert after != null && before != null;
-        return before.equals(after) ? emptyDiff() : ImmutableOpenMapDiff.create(before, after, keySerializer, valueSerializer);
-    }
-
     /**
      * Calculates diff between two Maps of Diffable objects.
      */
-    public static <K, T extends Diffable<T>> MapDiff<K, T, Map<K, T>> diff(
-        Map<K, T> before,
-        Map<K, T> after,
-        KeySerializer<K> keySerializer
-    ) {
+    public static <K, T extends Diffable<T>, M extends Map<K, T>> MapDiff<K, T, M> diff(M before, M after, KeySerializer<K> keySerializer) {
         assert after != null && before != null;
         return before.equals(after)
             ? emptyDiff()
-            : JdkMapDiff.create(before, after, keySerializer, DiffableValueSerializer.getWriteOnlyInstance());
+            : createDiff(before, after, keySerializer, DiffableValueSerializer.getWriteOnlyInstance());
     }
 
     /**
      * Calculates diff between two Maps of non-diffable objects
      */
-    public static <K, T> MapDiff<K, T, Map<K, T>> diff(
-        Map<K, T> before,
-        Map<K, T> after,
+    public static <K, T, M extends Map<K, T>> MapDiff<K, T, M> diff(
+        M before,
+        M after,
         KeySerializer<K> keySerializer,
         ValueSerializer<K, T> valueSerializer
     ) {
         assert after != null && before != null;
-        return before.equals(after) ? emptyDiff() : JdkMapDiff.create(before, after, keySerializer, valueSerializer);
+        return before.equals(after) ? emptyDiff() : createDiff(before, after, keySerializer, valueSerializer);
     }
 
     @SuppressWarnings("unchecked")
-    private static <K, T, M> MapDiff<K, T, M> emptyDiff() {
+    public static <K, T, M extends Map<K, T>> MapDiff<K, T, M> emptyDiff() {
         return (MapDiff<K, T, M>) EMPTY;
     }
 
@@ -120,7 +90,7 @@ public static <K, T> MapDiff<K, T, ImmutableOpenMap<K, T>> readImmutableOpenMapD
         KeySerializer<K> keySerializer,
         ValueSerializer<K, T> valueSerializer
     ) throws IOException {
-        return diffOrEmpty(new ImmutableOpenMapDiff<>(in, keySerializer, valueSerializer));
+        return diffOrEmpty(new MapDiff<>(in, keySerializer, valueSerializer, ImmutableOpenMapBuilder::new));
     }
 
     /**
@@ -131,7 +101,7 @@ public static <K, T> MapDiff<K, T, Map<K, T>> readJdkMapDiff(
         KeySerializer<K> keySerializer,
         ValueSerializer<K, T> valueSerializer
     ) throws IOException {
-        return diffOrEmpty(new JdkMapDiff<>(in, keySerializer, valueSerializer));
+        return diffOrEmpty(new MapDiff<>(in, keySerializer, valueSerializer, JdkMapBuilder::new));
     }
 
     /**
@@ -142,7 +112,7 @@ public static <K, T extends Diffable<T>> MapDiff<K, T, ImmutableOpenMap<K, T>> r
         KeySerializer<K> keySerializer,
         DiffableValueReader<K, T> diffableValueReader
     ) throws IOException {
-        return diffOrEmpty(new ImmutableOpenMapDiff<>(in, keySerializer, diffableValueReader));
+        return diffOrEmpty(new MapDiff<>(in, keySerializer, diffableValueReader, ImmutableOpenMapBuilder::new));
     }
 
     /**
@@ -154,10 +124,10 @@ public static <K, T extends Diffable<T>> MapDiff<K, T, Map<K, T>> readJdkMapDiff
         Reader<T> reader,
         Reader<Diff<T>> diffReader
     ) throws IOException {
-        return diffOrEmpty(new JdkMapDiff<>(in, keySerializer, new DiffableValueReader<>(reader, diffReader)));
+        return diffOrEmpty(new MapDiff<>(in, keySerializer, new DiffableValueReader<>(reader, diffReader), JdkMapBuilder::new));
     }
 
-    private static <K, T, H> MapDiff<K, T, H> diffOrEmpty(MapDiff<K, T, H> diff) {
+    private static <K, T, M extends Map<K, T>> MapDiff<K, T, M> diffOrEmpty(MapDiff<K, T, M> diff) {
         // TODO: refactor map diff reading to avoid having to construct empty diffs before throwing them away here
         if (diff.getUpserts().isEmpty() && diff.getDiffs().isEmpty() && diff.getDeletes().isEmpty()) {
             return emptyDiff();
@@ -165,166 +135,127 @@ private static <K, T, H> MapDiff<K, T, H> diffOrEmpty(MapDiff<K, T, H> diff) {
         return diff;
     }
 
-    /**
-     * Represents differences between two Maps of (possibly diffable) objects.
-     *
-     * @param <T> the diffable object
-     */
-    private static class JdkMapDiff<K, T> extends MapDiff<K, T, Map<K, T>> {
-
-        private JdkMapDiff(
-            KeySerializer<K> keySerializer,
-            ValueSerializer<K, T> valueSerializer,
-            List<K> deletes,
-            List<Map.Entry<K, Diff<T>>> diffs,
-            List<Map.Entry<K, T>> upserts
-        ) {
-            super(keySerializer, valueSerializer, deletes, diffs, upserts);
-        }
-
-        private JdkMapDiff(StreamInput in, KeySerializer<K> keySerializer, ValueSerializer<K, T> valueSerializer) throws IOException {
-            super(in, keySerializer, valueSerializer);
-        }
+    private static <K, T, M extends Map<K, T>> MapDiff<K, T, M> createDiff(
+        M before,
+        M after,
+        KeySerializer<K> keySerializer,
+        ValueSerializer<K, T> valueSerializer
+    ) {
+        assert after != null && before != null;
 
-        private static <K, T> JdkMapDiff<K, T> create(
-            Map<K, T> before,
-            Map<K, T> after,
-            KeySerializer<K> keySerializer,
-            ValueSerializer<K, T> valueSerializer
-        ) {
-            assert after != null && before != null;
-
-            int inserts = 0;
-            var upserts = new ArrayList<Map.Entry<K, T>>();
-            var diffs = new ArrayList<Map.Entry<K, Diff<T>>>();
-            for (Map.Entry<K, T> entry : after.entrySet()) {
-                T previousValue = before.get(entry.getKey());
-                if (previousValue == null) {
+        int inserts = 0;
+        var upserts = new ArrayList<Map.Entry<K, T>>();
+        var diffs = new ArrayList<Map.Entry<K, Diff<T>>>();
+        for (Map.Entry<K, T> entry : after.entrySet()) {
+            T previousValue = before.get(entry.getKey());
+            if (previousValue == null) {
+                upserts.add(entry);
+                inserts++;
+            } else if (entry.getValue().equals(previousValue) == false) {
+                if (valueSerializer.supportsDiffableValues()) {
+                    diffs.add(Map.entry(entry.getKey(), valueSerializer.diff(entry.getValue(), previousValue)));
+                } else {
                     upserts.add(entry);
-                    inserts++;
-                } else if (entry.getValue().equals(previousValue) == false) {
-                    if (valueSerializer.supportsDiffableValues()) {
-                        diffs.add(Map.entry(entry.getKey(), valueSerializer.diff(entry.getValue(), previousValue)));
-                    } else {
-                        upserts.add(entry);
-                    }
                 }
             }
+        }
 
-            int expectedDeletes = before.size() + inserts - after.size();
-            var deletes = new ArrayList<K>(expectedDeletes);
-            if (expectedDeletes > 0) {
-                for (K key : before.keySet()) {
-                    if (after.containsKey(key) == false) {
-                        deletes.add(key);
-                        if (--expectedDeletes == 0) {
-                            break;
-                        }
+        int expectedDeletes = before.size() + inserts - after.size();
+        var deletes = new ArrayList<K>(expectedDeletes);
+        if (expectedDeletes > 0) {
+            for (K key : before.keySet()) {
+                if (after.containsKey(key) == false) {
+                    deletes.add(key);
+                    if (--expectedDeletes == 0) {
+                        break;
                     }
                 }
             }
+        }
 
-            return new JdkMapDiff<>(keySerializer, valueSerializer, deletes, diffs, upserts);
+        Function<M, MapBuilder<K, T, M>> builderCtor;
+        if (before instanceof ImmutableOpenMap) {
+            builderCtor = DiffableUtils::createImmutableMapBuilder;
+        } else {
+            builderCtor = DiffableUtils::createJdkMapBuilder;
         }
 
-        @Override
-        public Map<K, T> apply(Map<K, T> map) {
-            Map<K, T> builder = new HashMap<>(map);
+        return new MapDiff<>(keySerializer, valueSerializer, deletes, diffs, upserts, builderCtor);
+    }
 
-            for (K part : deletes) {
-                builder.remove(part);
-            }
+    @SuppressWarnings("unchecked")
+    private static <K, T, M extends Map<K, T>> MapBuilder<K, T, M> createImmutableMapBuilder(Map<K, T> m) {
+        assert m instanceof ImmutableOpenMap<K, T>;
+        return (MapBuilder<K, T, M>) new ImmutableOpenMapBuilder<>((ImmutableOpenMap<K, T>) m);
+    }
 
-            for (Map.Entry<K, Diff<T>> diff : diffs) {
-                builder.put(diff.getKey(), diff.getValue().apply(builder.get(diff.getKey())));
-            }
+    @SuppressWarnings("unchecked")
+    private static <K, T, M extends Map<K, T>> MapBuilder<K, T, M> createJdkMapBuilder(Map<K, T> m) {
+        return (MapBuilder<K, T, M>) new JdkMapBuilder<>(m);
+    }
 
-            for (Map.Entry<K, T> upsert : upserts) {
-                builder.put(upsert.getKey(), upsert.getValue());
-            }
+    private interface MapBuilder<K, T, M extends Map<K, T>> {
+        void remove(K key);
 
-            return Collections.unmodifiableMap(builder);
-        }
+        T get(K key);
+
+        void put(K key, T value);
+
+        M build();
     }
 
-    /**
-     * Represents differences between two ImmutableOpenMap of (possibly diffable) objects
-     *
-     * @param <T> the object type
-     */
-    public static class ImmutableOpenMapDiff<K, T> extends MapDiff<K, T, ImmutableOpenMap<K, T>> {
+    private static class JdkMapBuilder<K, T> implements MapBuilder<K, T, Map<K, T>> {
+        private final Map<K, T> map;
 
-        private ImmutableOpenMapDiff(
-            KeySerializer<K> keySerializer,
-            ValueSerializer<K, T> valueSerializer,
-            List<K> deletes,
-            List<Map.Entry<K, Diff<T>>> diffs,
-            List<Map.Entry<K, T>> upserts
-        ) {
-            super(keySerializer, valueSerializer, deletes, diffs, upserts);
+        JdkMapBuilder(Map<K, T> map) {
+            this.map = new HashMap<>(map);
         }
 
-        private ImmutableOpenMapDiff(StreamInput in, KeySerializer<K> keySerializer, ValueSerializer<K, T> valueSerializer)
-            throws IOException {
-            super(in, keySerializer, valueSerializer);
+        @Override
+        public void remove(K key) {
+            map.remove(key);
         }
 
-        private static <K, T> ImmutableOpenMapDiff<K, T> create(
-            ImmutableOpenMap<K, T> before,
-            ImmutableOpenMap<K, T> after,
-            KeySerializer<K> keySerializer,
-            ValueSerializer<K, T> valueSerializer
-        ) {
-            assert after != null && before != null;
-
-            int inserts = 0;
-            var upserts = new ArrayList<Map.Entry<K, T>>();
-            var diffs = new ArrayList<Map.Entry<K, Diff<T>>>();
-            for (Map.Entry<K, T> entry : after.entrySet()) {
-                T beforeValue = before.get(entry.getKey());
-                if (beforeValue == null) {
-                    upserts.add(entry);
-                    inserts++;
-                } else if (entry.getValue().equals(beforeValue) == false) {
-                    if (valueSerializer.supportsDiffableValues()) {
-                        diffs.add(Map.entry(entry.getKey(), valueSerializer.diff(entry.getValue(), beforeValue)));
-                    } else {
-                        upserts.add(entry);
-                    }
-                }
-            }
+        @Override
+        public T get(K key) {
+            return map.get(key);
+        }
 
-            int expectedDeletes = before.size() + inserts - after.size();
-            var deletes = new ArrayList<K>(expectedDeletes);
-            if (expectedDeletes > 0) {
-                for (Map.Entry<K, T> key : before.entrySet()) {
-                    if (after.containsKey(key.getKey()) == false) {
-                        deletes.add(key.getKey());
-                        if (--expectedDeletes == 0) {
-                            break;
-                        }
-                    }
-                }
-            }
+        @Override
+        public void put(K key, T value) {
+            map.put(key, value);
+        }
+
+        @Override
+        public Map<K, T> build() {
+            return Collections.unmodifiableMap(map);
+        }
+    }
 
-            return new ImmutableOpenMapDiff<>(keySerializer, valueSerializer, deletes, diffs, upserts);
+    private static class ImmutableOpenMapBuilder<K, T> implements MapBuilder<K, T, ImmutableOpenMap<K, T>> {
+        private final ImmutableOpenMap.Builder<K, T> builder;
+
+        ImmutableOpenMapBuilder(ImmutableOpenMap<K, T> map) {
+            this.builder = ImmutableOpenMap.builder(map);
         }
 
         @Override
-        public ImmutableOpenMap<K, T> apply(ImmutableOpenMap<K, T> map) {
-            ImmutableOpenMap.Builder<K, T> builder = ImmutableOpenMap.builder(map);
+        public void remove(K key) {
+            builder.remove(key);
+        }
 
-            for (K part : deletes) {
-                builder.remove(part);
-            }
+        @Override
+        public T get(K key) {
+            return builder.get(key);
+        }
 
-            for (Map.Entry<K, Diff<T>> diff : diffs) {
-                builder.put(diff.getKey(), diff.getValue().apply(builder.get(diff.getKey())));
-            }
+        @Override
+        public void put(K key, T value) {
+            builder.put(key, value);
+        }
 
-            for (Map.Entry<K, T> upsert : upserts) {
-                builder.put(upsert.getKey(), upsert.getValue());
-            }
+        @Override
+        public ImmutableOpenMap<K, T> build() {
             return builder.build();
         }
     }
@@ -336,31 +267,38 @@ public ImmutableOpenMap<K, T> apply(ImmutableOpenMap<K, T> map) {
      *
      * @param <K> the type of map keys
      * @param <T> the type of map values
-     * @param <M> the map implementation type
      */
-    public abstract static class MapDiff<K, T, M> implements Diff<M> {
+    public static class MapDiff<K, T, M extends Map<K, T>> implements Diff<M> {
 
         protected final List<K> deletes;
         protected final List<Map.Entry<K, Diff<T>>> diffs; // incremental updates
         protected final List<Map.Entry<K, T>> upserts; // additions or full updates
         protected final KeySerializer<K> keySerializer;
         protected final ValueSerializer<K, T> valueSerializer;
+        private final Function<M, MapBuilder<K, T, M>> builderCtor;
 
-        protected MapDiff(
+        private MapDiff(
             KeySerializer<K> keySerializer,
             ValueSerializer<K, T> valueSerializer,
             List<K> deletes,
             List<Map.Entry<K, Diff<T>>> diffs,
-            List<Map.Entry<K, T>> upserts
+            List<Map.Entry<K, T>> upserts,
+            Function<M, MapBuilder<K, T, M>> builderCtor
         ) {
             this.keySerializer = keySerializer;
             this.valueSerializer = valueSerializer;
             this.deletes = deletes;
             this.diffs = diffs;
             this.upserts = upserts;
+            this.builderCtor = builderCtor;
         }
 
-        protected MapDiff(StreamInput in, KeySerializer<K> keySerializer, ValueSerializer<K, T> valueSerializer) throws IOException {
+        private MapDiff(
+            StreamInput in,
+            KeySerializer<K> keySerializer,
+            ValueSerializer<K, T> valueSerializer,
+            Function<M, MapBuilder<K, T, M>> builderCtor
+        ) throws IOException {
             this.keySerializer = keySerializer;
             this.valueSerializer = valueSerializer;
             deletes = in.readList(keySerializer::readKey);
@@ -378,6 +316,25 @@ protected MapDiff(StreamInput in, KeySerializer<K> keySerializer, ValueSerialize
                 T newValue = valueSerializer.read(in, key);
                 upserts.add(Map.entry(key, newValue));
             }
+            this.builderCtor = builderCtor;
+        }
+
+        @Override
+        public M apply(M map) {
+            MapBuilder<K, T, M> builder = builderCtor.apply(map);
+
+            for (K part : deletes) {
+                builder.remove(part);
+            }
+
+            for (Map.Entry<K, Diff<T>> diff : diffs) {
+                builder.put(diff.getKey(), diff.getValue().apply(builder.get(diff.getKey())));
+            }
+
+            for (Map.Entry<K, T> upsert : upserts) {
+                builder.put(upsert.getKey(), upsert.getValue());
+            }
+            return builder.build();
         }
 
         /**
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/ImmutableStateMetadata.java b/server/src/main/java/org/elasticsearch/cluster/metadata/ImmutableStateMetadata.java
index f876cf1328c47..22d8197aad8b3 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/ImmutableStateMetadata.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/ImmutableStateMetadata.java
@@ -9,7 +9,6 @@
 package org.elasticsearch.cluster.metadata;
 
 import org.elasticsearch.cluster.Diff;
-import org.elasticsearch.cluster.DiffableUtils;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -126,17 +125,6 @@ public static Diff<ImmutableStateMetadata> readDiffFrom(StreamInput in) throws I
         return SimpleDiffable.readDiffFrom(ImmutableStateMetadata::readFrom, in);
     }
 
-    /**
-     * Empty {@link org.elasticsearch.cluster.DiffableUtils.MapDiff} helper for metadata backwards compatibility.
-     */
-    public static final DiffableUtils.MapDiff<String, ImmutableStateMetadata, Map<String, ImmutableStateMetadata>> EMPTY_DIFF =
-        new DiffableUtils.MapDiff<>(null, null, List.of(), List.of(), List.of()) {
-            @Override
-            public Map<String, ImmutableStateMetadata> apply(Map<String, ImmutableStateMetadata> part) {
-                return part;
-            }
-        };
-
     /**
      * Convenience method for creating a {@link Builder} for {@link ImmutableStateMetadata}
      *
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java b/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java
index b7dea61814dc5..ac94180de6eaa 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java
@@ -1175,7 +1175,7 @@ private static class MetadataDiff implements Diff<Metadata> {
                     IMMUTABLE_DIFF_VALUE_READER
                 );
             } else {
-                immutableStateMetadata = ImmutableStateMetadata.EMPTY_DIFF;
+                immutableStateMetadata = DiffableUtils.emptyDiff();
             }
         }
 
diff --git a/server/src/test/java/org/elasticsearch/cluster/serialization/DiffableTests.java b/server/src/test/java/org/elasticsearch/cluster/serialization/DiffableTests.java
index da8eb24251b78..64dee2014f353 100644
--- a/server/src/test/java/org/elasticsearch/cluster/serialization/DiffableTests.java
+++ b/server/src/test/java/org/elasticsearch/cluster/serialization/DiffableTests.java
@@ -149,7 +149,7 @@ protected MapDiff<Integer, String, ImmutableOpenMap<Integer, String>> readDiff(S
      * @param <T> map type
      * @param <V> value type
      */
-    public abstract class MapDriver<T, V> {
+    public abstract class MapDriver<T extends Map<Integer, V>, V> {
         protected final Set<Integer> keys = randomPositiveIntSet();
         protected final Set<Integer> keysToRemove = new HashSet<>(randomSubsetOf(randomInt(keys.size()), keys.toArray(new Integer[0])));
         protected final Set<Integer> keysThatAreNotRemoved = Sets.difference(keys, keysToRemove);