Update last seen cluster state in commit phase (opensearch-project#16215

) * Update last seen cluster state on apply commit Signed-off-by: Sooraj Sinha <[email protected]>
dk2k · Oct 17, 2024 · 1ddf351 · 1ddf351
1 parent c8e97e9
commit 1ddf351
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 5 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -62,6 +62,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Enable coordinator search.request_stats_enabled by default ([#16290](https://github.com/opensearch-project/OpenSearch/pull/16290))
 - Code cleanup: Remove ApproximateIndexOrDocValuesQuery ([#16273](https://github.com/opensearch-project/OpenSearch/pull/16273))
 - Optimise clone operation for incremental full cluster snapshots ([#16296](https://github.com/opensearch-project/OpenSearch/pull/16296))
+- Update last seen cluster state in the commit phase ([#16215](https://github.com/opensearch-project/OpenSearch/pull/16215))
 
 ### Deprecated
 

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java b/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java
@@ -105,6 +105,7 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -383,14 +384,19 @@ void onFollowerCheckRequest(FollowerCheckRequest followerCheckRequest) {
         }
     }
 
-    private void handleApplyCommit(ApplyCommitRequest applyCommitRequest, ActionListener<Void> applyListener) {
+    private void handleApplyCommit(
+        ApplyCommitRequest applyCommitRequest,
+        Consumer<ClusterState> updateLastSeen,
+        ActionListener<Void> applyListener
+    ) {
         synchronized (mutex) {
             logger.trace("handleApplyCommit: applying commit {}", applyCommitRequest);
 
             coordinationState.get().handleCommit(applyCommitRequest);
             final ClusterState committedState = hideStateIfNotRecovered(coordinationState.get().getLastAcceptedState());
             applierState = mode == Mode.CANDIDATE ? clusterStateWithNoClusterManagerBlock(committedState) : committedState;
             clusterApplier.setPreCommitState(applierState);
+            updateLastSeen.accept(coordinationState.get().getLastAcceptedState());
 
             if (applyCommitRequest.getSourceNode().equals(getLocalNode())) {
                 // cluster-manager node applies the committed state at the end of the publication process, not here.

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java b/server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java
@@ -43,6 +43,7 @@
 import org.opensearch.cluster.coordination.PersistedStateRegistry.PersistedStateType;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.cluster.node.DiscoveryNodes;
+import org.opensearch.common.TriConsumer;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.bytes.BytesReference;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
@@ -65,7 +66,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
@@ -110,7 +110,7 @@ public PublicationTransportHandler(
         TransportService transportService,
         NamedWriteableRegistry namedWriteableRegistry,
         Function<PublishRequest, PublishWithJoinResponse> handlePublishRequest,
-        BiConsumer<ApplyCommitRequest, ActionListener<Void>> handleApplyCommit,
+        TriConsumer<ApplyCommitRequest, Consumer<ClusterState>, ActionListener<Void>> handleApplyCommit,
         RemoteClusterStateService remoteClusterStateService
     ) {
         this.transportService = transportService;
@@ -142,7 +142,7 @@ public PublicationTransportHandler(
             false,
             false,
             ApplyCommitRequest::new,
-            (request, channel, task) -> handleApplyCommit.accept(request, transportCommitCallback(channel))
+            (request, channel, task) -> handleApplyCommit.apply(request, this::updateLastSeen, transportCommitCallback(channel))
         );
     }
 
@@ -377,6 +377,10 @@ private boolean validateRemotePublicationConfiguredOnAllNodes(DiscoveryNodes dis
         return true;
     }
 
+    private void updateLastSeen(final ClusterState clusterState) {
+        lastSeenClusterState.set(clusterState);
+    }
+
     // package private for testing
     void setCurrentPublishRequestToSelf(PublishRequest publishRequest) {
         this.currentPublishRequestToSelf.set(publishRequest);

diff --git a/...r/src/test/java/org/opensearch/cluster/coordination/PublicationTransportHandlerTests.java b/...r/src/test/java/org/opensearch/cluster/coordination/PublicationTransportHandlerTests.java
@@ -466,7 +466,7 @@ private PublicationTransportHandler getPublicationTransportHandler(
             transportService,
             writableRegistry(),
             handlePublishRequest,
-            (pu, l) -> {},
+            (pu, uc, l) -> {},
             remoteClusterStateService
         );
         transportService.start();