Add feature flag for the task limit used by the executors handling feed in proton.

6 files changed, 49 insertions, 17 deletions

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 6da336ae20c..ec56f720af7 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
@@ -72,6 +72,7 @@ public interface ModelContext {
         @ModelFeatureFlag(owners = {"baldersheim"}, comment = "Revisit in May or June 2021") default double defaultTermwiseLimit() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"vekterli"}) default boolean useThreePhaseUpdates() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}, comment = "Select sequencer type use while feeding") default String feedSequencerType() { throw new UnsupportedOperationException("TODO specify default value"); }
+        @ModelFeatureFlag(owners = {"geirst, baldersheim"}) default int feedTaskLimit() { return 1000; }
         @ModelFeatureFlag(owners = {"baldersheim"}) default String responseSequencerType() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default int defaultNumResponseThreads() { return 2; }
         @ModelFeatureFlag(owners = {"baldersheim"}) default boolean skipCommunicationManagerThread() { throw new UnsupportedOperationException("TODO specify default value"); }
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 ce4e5f0c01f..a63344678da 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
@@ -41,6 +41,7 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     private double defaultTermwiseLimit = 1.0;
     private String jvmGCOptions = null;
     private String sequencerType = "LATENCY";
+    private int feedTaskLimit = 1000;
     private boolean firstTimeDeployment = false;
     private String responseSequencerType = "ADAPTIVE";
     private int responseNumThreads = 2;
@@ -80,6 +81,7 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     @Override public Set<ContainerEndpoint> endpoints() { return endpoints; }
     @Override public String jvmGCOptions(Optional<ClusterSpec.Type> clusterType) { return jvmGCOptions; }
     @Override public String feedSequencerType() { return sequencerType; }
+    @Override public int feedTaskLimit() { return feedTaskLimit; }
     @Override public boolean isBootstrap() { return false; }
     @Override public boolean isFirstTimeDeployment() { return firstTimeDeployment; }
     @Override public boolean useDedicatedNodeForLogserver() { return useDedicatedNodeForLogserver; }
@@ -155,6 +157,10 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
         sequencerType = type;
         return this;
     }
+    public TestProperties setFeedTaskLimit(int value) {
+        feedTaskLimit = value;
+        return this;
+    }
     public TestProperties setResponseSequencerType(String type) {
         responseSequencerType = type;
         return this;
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 3ab7e947a2f..1200b7467b4 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
@@ -64,6 +64,7 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
     private final Map<StorageGroup, NodeSpec> groupToSpecMap = new LinkedHashMap<>();
     private Optional<ResourceLimits> resourceLimits = Optional.empty();
     private final ProtonConfig.Indexing.Optimize.Enum feedSequencerType;
+    private final int feedTaskLimit;
     private final double defaultFeedConcurrency;
     private final double defaultDiskBloatFactor;
     private final int defaultDocStoreCompressionLevel;
@@ -210,6 +211,7 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
 
         this.combined = combined;
         feedSequencerType = convertFeedSequencerType(featureFlags.feedSequencerType());
+        feedTaskLimit = featureFlags.feedTaskLimit();
         defaultFeedConcurrency = featureFlags.feedConcurrency();
         defaultDocStoreCompressionLevel = featureFlags.docstoreCompressionLevel();
         defaultDiskBloatFactor = featureFlags.diskBloatFactor();
@@ -429,6 +431,7 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
         } else {
             builder.indexing.optimize(feedSequencerType);
         }
