Add feature flag for controlling max feed operation batch size

Currently defaults to 1, which disables batching.

7 files changed, 42 insertions, 1 deletions

diff --git a/config-model-api/abi-spec.json b/config-model-api/abi-spec.json
index 9a975f1b727..3fefd05bd55 100644
--- a/config-model-api/abi-spec.json
+++ b/config-model-api/abi-spec.json
@@ -1324,7 +1324,8 @@
       "public java.lang.String unknownConfigDefinition()",
       "public int searchHandlerThreadpool()",
       "public boolean alwaysMarkPhraseExpensive()",
-      "public boolean sortBlueprintsByCost()"
+      "public boolean sortBlueprintsByCost()",
+      "public int persistenceThreadMaxFeedOpBatchSize()"
     ],
     "fields" : [ ]
   },
diff --git a/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java b/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java
index 4a4d4648deb..de71c700741 100644
--- a/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java
+++ b/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java
@@ -112,6 +112,7 @@ public interface ModelContext {
         @ModelFeatureFlag(owners = {"hmusum"}) default int searchHandlerThreadpool() { return 2; }
         @ModelFeatureFlag(owners = {"baldersheim"}) default boolean alwaysMarkPhraseExpensive() { return false; }
         @ModelFeatureFlag(owners = {"baldersheim"}) default boolean sortBlueprintsByCost() { return false; }
+        @ModelFeatureFlag(owners = {"vekterli"}) default int persistenceThreadMaxFeedOpBatchSize() { return 1; }
     }
 
     /** Warning: As elsewhere in this package, do not make backwards incompatible changes that will break old config models! */
diff --git a/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java b/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java
index 3e91b8afe88..51c78ee796c 100644
--- a/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java
+++ b/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java
@@ -82,6 +82,7 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     private boolean allowUserFilters = true;
     private List<DataplaneToken> dataplaneTokens;
     private int contentLayerMetadataFeatureLevel = 0;
+    private int persistenceThreadMaxFeedOpBatchSize = 1;
 
     @Override public ModelContext.FeatureFlags featureFlags() { return this; }
     @Override public boolean multitenant() { return multitenant; }
@@ -138,6 +139,7 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     @Override public boolean allowUserFilters() { return allowUserFilters; }
     @Override public List<DataplaneToken> dataplaneTokens() { return dataplaneTokens; }
     @Override public int contentLayerMetadataFeatureLevel() { return contentLayerMetadataFeatureLevel; }
+    @Override public int persistenceThreadMaxFeedOpBatchSize() { return persistenceThreadMaxFeedOpBatchSize; }
 
     public TestProperties sharedStringRepoNoReclaim(boolean sharedStringRepoNoReclaim) {
         this.sharedStringRepoNoReclaim = sharedStringRepoNoReclaim;
@@ -368,6 +370,11 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
         return this;
     }
 
