Fix SegmentReplicationPressureService to not schedule async tasks whe…

…n remote failures are disabled (#10569)

* Fix SegmentReplicationPressureService to not schedule async tasks when remote failures are disabled

Today segment replication's pressure service provides a setting 'segrep.replication.time.limit' that if set to a positive value
will spawn an Async task every 30s validating if there are replicas over that limit.  If over the limit the task considers if shards should be failed remotely.
The async task is being rescheduled every 30s, even if the setting is set to 0.  This change ensures that task isn't scheduled at all and reconsiders only if
the setting is updated.

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

* PR feedback.

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

* Revert "PR feedback."

This reverts commit 7371161.

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

* Add better unit test to ensure task is scheduled/unscheduled.

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

---------

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

server/src/main/java/org/opensearch/index/SegmentReplicationPressureService.java

@@ -98,7 +98,7 @@ public class SegmentReplicationPressureService implements Closeable {
     private final SegmentReplicationStatsTracker tracker;
     private final ShardStateAction shardStateAction;
 
-    private final AsyncFailStaleReplicaTask failStaleReplicaTask;
+    private volatile AsyncFailStaleReplicaTask failStaleReplicaTask;
 
     @Inject
     public SegmentReplicationPressureService(
@@ -202,6 +202,15 @@ public void setMaxAllowedStaleReplicas(double maxAllowedStaleReplicas) {
 
     public void setReplicationTimeLimitFailReplica(TimeValue replicationTimeLimitFailReplica) {
         this.replicationTimeLimitFailReplica = replicationTimeLimitFailReplica;
+        updateAsyncFailReplicaTask();
+    }
+
+    private synchronized void updateAsyncFailReplicaTask() {
+        try {
+            failStaleReplicaTask.close();
+        } finally {
+            failStaleReplicaTask = new AsyncFailStaleReplicaTask(this);
+        }
     }
 
     public void setReplicationTimeLimitBackpressure(TimeValue replicationTimeLimitBackpressure) {
@@ -228,13 +237,13 @@ final static class AsyncFailStaleReplicaTask extends AbstractAsyncTask {
 
         @Override
         protected boolean mustReschedule() {
-            return true;
+            return pressureService.shouldScheduleAsyncFailTask();
         }
 
         @Override
         protected void runInternal() {
             // Do not fail the replicas if time limit is set to 0 (i.e. disabled).
-            if (TimeValue.ZERO.equals(pressureService.replicationTimeLimitFailReplica) == false) {
+            if (pressureService.shouldScheduleAsyncFailTask()) {
                 final SegmentReplicationStats stats = pressureService.tracker.getStats();
 
                 // Find the shardId in node which is having stale replicas with highest current replication time.
@@ -302,4 +311,8 @@ public String toString() {
 
     }
 
+    boolean shouldScheduleAsyncFailTask() {
+        return TimeValue.ZERO.equals(replicationTimeLimitFailReplica) == false;
+    }
+
 }

server/src/test/java/org/opensearch/index/SegmentReplicationPressureServiceTests.java

@@ -217,6 +217,8 @@ public void testFailStaleReplicaTask() throws Exception {
             assertEquals(5, shardStats.getCheckpointsBehindCount());
 
             // call the background task
+            assertTrue(service.getFailStaleReplicaTask().mustReschedule());
+            assertTrue(service.getFailStaleReplicaTask().isScheduled());
             service.getFailStaleReplicaTask().runInternal();
 
             // verify that remote shard failed method is called which fails the replica shards falling behind.
@@ -257,6 +259,41 @@ public void testFailStaleReplicaTaskDisabled() throws Exception {
         }
     }
 
+    public void testFailStaleReplicaTaskToggleOnOff() throws Exception {
+        final Settings settings = Settings.builder()
+            .put(IndexMetadata.SETTING_REPLICATION_TYPE, ReplicationType.SEGMENT)
+            .put(SEGMENT_REPLICATION_INDEXING_PRESSURE_ENABLED.getKey(), true)
+            .put(MAX_REPLICATION_TIME_BACKPRESSURE_SETTING.getKey(), TimeValue.timeValueMillis(10))
+            .put(MAX_REPLICATION_LIMIT_STALE_REPLICA_SETTING.getKey(), TimeValue.timeValueMillis(1))
+            .build();
+
+        try (ReplicationGroup shards = createGroup(1, settings, new NRTReplicationEngineFactory())) {
+            shards.startAll();
+            final IndexShard primaryShard = shards.getPrimary();
+            SegmentReplicationPressureService service = buildPressureService(settings, primaryShard);
+
+            // index docs in batches without refreshing
+            indexInBatches(5, shards, primaryShard);
+
+            // assert that replica shard is few checkpoints behind primary
+            Set<SegmentReplicationShardStats> replicationStats = primaryShard.getReplicationStatsForTrackedReplicas();
+            assertEquals(1, replicationStats.size());
+            SegmentReplicationShardStats shardStats = replicationStats.stream().findFirst().get();
+            assertEquals(5, shardStats.getCheckpointsBehindCount());
+
+            assertTrue(service.getFailStaleReplicaTask().mustReschedule());
+            assertTrue(service.getFailStaleReplicaTask().isScheduled());
+            replicateSegments(primaryShard, shards.getReplicas());
+
+            service.setReplicationTimeLimitFailReplica(TimeValue.ZERO);
+            assertFalse(service.getFailStaleReplicaTask().mustReschedule());
+            assertFalse(service.getFailStaleReplicaTask().isScheduled());
+            service.setReplicationTimeLimitFailReplica(TimeValue.timeValueMillis(1));
+            assertTrue(service.getFailStaleReplicaTask().mustReschedule());
+            assertTrue(service.getFailStaleReplicaTask().isScheduled());
+        }
+    }
+
     private int indexInBatches(int count, ReplicationGroup shards, IndexShard primaryShard) throws Exception {
         int totalDocs = 0;
         for (int i = 0; i < count; i++) {