Send manifest file name in remote publish request

Signed-off-by: Sooraj Sinha <soosinha@amazon.com>
shiv0408 · Jun 7, 2024 · fa24988 · fa24988
1 parent 79eb089
commit fa24988
Show file tree

Hide file tree

Showing 5 changed files with 13 additions and 8 deletions.
diff --git a/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java b/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java
@@ -1327,7 +1327,7 @@ assert getLocalNode().equals(clusterState.getNodes().get(getLocalNode().getId())
                 final PublicationTransportHandler.PublicationContext publicationContext = publicationHandler.newPublicationContext(
                     clusterChangedEvent,
                     coordinationState.get().isRemotePublicationEnabled(),
-                    ((RemotePersistedState) persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE))
+                    persistedStateRegistry
                 );
 
                 final PublishRequest publishRequest = coordinationState.get().handleClientValue(clusterState);

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java b/server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java
@@ -300,8 +300,8 @@ private PublishWithJoinResponse acceptStateOnLocalNode(RemotePublishRequest remo
     }
 
     public PublicationContext newPublicationContext(ClusterChangedEvent clusterChangedEvent, boolean isRemotePublicationEnabled,
-        String manifestFileName) {
-        final PublicationContext publicationContext = new PublicationContext(clusterChangedEvent, isRemotePublicationEnabled, manifestFileName);
+        PersistedStateRegistry persistedStateRegistry) {
+        final PublicationContext publicationContext = new PublicationContext(clusterChangedEvent, isRemotePublicationEnabled, persistedStateRegistry);
 
         // Build the serializations we expect to need now, early in the process, so that an error during serialization fails the publication
         // straight away. This isn't watertight since we send diffs on a best-effort basis and may fall back to sending a full state (and
@@ -347,15 +347,15 @@ public class PublicationContext {
         private final Map<Version, BytesReference> serializedStates = new HashMap<>();
         private final Map<Version, BytesReference> serializedDiffs = new HashMap<>();
         private final boolean sendRemoteState;
-        private final String manifestFileName;
+        private final PersistedStateRegistry persistedStateRegistry;
 
-        PublicationContext(ClusterChangedEvent clusterChangedEvent, boolean isRemotePublicationEnabled, String manifestFileName) {
+        PublicationContext(ClusterChangedEvent clusterChangedEvent, boolean isRemotePublicationEnabled, PersistedStateRegistry persistedStateRegistry) {
             discoveryNodes = clusterChangedEvent.state().nodes();
             newState = clusterChangedEvent.state();
             previousState = clusterChangedEvent.previousState();
             sendFullVersion = previousState.getBlocks().disableStatePersistence();
             sendRemoteState = isRemotePublicationEnabled;
-            this.manifestFileName = manifestFileName;
+            this.persistedStateRegistry = persistedStateRegistry;
         }
 
         void buildDiffAndSerializeStates() {
@@ -468,6 +468,7 @@ public String executor() {
 
         private void sendRemoteClusterState(final DiscoveryNode destination, final ClusterState clusterState, final ActionListener<PublishWithJoinResponse> listener) {
             try {
+                final String manifestFileName = ((RemotePersistedState) persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE)).getLastUploadedManifestFile();
                 final RemotePublishRequest remotePublishRequest = new RemotePublishRequest(discoveryNodes.getLocalNode(), clusterState.term(),
                     clusterState.getVersion(), clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), manifestFileName);
                 final Consumer<TransportException> transportExceptionHandler = exp -> {

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/RemotePublishRequest.java b/server/src/main/java/org/opensearch/cluster/coordination/RemotePublishRequest.java
@@ -16,7 +16,6 @@
 
 public class RemotePublishRequest extends TermVersionRequest {
 
-    // todo Do we need cluster name and UUID ?
     private final String clusterName;
     private final String clusterUUID;
     private final String manifestFile;

diff --git a/server/src/main/java/org/opensearch/gateway/GatewayMetaState.java b/server/src/main/java/org/opensearch/gateway/GatewayMetaState.java
@@ -666,6 +666,7 @@ public static class RemotePersistedState implements PersistedState {
 
         private ClusterState lastAcceptedState;
         private ClusterMetadataManifest lastAcceptedManifest;
+
         private String lastUploadedManifestFile;
         private final RemoteClusterStateService remoteClusterStateService;
         private String previousClusterUUID;
@@ -692,6 +693,10 @@ public void setCurrentTerm(long currentTerm) {
             // But for RemotePersistedState, the state is only pushed by the active cluster. So this method is not required.
         }
 
+        public String getLastUploadedManifestFile() {
+            return lastUploadedManifestFile;
+        }
+
         @Override
         public void setLastAcceptedState(ClusterState clusterState) {
             try {

diff --git a/...r/src/test/java/org/opensearch/cluster/coordination/PublicationTransportHandlerTests.java b/...r/src/test/java/org/opensearch/cluster/coordination/PublicationTransportHandlerTests.java
@@ -112,7 +112,7 @@ public void writeTo(StreamOutput out) throws IOException {
 
         OpenSearchException e = expectThrows(
             OpenSearchException.class,
-            () -> handler.newPublicationContext(new ClusterChangedEvent("test", unserializableClusterState, clusterState), false)
+            () -> handler.newPublicationContext(new ClusterChangedEvent("test", unserializableClusterState, clusterState), false, null)
         );
         assertNotNull(e.getCause());
         assertThat(e.getCause(), instanceOf(IOException.class));