[RFC][segment replication] Introduce a heuristic method to avoid long-term write blocking during primary shard relocation

[guojialiang92]

Is your feature request related to a problem? Please describe

Segment replication, as the core function of community storage and computation separation and read-write separation, has become increasingly important. When enabling segment replication, the replica shard does not perform segment construction, which can save a lot of CPU overhead. Therefore, it is widely used in production environments.

In production practice, we found that during the primary shard relocation, the index TPS (docs/s) may drop to 0 for a long time. Scaling nodes in the production environment cluster is a regular operation, and the migration of the primary shard often occurs, which prompts us to explore the reasons and find solutions.

As we all know, the relocation process of the primary shard will go through the peer recovery. In the document-replication scenario, the new primary will use InternalEngine and build segments during the migration process. The segment-replication scenario is somewhat different. In order to avoid file conflict, #[5344] introduces a round of force segment replication.

For the convenience of reading, the Description in #[5344] is quoted here：

Update the peer recovery logic to work with segment replication. The existing primary relocation is broken because of segment files conflicts. This happens because of target (new primary) is started with InternalEngine writes its own version of segment files which later on conflicts with file copies on replicas (copied from older primary); when this target is promoted as primary.
This change patches the recovery flow by first recovering the primary target with NRTReplicationEngine engine (to prevent target from generating segment files). Post phase 2 before finalizeRecovery and handoff, the target (new primary) is forced to refresh segment files from older primary (to catch up with other replicas) followed by switching engine to InternalEngine.

This method has been validated for a long time and can indeed avoid file conflicts. However, this proposal also has some costs, such as blocking writes during the forced segment replication phase (In order to catch up with the latest checkpoint before hand-off phase, write blocking is necessary).

We used three 16C64GB nodes, nyc_taxis workload, 1 primary shard and 1 replica shard for testing. During the write process, relocation the primary shard to reproduce the issue of write TPS drop.

Describe the solution you'd like

Background

