Add POJO classes required for cluster state publication from remote

server/src/main/java/org/opensearch/cluster/DiffableUtils.java

@@ -494,6 +494,18 @@ public void writeDiff(Diff<V> value, StreamOutput out) throws IOException {
      * @opensearch.internal
      */
     public abstract static class NonDiffableValueSerializer<K, V> implements ValueSerializer<K, V> {
+        private static final NonDiffableValueSerializer ABSTRACT_INSTANCE = new NonDiffableValueSerializer<>() {
+            @Override
+            public void write(Object value, StreamOutput out) {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public Object read(StreamInput in, Object key) {
+                throw new UnsupportedOperationException();
+            }
+        };
+
         @Override
         public boolean supportsDiffableValues() {
             return false;
@@ -513,6 +525,10 @@ public void writeDiff(Diff<V> value, StreamOutput out) throws IOException {
         public Diff<V> readDiff(StreamInput in, K key) throws IOException {
             throw new UnsupportedOperationException();
         }
+
+        public static <K, V> NonDiffableValueSerializer<K, V> getAbstractInstance() {
+            return ABSTRACT_INSTANCE;
+        }
     }
 
     /**

server/src/main/java/org/opensearch/cluster/metadata/DiffableStringMap.java

@@ -66,7 +66,7 @@ public static DiffableStringMap readFrom(StreamInput in) throws IOException {
         return map.isEmpty() ? EMPTY : new DiffableStringMap(map);
     }
 
-    DiffableStringMap(final Map<String, String> map) {
+    public DiffableStringMap(final Map<String, String> map) {
         this.innerMap = Collections.unmodifiableMap(map);
     }
 

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

@@ -973,6 +973,10 @@ public static boolean isSettingsMetadataEqual(Metadata metadata1, Metadata metad
         return metadata1.persistentSettings.equals(metadata2.persistentSettings);
     }
 
+    public static boolean isTransientSettingsMetadataEqual(Metadata metadata1, Metadata metadata2) {
+        return metadata1.transientSettings.equals(metadata2.transientSettings);
+    }
+
     public static boolean isTemplatesMetadataEqual(Metadata metadata1, Metadata metadata2) {
         return metadata1.templates.equals(metadata2.templates);
     }

server/src/main/java/org/opensearch/cluster/routing/remote/RemoteRoutingTableService.java

@@ -10,16 +10,22 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.cluster.DiffableUtils;
+import org.opensearch.cluster.routing.IndexRoutingTable;
+import org.opensearch.cluster.routing.RoutingTable;
 import org.opensearch.common.lifecycle.AbstractLifecycleComponent;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.io.IOUtils;
+import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.node.Node;
 import org.opensearch.node.remotestore.RemoteStoreNodeAttribute;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.repositories.Repository;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
 
 import java.io.IOException;
+import java.util.Map;
 import java.util.function.Supplier;
 
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled;
@@ -36,12 +42,37 @@ public class RemoteRoutingTableService extends AbstractLifecycleComponent {
     private final Supplier<RepositoriesService> repositoriesService;
     private BlobStoreRepository blobStoreRepository;
 
+    private static final DiffableUtils.NonDiffableValueSerializer<String, IndexRoutingTable> CUSTOM_ROUTING_TABLE_VALUE_SERIALIZER =
+        new DiffableUtils.NonDiffableValueSerializer<String, IndexRoutingTable>() {
+            @Override
+            public void write(IndexRoutingTable value, StreamOutput out) throws IOException {
+                value.writeTo(out);
+            }
+
+            @Override
+            public IndexRoutingTable read(StreamInput in, String key) throws IOException {
+                return IndexRoutingTable.readFrom(in);
+            }
+        };
+
     public RemoteRoutingTableService(Supplier<RepositoriesService> repositoriesService, Settings settings) {
         assert isRemoteRoutingTableEnabled(settings) : "Remote routing table is not enabled";
         this.repositoriesService = repositoriesService;
         this.settings = settings;
     }
 
+    public static DiffableUtils.MapDiff<String, IndexRoutingTable, Map<String, IndexRoutingTable>> getIndicesRoutingMapDiff(
+        RoutingTable before,
+        RoutingTable after
+    ) {
+        return DiffableUtils.diff(
+            before.getIndicesRouting(),
+            after.getIndicesRouting(),
+            DiffableUtils.getStringKeySerializer(),
+            CUSTOM_ROUTING_TABLE_VALUE_SERIALIZER
+        );
+    }
+
     @Override
     protected void doClose() throws IOException {
         if (blobStoreRepository != null) {