[BUG] SegRep replication lag metric reset if same checkpoint is published twice

[mch2]

Describe the bug
Retries within RemoteStoreRefreshListener run outside of the refresh thread. Publishing checkpoints is included in the retry logic, meaning its possible the same checkpoint is published twice. This has no impact on replica shards as they dedupe the checkpoint, but it will invoke onCheckpointPublished within IndexShard on the primary, which starts timers to compute replication lag. If executed twice, this method will attempt to .start() the same timer twice. In tests this throws an assertion error while outside of tests it will reset the timer from the current timestamp. This will skew lag metrics.

[2023-10-19T18:23:41,312][INFO ][o.o.i.s.RemoteStoreRefreshListener] [node_s1] [test-idx-1][6] PUBLISHED CHECKPOINT ReplicationCheckpoint{shardId=[test-idx-1][6], primaryTerm=1, segmentsGen=3, version=7, size=0, codec=Lucene95} THREAD opensearch[node_s1][refresh][T#1]
[2023-10-19T18:23:41,321][INFO ][o.o.i.s.RemoteStoreRefreshListener] [node_s1] [test-idx-1][6] PUBLISHED CHECKPOINT ReplicationCheckpoint{shardId=[test-idx-1][6], primaryTerm=1, segmentsGen=3, version=7, size=0, codec=Lucene95} THREAD opensearch[node_s1][remote_refresh_retry][T#1]
[2023-10-19T18:23:41,321][ERROR][o.o.i.s.RemoteStoreRefreshListener] [node_s2] [test-idx-1][1] Exception in RemoteStoreRefreshListener.afterRefresh()
java.lang.AssertionError: already started
	at org.opensearch.indices.replication.common.ReplicationTimer.start(ReplicationTimer.java:49) ~[classes/:?]
	at org.opensearch.index.seqno.ReplicationTracker.lambda$startReplicationLagTimers$21(ReplicationTracker.java:1274) ~[classes/:?]
	at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183) ~[?:?]
	at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:179) ~[?:?]
	at java.util.HashMap$EntrySpliterator.forEachRemaining(HashMap.java:1858) ~[?:?]
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
	at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150) ~[?:?]
	at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173) ~[?:?]
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:?]
	at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596) ~[?:?]
	at org.opensearch.index.seqno.ReplicationTracker.startReplicationLagTimers(ReplicationTracker.java:1265) ~[classes/:?]
	at org.opensearch.index.shard.IndexShard.onCheckpointPublished(IndexShard.java:1910) ~[classes/:?]
	at org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher.publish(SegmentReplicationCheckpointPublisher.java:37) ~[classes/:?]
	at org.opensearch.index.shard.RemoteStoreRefreshListener.onSuccessfulSegmentsSync(RemoteStoreRefreshListener.java:299) ~[classes/:?]
	at org.opensearch.index.shard.RemoteStoreRefreshListener$1.onResponse(RemoteStoreRefreshListener.java:219) ~[classes/:?]
	at org.opensearch.index.shard.RemoteStoreRefreshListener$1.onResponse(RemoteStoreRefreshListener.java:210) ~[classes/:?]
	at org.opensearch.action.LatchedActionListener.onResponse(LatchedActionListener.java:58) ~[classes/:?]
	at org.opensearch.index.shard.RemoteStoreRefreshListener.uploadNewSegments(RemoteStoreRefreshListener.java:356) ~[classes/:?]
	at org.opensearch.index.shard.RemoteStoreRefreshListener.syncSegments(RemoteStoreRefreshListener.java:244) ~[classes/:?]
	at org.opensearch.index.shard.RemoteStoreRefreshListener.performAfterRefreshWithPermit(RemoteStoreRefreshListener.java:155) ~[classes/:?]
	at org.opensearch.index.shard.CloseableRetryableRefreshListener.runAfterRefreshWithPermit(CloseableRetryableRefreshListener.java:157) ~[classes/:?]
	at org.opensearch.index.shard.CloseableRetryableRefreshListener.lambda$scheduleRetry$2(CloseableRetryableRefreshListener.java:123) ~[classes/:?]
	at org.opensearch.common.util.concurrent.ThreadContext$ContextPreservingRunnable.run(ThreadContext.java:849) [classes/:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) [?:?]
	at java.lang.Thread.run(Thread.java:1589) [?:?

To Reproduce
I have been running SegmentReplicationUsingRemoteStoreITs on repeat, and can see this in logs from various test cases, while the test does not fail the timer would be reset.

Expected behavior
The checkpoint timer should not be reset if already started.

[mch2]

@ashking94 Curious if you know why this is getting retried? The publish step would be after upload completes.

[ashking94]

@ashking94 Curious if you know why this is getting retried? The publish step would be after upload completes.

Looks like refresh listener is getting invoked with didRefresh=true, but the readers have not really changed. Basically there are no uploads to be done. Another reason that I can think of is - the original refresh might have failed leading to retry. During one of the retries which would have succeeded, a refresh would have occurred. But the retry might have uploaded the file and initiated the timer for the latest checkpoint. As you know, the sync segments is synchronised, we make sure that the syncSegments still run. So, for the sync segments which got triggered on account of 2nd refresh, it could have tried to upload, but did not find any new files to upload. We still trigger the onSuccessfulSync hook and that's why this timer assertion failed? Just thinking aloud.

[guojialiang92]

Hi, @ashking94 @mch2
I also found this problem in test SegmentReplicationUsingRemoteStoreIT, and then I wrote a very simple case, which will appear about once every 10 runs.

public void testSimple() {
    final String primaryNode = internalCluster().startDataOnlyNode();
    createIndex(INDEX_NAME, Settings.builder().put(indexSettings()).put("index.refresh_interval", -1).build());
    ensureYellowAndNoInitializingShards(INDEX_NAME);
    final String replicaNode = internalCluster().startDataOnlyNode();
    ensureGreen(INDEX_NAME);

    client().prepareIndex(INDEX_NAME).setId("1").setSource("foo", "bar").setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
    flush(INDEX_NAME);
}

I tracked this case, and the execution order is:

1. Node startup, execute IndexShard#flush in IndexShard#updateShardState. Because didRefresh = true, execute RemoteStoreRefreshListener#syncSegments.
2. When primary shard is restored from remote store, the recovery happens first followed by changing primaryMode to true. RemoteStoreRefreshListener#isReadyForUpload returns false and prints the log Skipped syncing segments with primaryMode=false indexShardState=STARTED engineType=InternalEngine recoverySourceType=EMPTY_STORE primary=true.
3. Because indexShard.state() != IndexShardState.STARTED || !(indexShard.getEngine() instanceof InternalEngine) return true, execute ReleasableRetryableRefreshListener#scheduleRetry. Rescheduling is executed by default after 1s.
4. During the waiting period for scheduling, refresh and flush operations occurred. The index file is uploaded and the timer is started.
5. When rescheduling is performed, a java.lang.AssertionError: already started will be thrown.

Of course, there may be other cases that can lead to similar problems. To avoid resetting the timer causing lag metrics to skew, can we make strict judgments cps.checkpointTimers.get(latestReplicationCheckpoint).startTime() == 0 in ReplicationTracker#startReplicationLagTimers to ensure that timing only starts on the first startup?

At the same time, I also submitted a relevant PR 17831.

Looking forward to your reply.

[mch2]

@guojialiang92 Thanks for picking this up, that makes sense to me as a precaution.

We introduced with reader writer separation and polling based search replicas computation of lag metrics at the shard level. We haven't yet but I'm thinking it would be beneficial for us to simplify all lag computations to use that rather than this complex tracking at primaries.