Removed PublishWithManifestRequest class

Signed-off-by: Shivansh Arora <[email protected]>
shiv0408 · Sep 3, 2024 · 61190c4 · 61190c4
1 parent 2fcaed3
commit 61190c4
Show file tree

Hide file tree

Showing 5 changed files with 25 additions and 69 deletions.
diff --git a/server/src/main/java/org/opensearch/cluster/coordination/CoordinationState.java b/server/src/main/java/org/opensearch/cluster/coordination/CoordinationState.java
@@ -632,11 +632,11 @@ private boolean shouldUpdateRemotePersistedState(PublishRequest publishRequest)
     }
 
     private void updateRemotePersistedStateOnPublishRequest(PublishRequest publishRequest) {
-        if (publishRequest instanceof PublishWithManifestRequest) {
+        if (publishRequest.hasManifest()) {
             assert persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE) != null;
             persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).setLastAcceptedState(publishRequest.getAcceptedState());
             persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE)
-                .setLastAcceptedManifest(((PublishWithManifestRequest) publishRequest).getAcceptedManifest());
+                .setLastAcceptedManifest(publishRequest.getAcceptedManifest().get());
         } else {
             // We will end up here if PublishRequest was sent not using Remote Store even with remotePublication enabled on this node
             persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).setLastAcceptedState(null);

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java b/server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java
@@ -325,7 +325,7 @@ private PublishWithJoinResponse acceptState(ClusterState incomingState, ClusterM
             }
         }
         if (manifest != null) {
-            return handlePublishRequest.apply(new PublishWithManifestRequest(incomingState, manifest));
+            return handlePublishRequest.apply(new PublishRequest(incomingState, manifest));
         }
         return handlePublishRequest.apply(new PublishRequest(incomingState));
     }

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/PublishRequest.java b/server/src/main/java/org/opensearch/cluster/coordination/PublishRequest.java
@@ -32,8 +32,10 @@
 package org.opensearch.cluster.coordination;
 
 import org.opensearch.cluster.ClusterState;
+import org.opensearch.gateway.remote.ClusterMetadataManifest;
 
 import java.util.Objects;
+import java.util.Optional;
 
 /**
  * Request which is used by the cluster-manager node to publish cluster state changes.
@@ -44,32 +46,47 @@
 public class PublishRequest {
 
     private final ClusterState acceptedState;
+    private final ClusterMetadataManifest acceptedManifest;
 
     public PublishRequest(ClusterState acceptedState) {
         this.acceptedState = acceptedState;
+        this.acceptedManifest = null;
+    }
+
+    public PublishRequest(ClusterState acceptedState, ClusterMetadataManifest acceptedManifest) {
+        this.acceptedState = acceptedState;
+        this.acceptedManifest = acceptedManifest;
     }
 
     public ClusterState getAcceptedState() {
         return acceptedState;
     }
 
+    public boolean hasManifest() {
+        return acceptedManifest != null;
+    }
+
+    public Optional<ClusterMetadataManifest> getAcceptedManifest() {
+        return Optional.ofNullable(acceptedManifest);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
         if (!(o instanceof PublishRequest)) return false;
 
         PublishRequest that = (PublishRequest) o;
 
-        return acceptedState.term() == that.acceptedState.term() && acceptedState.version() == that.acceptedState.version();
+        return acceptedState.term() == that.acceptedState.term() && acceptedState.version() == that.acceptedState.version() && Objects.equals(acceptedManifest, that.acceptedManifest);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(acceptedState.term(), acceptedState.version());
+        return Objects.hash(acceptedState.term(), acceptedState.version(), acceptedManifest);
     }
 
     @Override
     public String toString() {
-        return "PublishRequest{term=" + acceptedState.term() + ", version=" + acceptedState.version() + ", state=" + acceptedState + '}';
+        return "PublishRequest{term=" + acceptedState.term() + ", version=" + acceptedState.version() + ", state=" + acceptedState + ", manifest=" + acceptedManifest + '}';
     }
 }
diff --git a/server/src/main/java/org/opensearch/cluster/coordination/PublishWithManifestRequest.java b/server/src/main/java/org/opensearch/cluster/coordination/PublishWithManifestRequest.java
diff --git a/server/src/test/java/org/opensearch/cluster/coordination/CoordinationStateTests.java b/server/src/test/java/org/opensearch/cluster/coordination/CoordinationStateTests.java
@@ -1066,7 +1066,7 @@ public void testHandlePublishRequestOnFollowerWhenRemotePublicationEnabled() {
             .clusterUUIDCommitted(true)
             .build();
 
-        PublishResponse publishResponse = coordinationState.handlePublishRequest(new PublishWithManifestRequest(state2, manifest));
+        PublishResponse publishResponse = coordinationState.handlePublishRequest(new PublishRequest(state2, manifest));
         assertEquals(state2.term(), publishResponse.getTerm());
         assertEquals(state2.version(), publishResponse.getVersion());
         verifyNoInteractions(remoteClusterStateService);
@@ -1141,7 +1141,7 @@ public void testHandleCommitOnFollowerNodeWhenRemotePublicationEnabled() {
             .build();
 
         PublishRequest publishRequest = coordinationState.handleClientValue(state2);
-        coordinationState.handlePublishRequest(new PublishWithManifestRequest(publishRequest.getAcceptedState(), manifest));
+        coordinationState.handlePublishRequest(new PublishRequest(publishRequest.getAcceptedState(), manifest));
         ApplyCommitRequest applyCommitRequest = new ApplyCommitRequest(node2, state2.term(), state2.version());
         coordinationState.handleCommit(applyCommitRequest);
         verifyNoInteractions(remoteClusterStateService);