Fix the simulation scheduler to avoid fast-forward in external use

simulation/src/main/java/com/palantir/dialogue/core/Benchmark.java

@@ -142,7 +142,7 @@ public void update(Duration _time, long _requestsStarted, long responsesReceived
     public Benchmark abortAfter(Duration cutoff) {
         simulation
                 .scheduler()
-                .schedule(
+                .scheduleAndFastForward(
                         () -> {
                             log.warn(
                                     "Aborted running benchmark after cutoff reached - strategy might be buggy {}",
@@ -192,7 +192,7 @@ public void onFailure(Throwable throwable) {
 
             simulation
                     .scheduler()
-                    .schedule(
+                    .scheduleAndFastForward(
                             () -> {
                                 log.debug(
                                         "time={} starting num={} {}",

simulation/src/main/java/com/palantir/dialogue/core/ExternalDeterministicScheduler.java

@@ -39,26 +39,25 @@ final class ExternalDeterministicScheduler implements ListeningScheduledExecutor
 
     @Override
     public ListenableScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
-        return schedule(
-                () -> {
-                    command.run();
-                    return null;
-                },
-                delay,
-                unit);
+        return delegate.schedule(command, delay, unit);
     }
 
     @Override
     public <V> ListenableScheduledFuture<V> schedule(Callable<V> command, long delay, TimeUnit unit) {
+        return delegate.schedule(command, delay, unit);
+    }
+
+    @SuppressWarnings({"CheckReturnValue", "FutureReturnValueIgnored"})
+    public void scheduleAndFastForward(Runnable command, long delay, TimeUnit unit) {
         long scheduleTime = ticker.read();
         long delayNanos = unit.toNanos(delay);
 
         RuntimeException exceptionForStackTrace = new RuntimeException();
-        return delegate.schedule(
+        delegate.schedule(
                 () -> {
                     try {
                         ticker.advanceTo(Duration.ofNanos(scheduleTime + delayNanos));
-                        return command.call();
+                        command.run();
                     } catch (Throwable e) {
                         e.addSuppressed(exceptionForStackTrace);
                         throw e;

simulation/src/main/java/com/palantir/dialogue/core/Simulation.java

@@ -17,7 +17,6 @@
 package com.palantir.dialogue.core;
 
 import com.github.benmanes.caffeine.cache.Ticker;
-import com.google.common.util.concurrent.ListeningScheduledExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
@@ -30,7 +29,7 @@ final class Simulation {
     private static final Logger log = LoggerFactory.getLogger(Simulation.class);
 
     private final DeterministicScheduler deterministicExecutor = new DeterministicScheduler();
-    private final ListeningScheduledExecutorService listenableExecutor;
+    private final ExternalDeterministicScheduler listenableExecutor;
 
     private final TestCaffeineTicker ticker = new TestCaffeineTicker();
     private final SimulationMetricsReporter metrics = new SimulationMetricsReporter(this);
@@ -48,7 +47,7 @@ public Ticker clock() {
         return ticker; // read only!
     }
 
-    public ListeningScheduledExecutorService scheduler() {
+    public ExternalDeterministicScheduler scheduler() {
         return listenableExecutor;
     }