Support variable reindexing speed, based on config

6 files changed, 117 insertions, 49 deletions

diff --git a/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/Reindexer.java b/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/Reindexer.java
index 1d4571e3fc6..638968cc03e 100644
--- a/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/Reindexer.java
+++ b/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/Reindexer.java
@@ -2,6 +2,7 @@
 package ai.vespa.reindexing;
 
 import ai.vespa.reindexing.Reindexing.Status;
+import ai.vespa.reindexing.Reindexing.Trigger;
 import ai.vespa.reindexing.ReindexingCurator.ReindexingLockException;
 import com.yahoo.document.DocumentType;
 import com.yahoo.document.select.parser.ParseException;
@@ -18,18 +19,20 @@ import com.yahoo.vespa.curator.Lock;
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.TreeMap;
 import java.util.concurrent.Phaser;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.logging.Logger;
 
+import static java.util.Comparator.comparingDouble;
 import static java.util.Objects.requireNonNull;
 import static java.util.logging.Level.FINE;
 import static java.util.logging.Level.INFO;
 import static java.util.logging.Level.WARNING;
+import static java.util.stream.Collectors.toUnmodifiableList;
 
 /**
  * Progresses reindexing efforts by creating visitor sessions against its own content cluster,
@@ -45,14 +48,14 @@ public class Reindexer {
     static final Duration failureGrace = Duration.ofMinutes(10);
 
     private final Cluster cluster;
-    private final Map<DocumentType, Instant> ready;
+    private final List<Trigger> ready;
     private final ReindexingCurator database;
     private final Function<VisitorParameters, Runnable> visitorSessions;
     private final ReindexingMetrics metrics;
     private final Clock clock;
     private final Phaser phaser = new Phaser(2); // Reindexer and visitor.
 
-    public Reindexer(Cluster cluster, Map<DocumentType, Instant> ready, ReindexingCurator database,
+    public Reindexer(Cluster cluster, List<Trigger> ready, ReindexingCurator database,
                      DocumentAccess access, Metric metric, Clock clock) {
         this(cluster,
              ready,
@@ -70,13 +73,15 @@ public class Reindexer {
         );
     }
 
-    Reindexer(Cluster cluster, Map<DocumentType, Instant> ready, ReindexingCurator database,
+    Reindexer(Cluster cluster, List<Trigger> ready, ReindexingCurator database,
               Function<VisitorParameters, Runnable> visitorSessions, Metric metric, Clock clock) {
-        for (DocumentType type : ready.keySet())
-            cluster.bucketSpaceOf(type); // Verifies this is known.
+        for (Trigger trigger : ready)
+            cluster.bucketSpaceOf(trigger.type()); // Verifies this is known.
 
         this.cluster = cluster;
-        this.ready = new TreeMap<>(ready); // Iterate through document types in consistent order.
+        this.ready = ready.stream() // Iterate through document types in consistent order.
+                          .sorted(comparingDouble(Trigger::speed).reversed().thenComparing(Trigger::readyAt).thenComparing(Trigger::type))
+                          .collect(toUnmodifiableList());
         this.database = database;
         this.visitorSessions = visitorSessions;
         this.metrics = new ReindexingMetrics(metric, cluster.name);
@@ -104,12 +109,13 @@ public class Reindexer {
             database.writeReindexing(reindexing.get(), cluster.name());
             metrics.dump(reindexing.get());
 
-            for (DocumentType type : ready.keySet()) { // We consider only document types for which we have config.
-                if (ready.get(type).isAfter(clock.instant()))
+            // We consider only document types for which we have config.
+            for (Trigger trigger : ready) {
+                if (trigger.readyAt().isAfter(clock.instant()))
                     log.log(INFO, "Received config for reindexing which is ready in the future — will process later " +
-                                  "(" + ready.get(type) + " is after " + clock.instant() + ")");
+                                  "(" + trigger.readyAt() + " is after " + clock.instant() + ")");
                 else
-                    progress(type, reindexing, new AtomicReference<>(reindexing.get().status().get(type)));
+                    progress(trigger.type(), trigger.speed(), reindexing, new AtomicReference<>(reindexing.get().status().get(trigger.type())));
 
                 if (phaser.isTerminated())
                     break;
@@ -117,21 +123,21 @@ public class Reindexer {
         }
     }
 
-    static Reindexing updateWithReady(Map<DocumentType, Instant> ready, Reindexing reindexing, Instant now) {
-        for (DocumentType type : ready.keySet()) { // We update only for document types for which we have config.
-            if ( ! ready.get(type).isAfter(now)) {
-                Status status = reindexing.status().getOrDefault(type, Status.ready(now));
-                if (status.startedAt().isBefore(ready.get(type)))
+    static Reindexing updateWithReady(List<Trigger> ready, Reindexing reindexing, Instant now) {
+        for (Trigger trigger : ready) { // We update only for document types for which we have config.
+            if ( ! trigger.readyAt().isAfter(now)) {
+                Status status = reindexing.status().get(trigger.type());
+                if (status == null || status.startedAt().isBefore(trigger.readyAt()))
                     status = Status.ready(now);
 
-                reindexing = reindexing.with(type, status);
+                reindexing = reindexing.with(trigger.type(), status);
             }
         }
         return reindexing;
     }
 
     @SuppressWarnings("fallthrough") // (ノಠ ∩ಠ)ノ彡( \o°o)\
-    private void progress(DocumentType type, AtomicReference<Reindexing> reindexing, AtomicReference<Status> status) {
+    private void progress(DocumentType type, double speed, AtomicReference<Reindexing> reindexing, AtomicReference<Status> status) {
         switch (status.get().state()) {
             default:
                 log.log(WARNING, "Unknown reindexing state '" + status.get().state() + "'—not continuing reindexing of " + type);
@@ -167,7 +173,7 @@ public class Reindexer {
             }
         };
 
-        VisitorParameters parameters = createParameters(type, status.get().progress().orElse(null));
+        VisitorParameters parameters = createParameters(type, speed, status.get().progress().orElse(null));
         parameters.setControlHandler(control);
         Runnable sessionShutdown = visitorSessions.apply(parameters); // Also starts the visitor session.
         log.log(FINE, () -> "Running reindexing of " + type);
@@ -197,12 +203,12 @@ public class Reindexer {
         metrics.dump(reindexing.get());
     }
 
-    VisitorParameters createParameters(DocumentType type, ProgressToken progress) {
+    VisitorParameters createParameters(DocumentType type, double speed, ProgressToken progress) {
         VisitorParameters parameters = new VisitorParameters(type.getName());
-        parameters.setThrottlePolicy(new DynamicThrottlePolicy().setWindowSizeIncrement(0.2)
+        parameters.setThrottlePolicy(new DynamicThrottlePolicy().setWindowSizeIncrement(speed)
                                                                 .setWindowSizeDecrementFactor(5)
                                                                 .setResizeRate(10)
-                                                                .setMinWindowSize(1));
+                                                                .setMinWindowSize((int) (5 + speed)));
         parameters.setRemoteDataHandler(cluster.name());
         parameters.setMaxPending(8);
         parameters.setResumeToken(progress);
diff --git a/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/Reindexing.java b/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/Reindexing.java
index 2fd38bb6aa5..6f13a07414c 100644
--- a/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/Reindexing.java
+++ b/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/Reindexing.java
@@ -191,4 +191,53 @@ public class Reindexing {
 
     }
 
+
+    public static class Trigger {
+
+        private final DocumentType type;
+        private final Instant readyAt;
+        private final double speed;
+
+        public Trigger(DocumentType type, Instant readyAt, double speed) {
+            this.type = requireNonNull(type);
+            this.readyAt = requireNonNull(readyAt);
+            this.speed = speed;
+        }
+
+        public DocumentType type() {
+            return type;
+        }
+
+        public Instant readyAt() {
+            return readyAt;
+        }
+
+        public double speed() {
+            return speed;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Trigger trigger = (Trigger) o;
+            return Double.compare(trigger.speed, speed) == 0 && type.equals(trigger.type) && readyAt.equals(trigger.readyAt);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(type, readyAt, speed);
+        }
+
+        @Override
+        public String toString() {
+            return "Trigger{" +
+                   "type=" + type +
+                   ", readyAt=" + readyAt +
+                   ", speed=" + speed +
+                   '}';
+        }
+
+    }
+
 }
diff --git a/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/ReindexingCurator.java b/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/ReindexingCurator.java
index 6dd4079835d..0d77792de6d 100644
--- a/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/ReindexingCurator.java
+++ b/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/ReindexingCurator.java
@@ -2,6 +2,7 @@
 package ai.vespa.reindexing;
 
 import ai.vespa.reindexing.Reindexing.Status;
+import ai.vespa.reindexing.Reindexing.Trigger;
 import com.google.common.util.concurrent.UncheckedTimeoutException;
 import com.yahoo.document.DocumentType;
 import com.yahoo.document.DocumentTypeManager;
@@ -16,6 +17,7 @@ import com.yahoo.yolean.Exceptions;
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
 import java.util.logging.Level;
@@ -48,16 +50,16 @@ public class ReindexingCurator {
     }
 
     /** If no reindexing data exists (has been wiped), assume current ready documents are already done. */
