modify test

Signed-off-by: guojialiang <[email protected]>

Author: guojialiang92

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

@@ -185,16 +185,51 @@ public void testPublishCheckPointFail() throws Exception {
     }
 
     public void testPrimaryStopped_ReplicaPromoted() throws Exception {
-        final String primary = internalCluster().startDataOnlyNode();
+        Settings mockNodeSetting = Settings.builder()
+            .put(TransportReplicationAction.REPLICATION_RETRY_TIMEOUT.getKey(), TimeValue.timeValueSeconds(0))
+            .build();
+
+        final String primary = internalCluster().startDataOnlyNode(mockNodeSetting);
         createIndex(INDEX_NAME);
         ensureYellowAndNoInitializingShards(INDEX_NAME);
-        final String replica = internalCluster().startDataOnlyNode();
+        final String replica = internalCluster().startDataOnlyNode(mockNodeSetting);
         ensureGreen(INDEX_NAME);
 
+        MockTransportService replicaTransportService = ((MockTransportService) internalCluster().getInstance(
+            TransportService.class,
+            replica
+        ));
+        AtomicBoolean mockReplicaReceivePublishCheckpointException = new AtomicBoolean(true);
+        replicaTransportService.addRequestHandlingBehavior(
+            PublishCheckpointAction.ACTION_NAME + TransportReplicationAction.REPLICA_ACTION_SUFFIX,
+            (handler, request, channel, task) -> {
+                if (mockReplicaReceivePublishCheckpointException.get()) {
+                    logger.info("mock remote transport exception");
+                    throw new RemoteTransportException("mock remote transport exception", new OpenSearchRejectedExecutionException());
+                }
+                logger.info("replica receive publish checkpoint request");
+                handler.messageReceived(request, channel, task);
+            }
+        );
+
         client().prepareIndex(INDEX_NAME).setId("1").setSource("foo", "bar").setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
         refresh(INDEX_NAME);
+        logger.info("ensure publish checkpoint request can be process");
+        mockReplicaReceivePublishCheckpointException.set(false);
 
         waitForSearchableDocs(1, primary, replica);
+        replicaTransportService.clearAllRules();
+
+        assertAcked(
+            client().admin()
+                .indices()
+                .prepareUpdateSettings(INDEX_NAME)
+                .setSettings(
+                    Settings.builder()
+                        .put(IndexSettings.INDEX_PUBLISH_CHECKPOINT_INTERVAL_SETTING.getKey(), TimeValue.timeValueSeconds(3))
+                        .put(IndexSettings.INDEX_LAG_TIME_BEFORE_RESEND_CHECKPOINT_SETTING.getKey(), TimeValue.timeValueSeconds(3))
+                )
+        );
 
         // index another doc but don't refresh, we will ensure this is searchable once replica is promoted.
         client().prepareIndex(INDEX_NAME).setId("2").setSource("bar", "baz").setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();

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

@@ -792,8 +792,8 @@ public static IndexMergePolicy fromString(String text) {
 
     public static final Setting<TimeValue> INDEX_LAG_TIME_BEFORE_RESEND_CHECKPOINT_SETTING = Setting.timeSetting(
         "index.lag_time_before_resend_checkpoint",
-        TimeValue.timeValueSeconds(1),
-        TimeValue.timeValueSeconds(1),
+        TimeValue.timeValueSeconds(0),
+        TimeValue.timeValueSeconds(0),
         Property.Dynamic,
         Property.IndexScope
     );

server/src/main/java/org/opensearch/index/seqno/ReplicationTracker.java

@@ -1305,7 +1305,7 @@ public synchronized boolean hasLaggingReplicas() {
                     && replicationGroup.getUnavailableInSyncShards().contains(allocationId) == false
                     && shouldSkipReplicationTimer(allocationId) == false
                     && cps.checkpointTimers.containsKey(latestReplicationCheckpoint)
-                    && cps.checkpointTimers.get(latestReplicationCheckpoint).time() > indexSettings.getLagTimeBeforeResendCheckpoint()
+                    && cps.checkpointTimers.get(latestReplicationCheckpoint).time() >= indexSettings.getLagTimeBeforeResendCheckpoint()
                         .millis()) {
                     hasLaggingReplicas = true;
                 }

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -4720,7 +4720,6 @@ public void scheduledPublishCheckpoint() {
         assert indexSettings.isSegRepEnabledOrRemoteNode();
         assert isPrimaryMode();
         if (replicationTracker.hasLaggingReplicas()) {
-            logger.trace("publish checkpoint {} to replicas", getLatestReplicationCheckpoint());
             checkpointPublisher.publish(this, getLatestReplicationCheckpoint());
         }
     }