Renewing retention lease with global-ckp + 1 , as we want operations …

…from that seq number (#206) Signed-off-by: Gaurav Bafna <[email protected]>
opensearch-project · Oct 26, 2021 · 12fe6b5 · 12fe6b5
1 parent a9d9c6f
commit 12fe6b5
Showing 1 changed file with 7 additions and 3 deletions.
diff --git a/src/main/kotlin/org/opensearch/replication/task/shard/ShardReplicationTask.kt b/src/main/kotlin/org/opensearch/replication/task/shard/ShardReplicationTask.kt
@@ -199,7 +199,9 @@ class ShardReplicationTask(id: Long, type: String, action: String, description:
         val indexShard = followerIndexService.getShard(followerShardId.id)
 
         try {
-            retentionLeaseHelper.renewRetentionLease(leaderShardId, indexShard.lastSyncedGlobalCheckpoint, followerShardId)
+            //Retention leases preserve the operations including and starting from the retainingSequenceNumber we specify when we take the lease .
+            //hence renew retention lease with lastSyncedGlobalCheckpoint + 1
+            retentionLeaseHelper.renewRetentionLease(leaderShardId, indexShard.lastSyncedGlobalCheckpoint + 1, followerShardId)
         } catch (ex: Exception) {
             // In case of a failure, we just log it and move on. All failures scenarios are being handled below with
             // retries and backoff depending on exception.
@@ -266,9 +268,11 @@ class ShardReplicationTask(id: Long, type: String, action: String, description:
                     }
                 }
 
-                //renew retention lease with global checkpoint so that any shard that picks up shard replication task has data until then.
+
+                //Retention leases preserve the operations including and starting from the retainingSequenceNumber we specify when we take the lease .
+                //hence renew retention lease with lastSyncedGlobalCheckpoint + 1 so that any shard that picks up shard replication task has data until then.
                 try {
-                    retentionLeaseHelper.renewRetentionLease(leaderShardId, indexShard.lastSyncedGlobalCheckpoint, followerShardId)
+                    retentionLeaseHelper.renewRetentionLease(leaderShardId, indexShard.lastSyncedGlobalCheckpoint + 1, followerShardId)
                     followerClusterStats.stats[followerShardId]!!.followerCheckpoint = indexShard.lastSyncedGlobalCheckpoint
                     lastLeaseRenewalMillis = System.currentTimeMillis()
                 } catch (ex: Exception) {