+        builder.indexing.tasklimit(feedTaskLimit);
     }
 
     private boolean isGloballyDistributed(NewDocumentType docType) {
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/builder/xml/dom/ContentBuilderTest.java b/config-model/src/test/java/com/yahoo/vespa/model/builder/xml/dom/ContentBuilderTest.java
index 891a870dc01..5395b0868cb 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/builder/xml/dom/ContentBuilderTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/builder/xml/dom/ContentBuilderTest.java
@@ -777,21 +777,24 @@ public class ContentBuilderTest extends DomBuilderTest {
                 "</services>";
     }
 
+    private ProtonConfig resolveProtonConfig(TestProperties props, String hostedXml) {
+        var deployStateBuilder = new DeployState.Builder().properties(props);
+        var model = new VespaModelCreatorWithMockPkg(new MockApplicationPackage.Builder()
+                .withServices(hostedXml)
+                .withSearchDefinition(MockApplicationPackage.MUSIC_SEARCHDEFINITION)
+                .build())
+                .create(deployStateBuilder);
+        return getProtonConfig(model.getContentClusters().values().iterator().next());
+    }
+
     private void verifyFeedSequencer(String input, String expected) {
         verifyFeedSequencer(input, expected, 0);
     }
+
     private void verifyFeedSequencer(String input, String expected, double visibilityDelay) {
         String hostedXml = xmlWithVisibilityDelay(visibilityDelay);
-
-        DeployState.Builder deployStateBuilder = new DeployState.Builder().properties(new TestProperties().setFeedSequencerType(input));
-        VespaModel model = new VespaModelCreatorWithMockPkg(new MockApplicationPackage.Builder()
-                .withServices(hostedXml)
-                .withSearchDefinition(MockApplicationPackage.MUSIC_SEARCHDEFINITION)
-                .build())
-                .create(deployStateBuilder);
-        ProtonConfig config = getProtonConfig(model.getContentClusters().values().iterator().next());
+        var config = resolveProtonConfig(new TestProperties().setFeedSequencerType(input), hostedXml);
         assertEquals(expected, config.indexing().optimize().toString());
-
     }
 
     @Test
@@ -806,17 +809,26 @@ public class ContentBuilderTest extends DomBuilderTest {
 
     }
 
+    @Test
+    public void feed_task_limit_is_controlled_by_feature_flag() {
+        assertEquals(1000, resolveFeedTaskLimitConfigWithFeatureFlag(null));
+        assertEquals(2000, resolveFeedTaskLimitConfigWithFeatureFlag(2000));
+    }
+
+    private int resolveFeedTaskLimitConfigWithFeatureFlag(Integer value) {
+        var props = new TestProperties();
+        if (value != null) {
+            props.setFeedTaskLimit(value);
+        }
+        return resolveProtonConfig(props, singleNodeContentXml()).indexing().tasklimit();
+    }
+
     private void verifyThatFeatureFlagControlsVisibilityDelayDefault(Double xmlOverride, double expected) {
         String hostedXml = xmlWithVisibilityDelay(xmlOverride);
-        DeployState.Builder deployStateBuilder = new DeployState.Builder().properties(new TestProperties());
-        VespaModel model = new VespaModelCreatorWithMockPkg(new MockApplicationPackage.Builder()
-                .withServices(hostedXml)
-                .withSearchDefinition(MockApplicationPackage.MUSIC_SEARCHDEFINITION)
-                .build())
-                .create(deployStateBuilder);
-        ProtonConfig config = getProtonConfig(model.getContentClusters().values().iterator().next());
+        var config = resolveProtonConfig(new TestProperties(), hostedXml);
         assertEquals(expected, config.documentdb(0).visibilitydelay(), 0.0);
     }
+
     @Test
     public void verifyThatFeatureFlagControlsVisibilityDelayDefault() {
         verifyThatFeatureFlagControlsVisibilityDelayDefault(null, 0.0);
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 b25bbf1a5bb..7d09e033dc8 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
@@ -167,6 +167,7 @@ public class ModelContextImpl implements ModelContext {
         private final double defaultTermwiseLimit;
         private final boolean useThreePhaseUpdates;
         private final String feedSequencer;
+        private final int feedTaskLimit;
         private final String responseSequencer;
         private final int numResponseThreads;
         private final boolean skipCommunicationManagerThread;
@@ -199,6 +200,7 @@ public class ModelContextImpl implements ModelContext {
             this.defaultTermwiseLimit = flagValue(source, appId, Flags.DEFAULT_TERM_WISE_LIMIT);
             this.useThreePhaseUpdates = flagValue(source, appId, Flags.USE_THREE_PHASE_UPDATES);
             this.feedSequencer = flagValue(source, appId, Flags.FEED_SEQUENCER_TYPE);
+            this.feedTaskLimit = flagValue(source, appId, Flags.FEED_TASK_LIMIT);
             this.responseSequencer = flagValue(source, appId, Flags.RESPONSE_SEQUENCER_TYPE);
             this.numResponseThreads = flagValue(source, appId, Flags.RESPONSE_NUM_THREADS);
             this.skipCommunicationManagerThread = flagValue(source, appId, Flags.SKIP_COMMUNICATIONMANAGER_THREAD);
@@ -231,6 +233,7 @@ public class ModelContextImpl implements ModelContext {
         @Override public double defaultTermwiseLimit() { return defaultTermwiseLimit; }
         @Override public boolean useThreePhaseUpdates() { return useThreePhaseUpdates; }
         @Override public String feedSequencerType() { return feedSequencer; }
+        @Override public int feedTaskLimit() { return feedTaskLimit; }
         @Override public String responseSequencerType() { return responseSequencer; }
         @Override public int defaultNumResponseThreads() { return numResponseThreads; }
         @Override public boolean skipCommunicationManagerThread() { return skipCommunicationManagerThread; }
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 c8c55faa2c7..61c8902496b 100644
--- a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
+++ b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
@@ -60,7 +60,14 @@ public class Flags {
     public static final UnboundStringFlag FEED_SEQUENCER_TYPE = defineStringFlag(
             "feed-sequencer-type", "LATENCY",
             List.of("baldersheim"), "2020-12-02", "2022-01-01",
-            "Selects type of sequenced executor used for feeding, valid values are LATENCY, ADAPTIVE, THROUGHPUT",
+            "Selects type of sequenced executor used for feeding in proton, valid values are LATENCY, ADAPTIVE, THROUGHPUT",
+            "Takes effect at redeployment (requires restart)",
+            ZONE_ID, APPLICATION_ID);
+
+    public static final UnboundIntFlag FEED_TASK_LIMIT = defineIntFlag(
+            "feed-task-limit", 1000,
+            List.of("geirst, baldersheim"), "2021-10-14", "2022-01-01",
+            "The task limit used by the executors handling feed in proton",
             "Takes effect at redeployment",
             ZONE_ID, APPLICATION_ID);