Merge pull request #20930 from vespa-engine/toregge/use-persistence-async-throttling-feature-flag-to-control-replay-throttling-policy-type-in-proton-config

Use persistence-async-throttling feature flag to control replay_throt…

2 files changed, 33 insertions, 0 deletions

diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
index 602b3c0ed3a..c703584eccc 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
@@ -73,6 +73,7 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
     private final double defaultFeedConcurrency;
     private final boolean forwardIssuesToQrs;
     private final int defaultMaxCompactBuffers;
+    private final ProtonConfig.Replay_throttling_policy.Type.Enum persistenceAsyncThrottling;
 
     /** Whether the nodes of this cluster also hosts a container cluster in a hosted system */
     private final double fractionOfMemoryReserved;
@@ -202,6 +203,14 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
         }
     }
 
+    private static ProtonConfig.Replay_throttling_policy.Type.Enum convertPersistenceAsyncThrottling(String value) {
+        try {
+            return ProtonConfig.Replay_throttling_policy.Type.Enum.valueOf(value);
+        } catch (Throwable t) {
+            return ProtonConfig.Replay_throttling_policy.Type.Enum.UNLIMITED;
+        }
+    }
+
     private ContentSearchCluster(AbstractConfigProducer<?> parent,
                                  String clusterName,
                                  ModelContext.FeatureFlags featureFlags,
@@ -226,6 +235,7 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
         this.defaultFeedConcurrency = featureFlags.feedConcurrency();
         this.forwardIssuesToQrs = featureFlags.forwardIssuesAsErrors();
         this.defaultMaxCompactBuffers = featureFlags.maxCompactBuffers();
+        this.persistenceAsyncThrottling = convertPersistenceAsyncThrottling(featureFlags.persistenceAsyncThrottling());
     }
 
     public void setVisibilityDelay(double delay) {
@@ -444,6 +454,7 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
         builder.indexing.tasklimit(feedTaskLimit);
         builder.feeding.master_task_limit(feedMasterTaskLimit);
         builder.feeding.shared_field_writer_executor(sharedFieldWriterExecutor);
+        builder.replay_throttling_policy.type(persistenceAsyncThrottling);
     }
 
     private boolean isGloballyDistributed(NewDocumentType docType) {
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/content/ContentClusterTest.java b/config-model/src/test/java/com/yahoo/vespa/model/content/ContentClusterTest.java
index 10a2feaba5b..88073d281c5 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/content/ContentClusterTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/content/ContentClusterTest.java
@@ -1045,6 +1045,28 @@ public class ContentClusterTest extends ContentBaseTest {
         assertEquals(7, resolveMaxCompactBuffers(OptionalInt.of(7)));
     }
 
+    private ProtonConfig.Replay_throttling_policy.Type.Enum resolveReplayThrottlePolicyType(Optional<String> throttlerType) {
+        TestProperties testProperties = new TestProperties();
+        if (throttlerType.isPresent()) {
+            testProperties.setPersistenceAsyncThrottling(throttlerType.get());
+        }
+        VespaModel model = createEnd2EndOneNode(testProperties);
+        ContentCluster cc = model.getContentClusters().get("storage");
+        ProtonConfig.Builder protonBuilder = new ProtonConfig.Builder();
+        cc.getSearch().getConfig(protonBuilder);
+        ProtonConfig protonConfig = new ProtonConfig(protonBuilder);
+        assertEquals(1, protonConfig.documentdb().size());
+        return protonConfig.replay_throttling_policy().type();
+    }
+
+    @Test
+    public void replay_throttling_policy_type_controlled_by_properties() {
+        assertEquals(ProtonConfig.Replay_throttling_policy.Type.Enum.UNLIMITED, resolveReplayThrottlePolicyType(Optional.empty()));
+        assertEquals(ProtonConfig.Replay_throttling_policy.Type.Enum.UNLIMITED, resolveReplayThrottlePolicyType(Optional.of("UNLIMITED")));
+        assertEquals(ProtonConfig.Replay_throttling_policy.Type.Enum.UNLIMITED, resolveReplayThrottlePolicyType(Optional.of("INVALID")));
+        assertEquals(ProtonConfig.Replay_throttling_policy.Type.Enum.DYNAMIC, resolveReplayThrottlePolicyType(Optional.of("DYNAMIC")));
+    }
+
     private long resolveMaxTLSSize(Optional<Flavor> flavor) throws Exception {
         TestProperties testProperties = new TestProperties();