[Backport] Support for translog pruning based on retention leases

server/src/internalClusterTest/java/org/opensearch/indices/settings/UpdateSettingsIT.java

@@ -661,6 +661,33 @@ public void testNumberOfReplicasSettingsVersion() {
         assertThat(newSettingsVersion, equalTo(1 + settingsVersion));
     }
 
+    public void testTranslogPruningSetting() {
+        createIndex("test");
+        ensureGreen("test");
+        final String REPLICATION_TRANSLOG_SETTING = "index.plugins.replication.translog.retention_lease.pruning.enabled";
+        final long settingsVersion =
+            client().admin().cluster().prepareState().get().getState().metadata().index("test").getSettingsVersion();
+        GetSettingsResponse settingsResponse = client().admin().indices().prepareGetSettings("test").get();
+        boolean shouldPruneTranslogByRetentionLease = Boolean.parseBoolean(
+            settingsResponse.getSetting("test", REPLICATION_TRANSLOG_SETTING)
+        );
+
+        assertFalse(shouldPruneTranslogByRetentionLease);
+        assertAcked(client().admin().indices()
+            .prepareUpdateSettings("test")
+            .setSettings(Settings.builder()
+                .put("index.plugins.replication.translog.retention_lease.pruning.enabled", true)
+            ));
+        final long newSettingsVersion =
+            client().admin().cluster().prepareState().get().getState().metadata().index("test").getSettingsVersion();
+        assertThat(newSettingsVersion, equalTo(1 + settingsVersion));
+        settingsResponse = client().admin().indices().prepareGetSettings("test").get();
+        shouldPruneTranslogByRetentionLease = Boolean.parseBoolean(
+            settingsResponse.getSetting("test", REPLICATION_TRANSLOG_SETTING)
+        );
+        assertTrue(shouldPruneTranslogByRetentionLease);
+    }
+
     /*
      * Test that we are able to set the setting index.number_of_replicas to the default.
      */

server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java

@@ -1258,11 +1258,19 @@ public static int calculateNumRoutingShards(int numShards, Version indexVersionC
     }
 
     public static void validateTranslogRetentionSettings(Settings indexSettings) {
-        if (IndexSettings.INDEX_SOFT_DELETES_SETTING.get(indexSettings) &&
-            (IndexSettings.INDEX_TRANSLOG_RETENTION_AGE_SETTING.exists(indexSettings)
-                || IndexSettings.INDEX_TRANSLOG_RETENTION_SIZE_SETTING.exists(indexSettings))) {
-            DEPRECATION_LOGGER.deprecate("translog_retention", "Translog retention settings [index.translog.retention.age] "
-                + "and [index.translog.retention.size] are deprecated and effectively ignored. They will be removed in a future version.");
+        if (IndexSettings.INDEX_SOFT_DELETES_SETTING.get(indexSettings)) {
+            if(IndexSettings.INDEX_TRANSLOG_RETENTION_AGE_SETTING.exists(indexSettings)
+                || IndexSettings.INDEX_TRANSLOG_RETENTION_SIZE_SETTING.exists(indexSettings)) {
+                DEPRECATION_LOGGER.deprecate("translog_retention", "Translog retention settings " +
+                    "[index.translog.retention.age] "
+                    + "and [index.translog.retention.size] are deprecated and effectively ignored. " +
+                    "They will be removed in a future version.");
+            } else if(IndexSettings.INDEX_PLUGINS_REPLICATION_TRANSLOG_RETENTION_LEASE_PRUNING_ENABLED_SETTING.exists(indexSettings)) {
+                DEPRECATION_LOGGER.deprecate("translog_pruning_retention_lease",
+                    "[index.plugins.replication.translog.retention_lease.pruning.enabled] " +
+                        "setting was deprecated in OpenSearch and will be removed in a future release! " +
+                        "See the breaking changes documentation for the next major version.");
+            }
         }
     }
 }

server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java