-    public void initializeIfEmpty(String cluster, Map<DocumentType, Instant> ready, Instant now) {
+    public void initializeIfEmpty(String cluster, List<Trigger> ready, Instant now) {
         if ( ! curator.exists(statusPath(cluster))) {
             try (Lock lock = lockReindexing(cluster)) {
                 if (curator.exists(statusPath(cluster)))
                     return; // Some other node already did this.
 
                 Reindexing reindexing = Reindexing.empty();
-                for (DocumentType type : ready.keySet())
-                    if (ready.get(type).isBefore(now))
-                        reindexing = reindexing.with(type, Status.ready(now).running().successful(now));
+                for (Trigger trigger : ready)
+                    if (trigger.readyAt().isBefore(now))
+                        reindexing = reindexing.with(trigger.type(), Status.ready(now).running().successful(now));
 
                 log.log(Level.INFO, "Creating initial reindexing status at '" + statusPath(cluster) + "'");
                 writeReindexing(reindexing, cluster);
diff --git a/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/ReindexingMaintainer.java b/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/ReindexingMaintainer.java
index 405a6991d23..e784f070188 100644
--- a/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/ReindexingMaintainer.java
+++ b/clustercontroller-reindexer/src/main/java/ai/vespa/reindexing/ReindexingMaintainer.java
@@ -2,15 +2,14 @@
 package ai.vespa.reindexing;
 
 import ai.vespa.reindexing.Reindexer.Cluster;
+import ai.vespa.reindexing.Reindexing.Trigger;
 import ai.vespa.reindexing.ReindexingCurator.ReindexingLockException;
 import com.google.inject.Inject;
 import com.yahoo.cloud.config.ClusterListConfig;
 import com.yahoo.cloud.config.ZookeepersConfig;
 import com.yahoo.component.AbstractComponent;
 import com.yahoo.concurrent.DaemonThreadFactory;
-import com.yahoo.document.DocumentType;
 import com.yahoo.document.DocumentTypeManager;
-import com.yahoo.document.config.DocumentmanagerConfig;
 import com.yahoo.documentapi.DocumentAccess;
 import com.yahoo.jdisc.Metric;
 import com.yahoo.net.HostName;
@@ -23,13 +22,11 @@ import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.logging.Logger;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import static java.util.logging.Level.FINE;
@@ -37,7 +34,6 @@ import static java.util.logging.Level.WARNING;
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toMap;
 import static java.util.stream.Collectors.toUnmodifiableList;
-import static java.util.stream.Collectors.toUnmodifiableMap;
 
 /**
  * Runs in all cluster controller containers, and progresses reindexing efforts.
@@ -117,10 +113,12 @@ public class ReindexingMaintainer extends AbstractComponent {
         curator.close();
     }
 
-    static Map<DocumentType, Instant> parseReady(ReindexingConfig.Clusters cluster, DocumentTypeManager manager) {
+    static List<Trigger> parseReady(ReindexingConfig.Clusters cluster, DocumentTypeManager manager) {
         return cluster.documentTypes().entrySet().stream()
-                      .collect(toUnmodifiableMap(typeStatus -> manager.getDocumentType(typeStatus.getKey()),
-                                                 typeStatus -> Instant.ofEpochMilli(typeStatus.getValue().readyAtMillis())));
+                      .map(typeStatus -> new Trigger(manager.getDocumentType(typeStatus.getKey()),
+                                                     Instant.ofEpochMilli(typeStatus.getValue().readyAtMillis()),
+                                                     typeStatus.getValue().speed()))
+                      .collect(toUnmodifiableList());
     }
 
     /** Schedules a task with the given interval (across all containers in this ZK cluster). */
diff --git a/clustercontroller-reindexer/src/test/java/ai/vespa/reindexing/ReindexerTest.java b/clustercontroller-reindexer/src/test/java/ai/vespa/reindexing/ReindexerTest.java
index 70a65b4a072..5737f038a17 100644
--- a/clustercontroller-reindexer/src/test/java/ai/vespa/reindexing/ReindexerTest.java
+++ b/clustercontroller-reindexer/src/test/java/ai/vespa/reindexing/ReindexerTest.java
@@ -3,6 +3,7 @@ package ai.vespa.reindexing;
 
 import ai.vespa.reindexing.Reindexer.Cluster;
 import ai.vespa.reindexing.Reindexing.Status;
+import ai.vespa.reindexing.Reindexing.Trigger;
 import ai.vespa.reindexing.ReindexingCurator.ReindexingLockException;
 import com.yahoo.document.DocumentType;
 import com.yahoo.document.DocumentTypeManager;
@@ -12,6 +13,7 @@ import com.yahoo.documentapi.VisitorControlHandler;
 import com.yahoo.documentapi.VisitorParameters;
 import com.yahoo.documentapi.messagebus.protocol.DocumentProtocol;
 import com.yahoo.jdisc.test.MockMetric;
+import com.yahoo.messagebus.DynamicThrottlePolicy;
 import com.yahoo.searchdefinition.derived.Deriver;
 import com.yahoo.test.ManualClock;
 import com.yahoo.vespa.curator.mock.MockCurator;
@@ -21,13 +23,17 @@ import org.junit.jupiter.api.Timeout;
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
+import static java.util.stream.Collectors.toList;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -57,12 +63,12 @@ class ReindexerTest {
     @Test
     void throwsWhenUnknownBuckets() {
         assertThrows(NullPointerException.class,
-                     () -> new Reindexer(new Cluster("cluster", Map.of()), Map.of(music, Instant.EPOCH), database, ReindexerTest::failIfCalled, metric, clock));
+                     () -> new Reindexer(new Cluster("cluster", Map.of()), triggers(0), database, ReindexerTest::failIfCalled, metric, clock));
     }
 
     @Test
     void throwsWhenLockHeldElsewhere() throws InterruptedException, ExecutionException {
-        Reindexer reindexer = new Reindexer(cluster, Map.of(music, Instant.EPOCH), database, ReindexerTest::failIfCalled, metric, clock);
+        Reindexer reindexer = new Reindexer(cluster, triggers(0), database, ReindexerTest::failIfCalled, metric, clock);
         Executors.newSingleThreadExecutor().submit(() -> database.lockReindexing("cluster")).get();
         assertThrows(ReindexingLockException.class, reindexer::reindex);
     }
@@ -70,15 +76,15 @@ class ReindexerTest {
     @Test
     @Timeout(10)
     void nothingToDoWithEmptyConfig() throws ReindexingLockException {
-        new Reindexer(cluster, Map.of(), database, ReindexerTest::failIfCalled, metric, clock).reindex();
+        new Reindexer(cluster, triggers(), database, ReindexerTest::failIfCalled, metric, clock).reindex();
         assertEquals(Map.of(), metric.metrics());
     }
 
     @Test
     void testParameters() {
-        Reindexer reindexer = new Reindexer(cluster, Map.of(), database, ReindexerTest::failIfCalled, metric, clock);
+        Reindexer reindexer = new Reindexer(cluster, triggers(), database, ReindexerTest::failIfCalled, metric, clock);
         ProgressToken token = new ProgressToken();
-        VisitorParameters parameters = reindexer.createParameters(music, token);
+        VisitorParameters parameters = reindexer.createParameters(music, 0.3, token);
         assertEquals("music:[document]", parameters.getFieldSet());
         assertSame(token, parameters.getResumeToken());
         assertEquals("default", parameters.getBucketSpace());
@@ -86,27 +92,28 @@ class ReindexerTest {
         assertEquals("cluster", parameters.getRemoteDataHandler());
         assertEquals("music", parameters.getDocumentSelection());
         assertEquals(DocumentProtocol.Priority.NORMAL_3, parameters.getPriority());
+        assertEquals(0.3, ((DynamicThrottlePolicy) parameters.getThrottlePolicy()).getWindowSizeIncrement());
     }
 
     @Test
     @Timeout(10)
     void testReindexing() throws ReindexingLockException {
         // Reindexer is created against en empty database, so any ready document types are assumed already done.
-        new Reindexer(cluster, Map.of(music, Instant.ofEpochMilli(-10)), database, ReindexerTest::failIfCalled, metric, clock).reindex();
+        new Reindexer(cluster, triggers(-10), database, ReindexerTest::failIfCalled, metric, clock).reindex();
         Reindexing reindexing = Reindexing.empty().with(music, Status.ready(Instant.EPOCH).running().successful(Instant.EPOCH));
         assertEquals(reindexing, database.readReindexing("cluster"));
 
         // New config tells reindexer to reindex "music" documents no earlier than at 10 millis after EPOCH, which isn't yet.
         // Nothing happens, since it's not yet time. This isn't supposed to happen unless high clock skew.
         clock.advance(Duration.ofMillis(5));
-        new Reindexer(cluster, Map.of(music, Instant.ofEpochMilli(10)), database, ReindexerTest::failIfCalled, metric, clock).reindex();
+        new Reindexer(cluster, triggers(10), database, ReindexerTest::failIfCalled, metric, clock).reindex();
         assertEquals(reindexing, database.readReindexing("cluster"));
 
         // It's time to reindex the "music" documents — let this complete successfully.
         clock.advance(Duration.ofMillis(10));
         AtomicBoolean shutDown = new AtomicBoolean();
         Executor executor = Executors.newSingleThreadExecutor();
-        new Reindexer(cluster, Map.of(music, Instant.ofEpochMilli(10)), database, parameters -> {
+        new Reindexer(cluster, triggers(10), database, parameters -> {
                     database.writeReindexing(Reindexing.empty(), "cluster"); // Wipe database to verify we write data from reindexer.
                     executor.execute(() -> parameters.getControlHandler().onDone(VisitorControlHandler.CompletionCode.SUCCESS, "OK"));
                     return () -> shutDown.set(true);
@@ -137,7 +144,7 @@ class ReindexerTest {
         metric.metrics().clear();
         shutDown.set(false);
         AtomicReference<Reindexer> aborted = new AtomicReference<>();
-        aborted.set(new Reindexer(cluster, Map.of(music, Instant.ofEpochMilli(20)), database, parameters -> {
+        aborted.set(new Reindexer(cluster, triggers(20), database, parameters -> {
                     database.writeReindexing(Reindexing.empty(), "cluster"); // Wipe database to verify we write data from reindexer.
                     parameters.getControlHandler().onProgress(new ProgressToken());
                     aborted.get().shutdown();
@@ -157,13 +164,13 @@ class ReindexerTest {
                                        "state", "pending")));
 
         // Reindexer is created without any ready document types, which means nothing should run.
-        new Reindexer(cluster, Map.of(), database, ReindexerTest::failIfCalled, metric, clock).reindex();
+        new Reindexer(cluster, triggers(), database, ReindexerTest::failIfCalled, metric, clock).reindex();
         assertEquals(reindexing, database.readReindexing("cluster"));
 
         // Last reindexing fails.
         clock.advance(Duration.ofMillis(10));
         shutDown.set(false);
-        new Reindexer(cluster, Map.of(music, Instant.ofEpochMilli(30)), database, parameters -> {
+        new Reindexer(cluster, triggers(30), database, parameters -> {
                     database.writeReindexing(Reindexing.empty(), "cluster"); // Wipe database to verify we write data from reindexer.
                     executor.execute(() -> parameters.getControlHandler().onDone(VisitorControlHandler.CompletionCode.FAILURE, "Error"));
                     return () -> shutDown.set(true);
@@ -174,13 +181,13 @@ class ReindexerTest {
 
         // Document type is ignored in next run, as it has failed, and grace period is not yet over.
         clock.advance(Reindexer.failureGrace.minusMillis(1));
-        new Reindexer(cluster, Map.of(music, Instant.ofEpochMilli(30)), database, ReindexerTest::failIfCalled, metric, clock).reindex();
+        new Reindexer(cluster, triggers(30), database, ReindexerTest::failIfCalled, metric, clock).reindex();
         assertEquals(reindexing, database.readReindexing("cluster"));
 
         // When failure grace period is over, reindexing resumes as usual.
         clock.advance(Duration.ofMillis(1));
         shutDown.set(false);
-        new Reindexer(cluster, Map.of(music, Instant.ofEpochMilli(30)), database, parameters -> {
+        new Reindexer(cluster, triggers(30), database, parameters -> {
             executor.execute(() -> parameters.getControlHandler().onDone(VisitorControlHandler.CompletionCode.SUCCESS, "OK"));
             return () -> shutDown.set(true);
         }, metric, clock).reindex();
@@ -189,4 +196,8 @@ class ReindexerTest {
         assertTrue(shutDown.get(), "Session was shut down");
     }
 
+    List<Trigger> triggers(long... ready) {
+        return Arrays.stream(ready).mapToObj(at -> new Trigger(music, Instant.ofEpochMilli(at), 0.2)).collect(toList());
+    }
+
 }
diff --git a/clustercontroller-reindexer/src/test/java/ai/vespa/reindexing/ReindexingMaintainerTest.java b/clustercontroller-reindexer/src/test/java/ai/vespa/reindexing/ReindexingMaintainerTest.java
index 2d2119a4b97..c5043f0fac3 100644
--- a/clustercontroller-reindexer/src/test/java/ai/vespa/reindexing/ReindexingMaintainerTest.java
+++ b/clustercontroller-reindexer/src/test/java/ai/vespa/reindexing/ReindexingMaintainerTest.java
@@ -2,6 +2,7 @@
 package ai.vespa.reindexing;
 
 import ai.vespa.reindexing.Reindexer.Cluster;
+import ai.vespa.reindexing.Reindexing.Trigger;
 import com.yahoo.cloud.config.ClusterListConfig;
 import com.yahoo.document.DocumentTypeManager;
 import com.yahoo.document.config.DocumentmanagerConfig;
@@ -12,6 +13,7 @@ import org.junit.jupiter.api.Test;
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 import java.util.Map;
 
 import static ai.vespa.reindexing.ReindexingMaintainer.parseCluster;
@@ -30,9 +32,9 @@ class ReindexingMaintainerTest {
         DocumentmanagerConfig musicConfig = Deriver.getDocumentManagerConfig("src/test/resources/schemas/music.sd").build();
         DocumentTypeManager manager = new DocumentTypeManager(musicConfig);
 
-        assertEquals(Map.of(manager.getDocumentType("music"), Instant.ofEpochMilli(123)),
+        assertEquals(List.of(new Trigger(manager.getDocumentType("music"), Instant.ofEpochMilli(123), 0.5)),
                      parseReady(new ReindexingConfig.Clusters.Builder()
-                                        .documentTypes("music", new ReindexingConfig.Clusters.DocumentTypes.Builder().readyAtMillis(123))
+                                        .documentTypes("music", new ReindexingConfig.Clusters.DocumentTypes.Builder().readyAtMillis(123).speed(0.5))
                                         .build(),
                                 manager));