Fix bug where timer would not be stopped if replica did not fetch any…

… segments.

This can happen when there is a new checkpoint published because of a SegmentInfos bump,
but the replica does not need to fetch any files.

Signed-off-by: Marc Handalian <handalm@amazon.com>

server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java

@@ -579,7 +579,7 @@ public void testSegmentReplicationStats() throws Exception {
             ensureGreen(INDEX_NAME);
             String docId = String.valueOf(initialDocCount + 1);
             client().prepareIndex(INDEX_NAME).setId(docId).setSource("foo", "bar").get();
-            flushAndRefresh(INDEX_NAME);
+            refresh(INDEX_NAME);
             waitForSearchableDocs(initialDocCount + 1, replicaNode_2);
 
             replicaNode_service = internalCluster().getInstance(SegmentReplicationPressureService.class, replicaNode);
@@ -588,6 +588,23 @@ public void testSegmentReplicationStats() throws Exception {
             replicaStats = replicaNode_service.nodeStats().getShardStats().get(primaryShard.shardId()).getReplicaStats();
             assertEquals(1, replicaStats.size());
             assertEquals(replica.routingEntry().currentNodeId(), replicaStats.stream().findFirst().get().getNodeId());
+
+            // test a checkpoint without any new segments
+            flush(INDEX_NAME);
+            assertBusy(() -> {
+                final SegmentReplicationPressureService service = internalCluster().getInstance(
+                    SegmentReplicationPressureService.class,
+                    replicaNode
+                );
+                assertEquals(1, service.nodeStats().getShardStats().size());
+                final Set<SegmentReplicationShardStats> shardStatsSet = service.nodeStats()
+                    .getShardStats()
+                    .get(primaryShard.shardId())
+                    .getReplicaStats();
+                assertEquals(1, shardStatsSet.size());
+                final SegmentReplicationShardStats stats = shardStatsSet.stream().findFirst().get();
+                assertEquals(0, stats.getCheckpointsBehindCount());
+            });
         }
     }
 }

server/src/main/java/org/opensearch/indices/replication/OngoingSegmentReplications.java

@@ -115,6 +115,10 @@ void startSegmentCopy(GetSegmentFilesRequest request, ActionListener<GetSegmentF
                 }
             });
             if (request.getFilesToFetch().isEmpty()) {
+                // before completion, alert the primary of the replica's state.
+                handler.getCopyState()
+                    .getShard()
+                    .updateVisibleCheckpointForShard(request.getTargetAllocationId(), handler.getCopyState().getCheckpoint());
                 wrappedListener.onResponse(new GetSegmentFilesResponse(Collections.emptyList()));
             } else {
                 handler.sendFiles(request, wrappedListener);