@@ -156,6 +156,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         EnableAllocationDecider.INDEX_ROUTING_REBALANCE_ENABLE_SETTING,
         EnableAllocationDecider.INDEX_ROUTING_ALLOCATION_ENABLE_SETTING,
         IndexSettings.INDEX_FLUSH_AFTER_MERGE_THRESHOLD_SIZE_SETTING,
+        IndexSettings.INDEX_PLUGINS_REPLICATION_TRANSLOG_RETENTION_LEASE_PRUNING_ENABLED_SETTING,
         IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING,
         IndexSettings.INDEX_TRANSLOG_GENERATION_THRESHOLD_SIZE_SETTING,
         IndexSettings.INDEX_TRANSLOG_RETENTION_AGE_SETTING,

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

@@ -260,6 +260,15 @@ public final class IndexSettings {
         settings -> Boolean.toString(IndexMetadata.SETTING_INDEX_VERSION_CREATED.get(settings).onOrAfter(LegacyESVersion.V_7_0_0)),
         Property.IndexScope, Property.Final);
 
+    /**
+     * Specifies if the index translog should prune based on retention leases.
+     * @deprecated EXPERT: this setting is specific to CCR and will be moved to a plugin in the next release
+     */
+    @Deprecated
+    public static final Setting<Boolean> INDEX_PLUGINS_REPLICATION_TRANSLOG_RETENTION_LEASE_PRUNING_ENABLED_SETTING =
+        Setting.boolSetting("index.plugins.replication.translog.retention_lease.pruning.enabled", false,
+            Property.IndexScope, Property.Dynamic);
+
     /**
      * Controls how many soft-deleted documents will be kept around before being merged away. Keeping more deleted
      * documents increases the chance of operation-based recoveries and allows querying a longer history of documents.
@@ -286,9 +295,12 @@ public final class IndexSettings {
      * the chance of ops based recoveries for indices with soft-deletes disabled.
      * This setting will be ignored if soft-deletes is used in peer recoveries (default in 7.4).
      **/