+    public TestProperties setPersistenceThreadMaxFeedOpBatchSize(int maxBatchSize) {
+        this.persistenceThreadMaxFeedOpBatchSize = maxBatchSize;
+        return this;
+    }
+
     public static class Spec implements ConfigServerSpec {
 
         private final String hostName;
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/storagecluster/FileStorProducer.java b/config-model/src/main/java/com/yahoo/vespa/model/content/storagecluster/FileStorProducer.java
index 6fd917c393d..ae84978793a 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/storagecluster/FileStorProducer.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/storagecluster/FileStorProducer.java
@@ -47,6 +47,7 @@ public class FileStorProducer implements StorFilestorConfig.Producer {
     private final int responseNumThreads;
     private final StorFilestorConfig.Response_sequencer_type.Enum responseSequencerType;
     private final boolean useAsyncMessageHandlingOnSchedule;
+    private final int persistenceThreadMaxFeedOpBatchSize;
 
     private static StorFilestorConfig.Response_sequencer_type.Enum convertResponseSequencerType(String sequencerType) {
         try {
@@ -62,6 +63,7 @@ public class FileStorProducer implements StorFilestorConfig.Producer {
         this.responseNumThreads = featureFlags.defaultNumResponseThreads();
         this.responseSequencerType = convertResponseSequencerType(featureFlags.responseSequencerType());
         this.useAsyncMessageHandlingOnSchedule = featureFlags.useAsyncMessageHandlingOnSchedule();
+        this.persistenceThreadMaxFeedOpBatchSize = featureFlags.persistenceThreadMaxFeedOpBatchSize();
     }
 
 
@@ -75,6 +77,7 @@ public class FileStorProducer implements StorFilestorConfig.Producer {
         builder.use_async_message_handling_on_schedule(useAsyncMessageHandlingOnSchedule);
         var throttleBuilder = new StorFilestorConfig.Async_operation_throttler.Builder();
         builder.async_operation_throttler(throttleBuilder);
+        builder.max_feed_op_batch_size(persistenceThreadMaxFeedOpBatchSize);
     }
 
 }
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java b/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java
index 0dd03d258b9..f56e642edd7 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java
@@ -336,6 +336,23 @@ public class StorageClusterTest {
         assertEquals(3.0, config.async_operation_throttler().resize_rate(), 0.0001);
     }
 
+    private void verifyMaxFeedOpBatchSize(int expected, Integer flagValue) {
+        var props = new TestProperties();
+        if (flagValue != null) {
+            props.setPersistenceThreadMaxFeedOpBatchSize(flagValue);
+        }
+        var config = filestorConfigFromProducer(simpleCluster(props));
+        assertEquals(expected, config.max_feed_op_batch_size());
+    }
+
+    @Test
+    void persistence_max_feed_op_batch_size_is_controlled_by_feature_flag() {
+        // TODO update default once rolled out and tested
+        verifyMaxFeedOpBatchSize(1, null);
+        verifyMaxFeedOpBatchSize(1, 1);
+        verifyMaxFeedOpBatchSize(1234, 1234);
+    }
+
     @Test
     void testCapacity() {
         String xml = joinLines(
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java b/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java
index 26732d2e20f..2750279919d 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java
@@ -206,6 +206,7 @@ public class ModelContextImpl implements ModelContext {
         private final int contentLayerMetadataFeatureLevel;
         private final String unknownConfigDefinition;
         private final int searchHandlerThreadpool;
+        private final int persistenceThreadMaxFeedOpBatchSize;
 
         public FeatureFlags(FlagSource source, ApplicationId appId, Version version) {
             this.defaultTermwiseLimit = flagValue(source, appId, version, Flags.DEFAULT_TERM_WISE_LIMIT);
@@ -248,6 +249,7 @@ public class ModelContextImpl implements ModelContext {
             this.searchHandlerThreadpool = flagValue(source, appId, version, Flags.SEARCH_HANDLER_THREADPOOL);
             this.alwaysMarkPhraseExpensive =  flagValue(source, appId, version, Flags.ALWAYS_MARK_PHRASE_EXPENSIVE);
             this.sortBlueprintsByCost = flagValue(source, appId, version, Flags.SORT_BLUEPRINTS_BY_COST);
+            this.persistenceThreadMaxFeedOpBatchSize = flagValue(source, appId, version, Flags.PERSISTENCE_THREAD_MAX_FEED_OP_BATCH_SIZE);
         }
 
         @Override public int heapSizePercentage() { return heapPercentage; }
@@ -298,6 +300,7 @@ public class ModelContextImpl implements ModelContext {
         @Override public String unknownConfigDefinition() { return unknownConfigDefinition; }
         @Override public int searchHandlerThreadpool() { return searchHandlerThreadpool; }
         @Override public boolean sortBlueprintsByCost() { return sortBlueprintsByCost; }
+        @Override public int persistenceThreadMaxFeedOpBatchSize() { return persistenceThreadMaxFeedOpBatchSize; }
 
         private static <V> V flagValue(FlagSource source, ApplicationId appId, Version vespaVersion, UnboundFlag<? extends V, ?, ?> flag) {
             return flag.bindTo(source)
diff --git a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
index 1c043ad0aa6..befd66c83b0 100644
--- a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
+++ b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
@@ -428,6 +428,15 @@ public class Flags {
             "Whether to read and write disk encryption key to new path",
             "Will be read only on boot.");
 
+    public static final UnboundIntFlag PERSISTENCE_THREAD_MAX_FEED_OP_BATCH_SIZE = defineIntFlag(
+            "persistence-thread-max-feed-op-batch-size", 1,
+            List.of("vekterli"), "2024-04-12", "2025-01-01",
+            "Maximum number of enqueued feed operations (put/update/remove) bound "+
+            "towards the same bucket that can be async dispatched as part of the " +
+            "same write-locked batch by a persistence thread.",
+            "Takes effect at redeployment",
+            INSTANCE_ID);
+
     /** WARNING: public for testing: All flags should be defined in {@link Flags}. */
     public static UnboundBooleanFlag defineFeatureFlag(String flagId, boolean defaultValue, List<String> owners,
                                                        String createdAt, String expiresAt, String description,