Update last seen cluster state in commit phase (#16215)

* Update last seen cluster state on apply commit

Signed-off-by: Sooraj Sinha <[email protected]>

Author: soosinha

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
 

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.

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);

server/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();