+    @Deprecated
+    private static final ByteSizeValue DEFAULT_TRANSLOG_RETENTION_SIZE = new ByteSizeValue(512, ByteSizeUnit.MB);
+
     public static final Setting<ByteSizeValue> INDEX_TRANSLOG_RETENTION_SIZE_SETTING =
         Setting.byteSizeSetting("index.translog.retention.size",
-            settings -> shouldDisableTranslogRetention(settings) ? "-1" : "512MB",
+            settings -> DEFAULT_TRANSLOG_RETENTION_SIZE.getStringRep(),
             Property.Dynamic, Property.IndexScope);
 
     /**
@@ -389,6 +401,8 @@ public final class IndexSettings {
     private final IndexScopedSettings scopedSettings;
     private long gcDeletesInMillis = DEFAULT_GC_DELETES.millis();
     private final boolean softDeleteEnabled;
+    @Deprecated
+    private volatile boolean translogPruningByRetentionLease;
     private volatile long softDeleteRetentionOperations;
 
     private volatile long retentionLeaseMillis;
@@ -525,6 +539,10 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         mergeSchedulerConfig = new MergeSchedulerConfig(this);
         gcDeletesInMillis = scopedSettings.get(INDEX_GC_DELETES_SETTING).getMillis();
         softDeleteEnabled = version.onOrAfter(LegacyESVersion.V_6_5_0) && scopedSettings.get(INDEX_SOFT_DELETES_SETTING);
+        // @todo move to CCR plugin
+        this.translogPruningByRetentionLease = version.onOrAfter(Version.V_1_1_0) &&
+            softDeleteEnabled &&
+            scopedSettings.get(INDEX_PLUGINS_REPLICATION_TRANSLOG_RETENTION_LEASE_PRUNING_ENABLED_SETTING);
         softDeleteRetentionOperations = scopedSettings.get(INDEX_SOFT_DELETES_RETENTION_OPERATIONS_SETTING);
         retentionLeaseMillis = scopedSettings.get(INDEX_SOFT_DELETES_RETENTION_LEASE_PERIOD_SETTING).millis();
         warmerEnabled = scopedSettings.get(INDEX_WARMER_ENABLED_SETTING);
@@ -593,6 +611,8 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
                 this::setGenerationThresholdSize);
         scopedSettings.addSettingsUpdateConsumer(INDEX_TRANSLOG_RETENTION_AGE_SETTING, this::setTranslogRetentionAge);
         scopedSettings.addSettingsUpdateConsumer(INDEX_TRANSLOG_RETENTION_SIZE_SETTING, this::setTranslogRetentionSize);
+        scopedSettings.addSettingsUpdateConsumer(INDEX_PLUGINS_REPLICATION_TRANSLOG_RETENTION_LEASE_PRUNING_ENABLED_SETTING,
+            this::setTranslogPruningByRetentionLease);
         scopedSettings.addSettingsUpdateConsumer(INDEX_REFRESH_INTERVAL_SETTING, this::setRefreshInterval);
         scopedSettings.addSettingsUpdateConsumer(MAX_REFRESH_LISTENERS_PER_SHARD, this::setMaxRefreshListeners);
         scopedSettings.addSettingsUpdateConsumer(MAX_ANALYZED_OFFSET_SETTING, this::setHighlightMaxAnalyzedOffset);
@@ -623,8 +643,20 @@ private void setFlushAfterMergeThresholdSize(ByteSizeValue byteSizeValue) {
         this.flushAfterMergeThresholdSize = byteSizeValue;
     }
 
+    /**
+     * Enable or disable translog pruning by retention lease requirement
+     *
+     * @deprecated EXPERT: this setting is specific to CCR and will be moved to a plugin in the next release
+     */
+    @Deprecated
+    private void setTranslogPruningByRetentionLease(boolean enabled) {
+        this.translogPruningByRetentionLease = INDEX_SOFT_DELETES_SETTING.get(settings) && enabled;
+    }
+
     private void setTranslogRetentionSize(ByteSizeValue byteSizeValue) {
-        if (shouldDisableTranslogRetention(settings) && byteSizeValue.getBytes() >= 0) {
+        if (shouldDisableTranslogRetention(settings) &&
+            shouldPruneTranslogByRetentionLease(settings) == false &&
+            byteSizeValue.getBytes() >= 0) {
             // ignore the translog retention settings if soft-deletes enabled
             this.translogRetentionSize = new ByteSizeValue(-1);
         } else {
@@ -826,7 +858,12 @@ public TimeValue getRefreshInterval() {
      * Returns the transaction log retention size which controls how much of the translog is kept around to allow for ops based recoveries
      */
     public ByteSizeValue getTranslogRetentionSize() {
-        assert shouldDisableTranslogRetention(settings) == false || translogRetentionSize.getBytes() == -1L : translogRetentionSize;
+        if(shouldDisableTranslogRetention(settings) && shouldPruneTranslogByRetentionLease(settings) == false) {
+            return new ByteSizeValue(-1);
+        }
+        else if(shouldPruneTranslogByRetentionLease(settings) && translogRetentionSize.getBytes() == -1) {
+            return DEFAULT_TRANSLOG_RETENTION_SIZE;
+        }
         return translogRetentionSize;
     }
 
@@ -1071,6 +1108,24 @@ public void setRequiredPipeline(final String requiredPipeline) {
         this.requiredPipeline = requiredPipeline;
     }
 
+    /**
+     * Returns <code>true</code> if translog ops should be pruned based on retention lease
+     * @deprecated EXPERT: this setting is specific to CCR and will be moved to a plugin in the next release
+     */
+    @Deprecated
+    public boolean shouldPruneTranslogByRetentionLease() {
+        return translogPruningByRetentionLease;
+    }
+
+    /**
+     * Returns <code>true</code> if translog ops should be pruned based on retention lease
+     * @deprecated EXPERT: this setting is specific to CCR and will be moved to a plugin in the next release
+     */
+    @Deprecated
+    public static boolean shouldPruneTranslogByRetentionLease(Settings settings) {
+        return INDEX_PLUGINS_REPLICATION_TRANSLOG_RETENTION_LEASE_PRUNING_ENABLED_SETTING.get(settings);
+    }
+
     /**
      * Returns <code>true</code> if soft-delete is enabled.
      */

server/src/main/java/org/opensearch/index/engine/Engine.java

@@ -1858,7 +1858,18 @@ public IndexCommit getIndexCommit() {
     }
 
     public void onSettingsChanged(TimeValue translogRetentionAge, ByteSizeValue translogRetentionSize, long softDeletesRetentionOps) {
+        onSettingsChanged(translogRetentionAge, translogRetentionSize, softDeletesRetentionOps, false);
+    }
 
+    /**
+     * callback when index settings change
+     *
+     * @deprecated EXPERT: this method is specific to CCR and will be moved to a plugin in the next release
+     */
+    @Deprecated
+    public void onSettingsChanged(TimeValue translogRetentionAge, ByteSizeValue translogRetentionSize,
+                                  long softDeletesRetentionOps, boolean translogPruningByRetentionLease) {
+        // @todo this is overly silent; make this abstract and force derived classes to noop in the next release
     }
 
     /**

server/src/main/java/org/opensearch/index/engine/InternalEngine.java

@@ -227,7 +227,8 @@ public InternalEngine(EngineConfig engineConfig) {
         final TranslogDeletionPolicy translogDeletionPolicy = new TranslogDeletionPolicy(
             engineConfig.getIndexSettings().getTranslogRetentionSize().getBytes(),
             engineConfig.getIndexSettings().getTranslogRetentionAge().getMillis(),
-            engineConfig.getIndexSettings().getTranslogRetentionTotalFiles()
+            engineConfig.getIndexSettings().getTranslogRetentionTotalFiles(),
+            engineConfig.retentionLeasesSupplier()
         );
         store.incRef();
         IndexWriter writer = null;
@@ -2572,7 +2573,8 @@ final void ensureCanFlush() {
     }
 
     @Override
-    public void onSettingsChanged(TimeValue translogRetentionAge, ByteSizeValue translogRetentionSize, long softDeletesRetentionOps) {
+    public void onSettingsChanged(TimeValue translogRetentionAge, ByteSizeValue translogRetentionSize,
+                                  long softDeletesRetentionOps, boolean translogPruningByRetentionLease) {
         mergeScheduler.refreshConfig();
         // config().isEnableGcDeletes() or config.getGcDeletesInMillis() may have changed:
         maybePruneDeletes();
@@ -2585,6 +2587,7 @@ public void onSettingsChanged(TimeValue translogRetentionAge, ByteSizeValue tran
         final TranslogDeletionPolicy translogDeletionPolicy = translog.getDeletionPolicy();
         translogDeletionPolicy.setRetentionAgeInMillis(translogRetentionAge.millis());
         translogDeletionPolicy.setRetentionSizeInBytes(translogRetentionSize.getBytes());
+        translogDeletionPolicy.shouldPruneTranslogByRetentionLease(translogPruningByRetentionLease);
         softDeletesPolicy.setRetentionOperations(softDeletesRetentionOps);
     }
 

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

@@ -1961,8 +1961,10 @@ public void onSettingsChanged() {
             final boolean disableTranslogRetention = indexSettings.isSoftDeleteEnabled() && useRetentionLeasesInPeerRecovery;
             engineOrNull.onSettingsChanged(
                 disableTranslogRetention ? TimeValue.MINUS_ONE : indexSettings.getTranslogRetentionAge(),
-                disableTranslogRetention ? new ByteSizeValue(-1) : indexSettings.getTranslogRetentionSize(),
-                indexSettings.getSoftDeleteRetentionOperations()
+                disableTranslogRetention && !indexSettings.shouldPruneTranslogByRetentionLease() ?
+                    new ByteSizeValue(-1) : indexSettings.getTranslogRetentionSize(),
+                indexSettings.getSoftDeleteRetentionOperations(),
+                indexSettings.shouldPruneTranslogByRetentionLease()
             );
         }
     }