Now, Lucene supports modifying segmentInfos.counter in IndexWriter (#[14417]). We believe that it is possible to avoid file conflicts and ensure that writing is not affected.

Proposed Solution

Just like document-replication, use InternalEngine directly when the new primary shard starts. At the same time, we will advance the segmentInfos.counter of the new primary shard based on the totalTranslogOps estimated by the old primary shard.

new_primary_advanced_segment_counter = base_segment_counter + totalTranslogOps * retry_count

• new_primary_advanced_segment_counter is the segment counter used by the new primary shard to create InternalEngine.
• base_segment_counter is the segment counter copied from the old primary shard in peer recovery phase1.
• totalTranslogOps is the number of translogs estimated by the old primary shard.
• retry_count is the number of retries, the starting value is 1.

Because the step size is at least totalTranslogOps, even if each write operation generates a new segment, it is difficult to have conflicts. However, because new write requests will continue to be received in the peer recovery process, we need to verify whether the local segment counter exceeds the new_primary_advanced_segment_counter in the old primary shard, and cancel the peer recovery process when it exceeds.

Failover

If a failure occurs, it means that our increased step size cannot cover the incremental writes in the peer recovery process. We need to increase the retry_count and expand the step size in the next migration.

Because the ReplicationTracker#primaryMode is set to false before the new primary shard is promoted, we will not let it publish checkpoints to other replicas. Therefore, it is safe to cancel the recovery process when the old primary shard detects a conflict.

Results

We use the same method to test and verify the effect.

Related component

Indexing:Replication

Describe alternatives you've considered

No response

Additional context

No response

[guojialiang92]

@Bukhtawar @sachinpkale @gbbafna @mch2 @ashking94 @shourya035 - Do take a look and give your feedback.

[Bukhtawar]

I believe the operations get buffered in a queue, while we force the segrep round and the proposal is to provide a conflict free way of continuing ingestion by reserving seq no upfront

[guojialiang92]

Thank you for your reply @Bukhtawar

I believe the operations get buffered in a queue, while we force the segrep round and the proposal is to provide a conflict free way of continuing ingestion by reserving seq no upfront

Yeah, your understanding is correct. Performing a round of segment replication before the end of the primary shard relocation can achieve the purpose of avoiding conflicts. If there is a long-term segment replication, the write request will also be added to the IndexShardOperationPermits#delayedOperations.

Objective

This RFC focuses on the problem of write blocking that may occur under the current scheme. We are trying to find a solution that can avoid both write blocking and file conflicts.

Reproduce

I will demonstrate write blocking through IT test (https://github.com/guojialiang92/OpenSearch/tree/dev/test_primary_relocate).

For the convenience of reading, I will briefly introduce the relevant changes.

• Added a test SegmentReplicationIT#testPrimaryRelocate. Continuously write documents during the primary shard relocation process, pay attention to the write time, and verify the number of documents for the new primary shard and replica shard after the relocation is complete.
• In SegmentReplicationTargetService#forceReplication, Thread.sleep (20000) logic was added to simulate long-term force replication processes.

public void testPrimaryRelocate() throws Exception {
    final String primary = internalCluster().startDataOnlyNode();
    createIndex(INDEX_NAME);
    ensureYellowAndNoInitializingShards(INDEX_NAME);
    final String replica = internalCluster().startDataOnlyNode();
    ensureGreen(INDEX_NAME);
    final String newNode = internalCluster().startDataOnlyNode();

    CountDownLatch countDownLatch = new CountDownLatch(1);
    Thread thread = new Thread(new Runnable() {
        @Override
        public void run() {
            for (int i = 0; i < 20; i++) {
                long startTime = System.currentTimeMillis();
                logger.info("begin write doc {}", i);
                try {
                    // case1: write request without timeout
                    // The write thread will be blocked, the index TPS (docs/s) will be affected, but no data will be lost
                    client().prepareIndex(INDEX_NAME).setId(String.valueOf(i)).setSource("foo" + i, "bar" + i).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();

                    // case2: write request with timeout
                    // After a long-term of force segment replication, it may have timed out when retrying, resulting in data loss
//                        client().prepareIndex(INDEX_NAME).setId(String.valueOf(i)).setSource("foo" + i, "bar" + i).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).setTimeout(TimeValue.timeValueSeconds(5)).get();
                    logger.info("after write doc {}, cost {}ms", i, System.currentTimeMillis() - startTime);
                } catch (Exception e) {
                    logger.error("write doc " + i + " cost " + (System.currentTimeMillis() - startTime) + "ms generate exception", e);
                }
                if (i == 10) {
                    countDownLatch.countDown();
                }
            }
            refresh(INDEX_NAME);
        }
    });
    thread.start();
    countDownLatch.await();

    logger.info("primary shard relocate from {} to {}", primary, newNode);
    client().admin()
        .cluster()
        .prepareReroute()
        .add(new MoveAllocationCommand(INDEX_NAME, 0, primary, newNode))
        .execute()
        .actionGet()
        .getState();

    thread.join();
    ensureGreen(INDEX_NAME);
    logger.info("relocation finish, index is green");
    waitForSearchableDocs(20, newNode, replica);
}

private void forceReplication(ForceSyncRequest request, ActionListener<TransportResponse> listener) {
    final ShardId shardId = request.getShardId();
    assert indicesService != null;
    final IndexShard indexShard = indicesService.getShardOrNull(shardId);
    try {
        logger.info("mock long-term forceReplication");
        Thread.sleep(20000);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
    // ....
}

Write block

As shown in the above code, we test two write modes:

• write request without timeout

Writing threads will be blocked for more than 20s, After the peer recovery process is completed, the operations in IndexShardOperationPermits#delayedOperations will be retried and no data will be lost. The relevant logs are as follows.

[2025-05-23T01:19:34,975][INFO ][o.o.i.r.SegmentReplicationIT] [testPrimaryRelocate] primary shard relocate from node_t1 to node_t3

[2025-05-23T01:19:35,184][INFO ][o.o.i.r.SegmentReplicationIT] [[Thread-1]] begin write doc 14

[2025-05-23T01:19:55,400][INFO ][o.o.i.r.RecoverySourceHandler] [node_t1] [test-idx-1][0][recover to node_t3] finalizing recovery took [20.2s]

[2025-05-23T01:19:55,402][INFO ][o.o.i.s.IndexShardOperationPermits] [node_t1] replay queue action [delayed] BulkShardRequest [[test-idx-1][0]] containing [index {[test-idx-1][14], source[{"foo14":"bar14"}]}] and a refresh

[2025-05-23T01:19:55,568][INFO ][o.o.i.r.SegmentReplicationIT] [[Thread-1]] after write doc 14, cost 20384ms

[2025-05-23T01:19:55,568][INFO ][o.o.i.r.SegmentReplicationIT] [[Thread-1]] begin write doc 15

[2025-05-23T01:19:55,648][INFO ][o.o.i.r.SegmentReplicationIT] [[Thread-1]] after write doc 15, cost 80ms

• write request with timeout

Writing threads will be blocked for more than 20s, After the peer recovery process is completed, the operations in IndexShardOperationPermits#delayedOperations will be retried, but data loss will occur due to timeout. The relevant logs are as follows.

[2025-05-23T11:25:32,370][INFO ][o.o.i.r.SegmentReplicationIT] [testPrimaryRelocate] primary shard relocate from node_t1 to node_t3

[2025-05-23T11:25:32,533][INFO ][o.o.i.r.SegmentReplicationIT] [[Thread-1]] begin write doc 14

[2025-05-23T11:25:52,674][INFO ][o.o.i.r.RecoverySourceHandler] [node_t1] [test-idx-1][0][recover to node_t3] finalizing recovery took [20.1s]

[2025-05-23T11:25:52,678][INFO ][o.o.a.b.TransportShardBulkAction] [node_t1] received an error from node [ZphISrAIRFK4pxsE22CHuA] for request [request: BulkShardRequest [[test-idx-1][0]] containing [index {[test-idx-1][14], source[{"foo14":"bar14"}]}] and a refresh, target allocation id: yySpXJpZQHmWuUvxfgcsOA, primary term: 1], scheduling a retry
org.opensearch.transport.RemoteTransportException: [node_t1][127.0.0.1:55719][indices:data/write/bulk[s][p]]
Caused by: org.opensearch.action.support.replication.ReplicationOperation$RetryOnPrimaryException: shard is not in primary mode

[2025-05-23T11:25:52,684][INFO ][o.o.a.b.TransportShardBulkAction] [node_t1] request is timeout true
org.opensearch.transport.RemoteTransportException: [node_t1][127.0.0.1:55719][indices:data/write/bulk[s][p]]
Caused by: org.opensearch.action.support.replication.ReplicationOperation$RetryOnPrimaryException: shard is not in primary mode

[2025-05-23T11:25:52,690][ERROR][o.o.i.r.SegmentReplicationIT] [[Thread-1]] write doc 14 cost 20156ms generate exception
org.opensearch.action.support.replication.ReplicationOperation$RetryOnPrimaryException: shard is not in primary mode

[2025-05-23T11:25:52,691][INFO ][o.o.i.r.SegmentReplicationIT] [[Thread-1]] begin write doc 15

[2025-05-23T11:25:52,766][INFO ][o.o.i.r.SegmentReplicationIT] [[Thread-1]] after write doc 15, cost 75ms

java.lang.AssertionError: Expected search hits on node: node_t3 to be at least 20 but was: 19

Expected behavior

Like document-replication, no write blocking occurs during primary shard relocation.

[sachinpkale]

Thanks for the RFC @guojialiang92.

We also need to test remote backed storage flow with this change as it is built on top of segment replication. For example, With Internal Engine, refresh will be triggered along with corresponding listeners. One of the listeners is RemoteStoreRefreshListener which uploads the segments to configured remote store. Even though we have added guards to avoid uploading segments before shard is started and primary mode is not started (link), there are some conditions that specifically check for InternalEngine (link).

Also, we are expecting that relocation time (at least the last phase) would be significantly reduce with this approach, right? As the new primary will start creating new segments (that are different from old primary), the number of new segments should be kept in check. Else, once the new primary starts, it would need to upload all these segments and can cause replication lag.

[guojialiang92]

Thanks, @sachinpkale.

We also need to test remote backed storage flow with this change as it is built on top of segment replication. For example, With Internal Engine, refresh will be triggered along with corresponding listeners. One of the listeners is RemoteStoreRefreshListener which uploads the segments to configured remote store. Even though we have added guards to avoid uploading segments before shard is started and primary mode is not started (link), there are some conditions that specifically check for InternalEngine (link).

Yeah, Remote Store should also be considered.
At the same time, I also appreciate your suggestions. I have read the link you provided and would like to confirm whether my understanding is correct.
Through tracking tests RemoteStoreRestoreIT, I found that it is possible for indexShard.state() != IndexShardState.STARTED || !(indexShard.getEngine() instanceof InternalEngine) to return false when recovery source is EMPTY_STORE and REMOTE_STORE.
Therefore, it is necessary to ensure that when the recovery source is PEER, if the primary mode is not true, we should not upload and retry.

Also, we are expecting that relocation time (at least the last phase) would be significantly reduce with this approach, right?

Yes, because there is no need for a round of segment replication, the time will be significantly reduced.

As the new primary will start creating new segments (that are different from old primary), the number of new segments should be kept in check. Else, once the new primary starts, it would need to upload all these segments and can cause replication lag.

I think replica shard and new primary shard will ensure eventual consistency. Reducing the replication lag of replicas may be an orthogonal problem. I think there are two solutions.

1. Only part of the data is cached locally, so the replica does not need to download the full amount of data, greatly reducing lag. Just like writable warm, the company I work for also has a similar implementation.
2. Provides hybrid mode for replicas to fetch data from remote store and primary shards. I have also noticed RFC with similar idea, and perhaps we can collaborate.

[mch2]

This is a great improvement @guojialiang92. I think this makes sense and I like that we won't need the hack in recovery to open the new primary engine as NRT, we could also remove the existing counter bump hacks on NRT engine close.

One thought on the remote case, thinking we could also skip the segment & translog upload listener drain logic here along with the force sync, further reducing the block duration. Sachin, am I off here? The new primary's translog should be kept up to date after engine open (fetch from remote) and normal op based recovery steps where we'd get the translog size for the counter bump directly from the old.

[guojialiang92]

Thank you for your reply, @mch2.

I think this makes sense and I like that we won't need the hack in recovery to open the new primary engine as NRT, we could also remove the existing counter bump hacks on NRT engine close.

Yes, we can remove the existing counter bump hacks on NRT engine close.
When InternalEngine is opened, handle replica promotion and primary shard relocation separately. This can avoid the problem that NRT Engine close cannot be called in some extreme cases (say kill -9).

One thought on the remote case, thinking we could also skip the segment & translog upload listener drain logic here along with the force sync, further reducing the block duration. Sachin, am I off here? The new primary's translog should be kept up to date after engine open (fetch from remote) and normal op based recovery steps where we'd get the translog size for the counter bump directly from the old.

Yes, these additional operations introduced in peer recovery can be removed.

[andrross]

Catch All Triage - 1 2