Repository: hyperledger/quilt
Pull request #420 (2020-01-27 23:31:59)

Make soldierOn loop sleep configurable
In `SimpleStreamSender`, we currently sleep for 100 ms in between packet send tries while ramping up.  This should be configurable per #406 

To fix this, I added an `Optional<UnsignedLong>` field to `SimpleStreamSender` and `SendMoneyAggregator` to configure this value, which gets defaulted to 100ms.  Any STREAM sender can configure this value via an env property to instantiate a `SimpleStreamSender`

lgtm-com[bot]: This pull request **introduces 1 alert** when merging 5792eadb49eb62f3662c9f84fb8897630f3ece0e into 549c030db034ff263b7d634eba13719e7ec2679d - [view on LGTM.com](https://lgtm.com/projects/g/hyperledger/quilt/rev/pr-ab86d843a5ca2c2a72856482e0289b8cac4b7af0)

**new alerts:**

* 1 for Unused format argument

Patch:
diff --git a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java
index 291e591..cffde82 100644
--- a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java
+++ b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java
@@ -86,6 +86,7 @@ public class SimpleStreamSender implements StreamSender {
   private final StreamEncryptionService streamEncryptionService;
   private final ExecutorService executorService;
   private final StreamConnectionManager streamConnectionManager;
+  private final Optional<UnsignedLong> sendPacketSleep;
 
   /**
    * Required-args Constructor.
@@ -96,6 +97,16 @@ public class SimpleStreamSender implements StreamSender {
     this(new JavaxStreamEncryptionService(), link);
   }
 
+  /**
+   * Required-args Constructor.
+   *
+   * @param link A {@link Link} that is used to send ILPv4 packets to an immediate peer.
+   * @param sendPacketSleep amount of time for a thread to sleep before sending more packets
+   */
+  public SimpleStreamSender(final Link link, final Optional<UnsignedLong> sendPacketSleep) {
+    this(new JavaxStreamEncryptionService(), link, sendPacketSleep);
+  }
+
   /**
    * Required-args Constructor.
    *
@@ -105,7 +116,22 @@ public class SimpleStreamSender implements StreamSender {
    * @param link                    A {@link Link} that is used to send ILPv4 packets to an immediate peer.
    */
   public SimpleStreamSender(final StreamEncryptionService streamEncryptionService, final Link link) {
-    this(streamEncryptionService, link, newDefaultExecutor());
+    this(streamEncryptionService, link, newDefaultExecutor(), Optional.empty());
+  }
+
+  /**
+   * Required-args Constructor.
+   *
+   * @param streamEncryptionService An instance of {@link StreamEncryptionService} used to encrypt and decrypted
+   *                                end-to-end STREAM packet data (i.e., packets that should only be visible between
+   *                                sender and receiver).
+   * @param link                    A {@link Link} that is used to send ILPv4 packets to an immediate peer.
+   * @param sendPacketSleep amount of time for a thread to sleep before sending more packets
+   */
+  public SimpleStreamSender(final StreamEncryptionService streamEncryptionService,
+                            final Link link,
+                            final Optional<UnsignedLong> sendPacketSleep) {
+    this(streamEncryptionService, link, newDefaultExecutor(), sendPacketSleep);
   }
 
   /**
@@ -120,25 +146,44 @@ public class SimpleStreamSender implements StreamSender {
   public SimpleStreamSender(
       final StreamEncryptionService streamEncryptionService, final Link link, ExecutorService executorService
   ) {
-    this(streamEncryptionService, link, executorService, new StreamConnectionManager());
+    this(streamEncryptionService, link, executorService, new StreamConnectionManager(), Optional.empty());
   }
 
   /**
    * Required-args Constructor.
    *
-   * @param streamEncryptionService A {@link StreamEncryptionService} used to encrypt and decrypted end-to-end STREAM
+   * @param streamEncryptionService An instance of {@link StreamEncryptionService} used to encrypt and decrypted
+   *                                end-to-end STREAM packet data (i.e., packets that should only be visible between
+   *                                sender and receiver).
+   * @param link                    A {@link Link} that is used to send ILPv4 packets to an immediate peer.
+   * @param executorService         executorService to run the payments
+   * @param         sendPacketSleep amount of time for a thread to sleep before sending more packets
+   */
+  public SimpleStreamSender(
+    final StreamEncryptionService streamEncryptionService,
+    final Link link,
+final ExecutorService executorService,
+    final Optional<UnsignedLong> sendPacketSleep
+  ) {
+    this(streamEncryptionService, link, executorService, new StreamConnectionManager(), sendPacketSleep);
+  }
+
+  /**
+   * Required-args Constructor.
+   *  @param streamEncryptionService A {@link StreamEncryptionService} used to encrypt and decrypted end-to-end STREAM
    *                                packet data (i.e., packets that should only be visible between sender and
    *                                receiver).
    * @param link                    A {@link Link} that is used to send ILPv4 packets to an immediate peer.
    * @param executorService         A {@link ExecutorService} to run the payments.
    * @param streamConnectionManager A {@link StreamConnectionManager} that manages connections for all senders and
-   *                                receivers in this JVM.
+   * @param                         sendPacketSleep amount of time for a thread to sleep before sending more packets
    */
   public SimpleStreamSender(
-      final StreamEncryptionService streamEncryptionService,
-      final Link link,
-      final ExecutorService executorService,
-      final StreamConnectionManager streamConnectionManager
+    final StreamEncryptionService streamEncryptionService,
+    final Link link,
+    final ExecutorService executorService,
+    final StreamConnectionManager streamConnectionManager,
+  final Optional<UnsignedLong> sendPacketSleep
   ) {
     this.streamEncryptionService = Objects.requireNonNull(streamEncryptionService);
     this.link = Objects.requireNonNull(link);
@@ -147,6 +192,7 @@ public class SimpleStreamSender implements StreamSender {
     // created using {@link ThreadPoolExecutor} constructors.
     this.executorService = Objects.requireNonNull(executorService);
     this.streamConnectionManager = Objects.requireNonNull(streamConnectionManager);
+    this.sendPacketSleep = Objects.requireNonNull(sendPacketSleep);
   }
 
   private static ExecutorService newDefaultExecutor() {
@@ -162,17 +208,18 @@ public class SimpleStreamSender implements StreamSender {
     Objects.requireNonNull(request);
 
     final StreamConnection streamConnection = this.streamConnectionManager.openConnection(
-        StreamConnectionId.from(request.destinationAddress(), request.sharedSecret())
+      StreamConnectionId.from(request.destinationAddress(), request.sharedSecret())
     );
 
     return new SendMoneyAggregator(
-        this.executorService,
-        streamConnection,
-        StreamCodecContextFactory.oer(),
-        this.link,
-        new AimdCongestionController(),
-        this.streamEncryptionService,
-        request
+      this.executorService,
+      streamConnection,
+      StreamCodecContextFactory.oer(),
+      this.link,
+      new AimdCongestionController(),
+      this.streamEncryptionService,
+      request,
+      this.sendPacketSleep
     ).send();
   }
 
@@ -251,6 +298,8 @@ public class SimpleStreamSender implements StreamSender {
 
     private final AtomicBoolean unrecoverableErrorEncountered;
 
+    private long sendPacketSleep;
+
     /**
      * Required-args Constructor.
      *
@@ -271,7 +320,8 @@ public class SimpleStreamSender implements StreamSender {
         final Link link,
         final CongestionController congestionController,
         final StreamEncryptionService streamEncryptionService,
-        final SendMoneyRequest request
+        final SendMoneyRequest request,
+        final Optional<UnsignedLong> sendPacketSleep
     ) {
       this.executorService = Objects.requireNonNull(executorService);
       this.streamConnection = Objects.requireNonNull(streamConnection);
@@ -298,6 +348,9 @@ public class SimpleStreamSender implements StreamSender {
       this.receiverDenomination = Optional.empty();
 
       this.unrecoverableErrorEncountered = new AtomicBoolean(false);
+
+      Objects.requireNonNull(sendPacketSleep);
+      this.sendPacketSleep = sendPacketSleep.orElse(UnsignedLong.valueOf(10)).longValue();
     }
 
     /**
@@ -471,7 +524,7 @@ public class SimpleStreamSender implements StreamSender {
           try {
             // Don't send any more, but wait a bit for outstanding requests to complete so we don't cycle needlessly in
             // a while loop that doesn't do anything useful.
-            Thread.sleep(100);
+            Thread.sleep(sendPacketSleep);
           } catch (InterruptedException e) {
             throw new StreamSenderException(e.getMessage(), e);
           }
@@ -585,6 +638,13 @@ public class SimpleStreamSender implements StreamSender {
               paymentTracker.rollback(prepareAmounts, false);
             }
           }
+          else {
+            logger.info("timeout reached, not sending packet");
+            congestionController.reject(preparePacket.getAmount(), InterledgerRejectPacket.builder()
+                .code(InterledgerErrorCode.F99_APPLICATION_ERROR)
+                .message(String.format("Timeout reached before packet could be sent", preparePacket))
+                .build());
+          }
         });
       } catch (RejectedExecutionException e) {
         // If we get here, it means the task was unable to be scheduled, so we need to unwind the congestion


Test patch:
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java
index 98158b9..649794f 100644
--- a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java
@@ -114,7 +114,7 @@ public class SendMoneyAggregatorTest {
         .build();
     this.sendMoneyAggregator = new SendMoneyAggregator(
         executor, streamConnectionMock, streamCodecContextMock, linkMock, congestionControllerMock,
-        streamEncryptionServiceMock, request);
+        streamEncryptionServiceMock, request, Optional.empty());
 
     defaultPrepareAmounts = PrepareAmounts.from(samplePreparePacket(), sampleStreamPacket());
   }
@@ -224,7 +224,7 @@ public class SendMoneyAggregatorTest {
     ExecutorService executor = mock(ExecutorService.class);
     this.sendMoneyAggregator = new SendMoneyAggregator(
         executor, streamConnectionMock, streamCodecContextMock, linkMock, congestionControllerMock,
-        streamEncryptionServiceMock, request);
+        streamEncryptionServiceMock, request, Optional.empty());
 
     when(executor.submit(any(Runnable.class))).thenThrow(new RejectedExecutionException());
 
@@ -365,7 +365,7 @@ public class SendMoneyAggregatorTest {
 
     this.sendMoneyAggregator = new SendMoneyAggregator(
         executor, streamConnectionMock, streamCodecContextMock, linkMock, congestionControllerMock,
-        streamEncryptionServiceMock, request);
+        streamEncryptionServiceMock, request, Optional.empty());
 
     setSoldierOnBooleans(false, false, true);
     when(streamConnectionMock.nextSequence()).thenReturn(UnsignedLong.ONE);
@@ -490,7 +490,7 @@ public class SendMoneyAggregatorTest {
 
     this.sendMoneyAggregator = new SendMoneyAggregator(
       executor, streamConnectionMock, streamCodecContextMock, linkMock, congestionControllerMock,
-      streamEncryptionServiceMock, request);
+      streamEncryptionServiceMock, request, Optional.empty());
 
     when(congestionControllerMock.hasInFlight()).thenAnswer(new Answer<Boolean>() {
 
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderIT.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderIT.java
index 3b066c9..25fe1dd 100644
--- a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderIT.java
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderIT.java
@@ -2,6 +2,7 @@ package org.interledger.stream.sender;
 
 import static okhttp3.CookieJar.NO_COOKIES;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
 
 import org.interledger.codecs.ilp.InterledgerCodecContextFactory;
 import org.interledger.core.InterledgerAddress;
@@ -34,9 +35,11 @@ import okhttp3.OkHttpClient;
 import okhttp3.logging.HttpLoggingInterceptor;
 import org.assertj.core.data.Offset;
 import org.junit.Before;
+import org.junit.FixMethodOrder;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.Timeout;
+import org.junit.runners.MethodSorters;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
@@ -53,6 +56,7 @@ import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -65,6 +69,7 @@ import java.util.stream.Collectors;
  * Integration tests for {@link SimpleStreamSender} that connects to a running ILP Connector using the information
  * supplied in this link, and initiates a STREAM payment.
  */
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
 public class SimpleStreamSenderIT {
 
   private static final String AUTH_TOKEN = "password";
@@ -139,6 +144,7 @@ public class SimpleStreamSenderIT {
     RustNodeAccount sender = accountBuilder()
         .username(SENDER_ACCOUNT_USERNAME)
         .ilpAddress(SENDER_ADDRESS)
+        .routingRelation(RustNodeAccount.RoutingRelation.CHILD)
         .build();
 
     nodeClient.createAccount(sender);
@@ -310,6 +316,50 @@ public class SimpleStreamSenderIT {
     logger.info("Payment Sent: {}", sendMoneyResult);
   }
 
+  /**
+   * Two calls to {@link SimpleStreamSender#sendMoney(SendMoneyRequest)}} that involves multiple packets in parallel.
+   * First call is to a {@link SimpleStreamSender} with the default sleep time (100ms)
+   * Second call is to a {@link SimpleStreamSender} with a shorter sleep time (5ms)
+   */
+  @Test
+  public void sendMoneyMultiPacketDifferentSleepTimes() {
+    final SendMoneyResult heavySleeperResult = sendMoneyWithConfiguredSleep(Optional.of(UnsignedLong.valueOf(100)), 1000001);
+    final SendMoneyResult lightSleeperResult = sendMoneyWithConfiguredSleep(Optional.of(UnsignedLong.valueOf(5)), 1000002);
+
+    logger.info("Heavy sleeper took {} to send {} packets.", heavySleeperResult.sendMoneyDuration(), heavySleeperResult.totalPackets());
+    logger.info("Light sleeper took {} to send {} packets.", lightSleeperResult.sendMoneyDuration(), lightSleeperResult.totalPackets());
+    assertThat(heavySleeperResult.sendMoneyDuration()).isGreaterThan(lightSleeperResult.sendMoneyDuration());
+  }
+
+  private SendMoneyResult sendMoneyWithConfiguredSleep(Optional<UnsignedLong> sleepTime, int streamConnectionId) {
+    final UnsignedLong paymentAmount = UnsignedLong.valueOf(100000);
+
+    StreamSender heavySleeperSender = new SimpleStreamSender(
+      new JavaxStreamEncryptionService(), link, sleepTime
+    );
+
+    final StreamConnectionDetails connectionDetails = getStreamConnectionDetails(streamConnectionId);
+
+    final SendMoneyResult heavySleeperResult = heavySleeperSender.sendMoney(
+      SendMoneyRequest.builder()
+        .sourceAddress(SENDER_ADDRESS)
+        .amount(paymentAmount)
+        .denomination(Denominations.XRP)
+        .destinationAddress(connectionDetails.destinationAddress())
+        .sharedSecret(connectionDetails.sharedSecret())
+        .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
+        .build()
+    ).join();
+
+    assertThat(heavySleeperResult.amountDelivered()).isEqualTo(paymentAmount);
+    assertThat(heavySleeperResult.originalAmount()).isEqualTo(paymentAmount);
+    assertThat(heavySleeperResult.numFulfilledPackets()).isCloseTo(8, Offset.offset(1));
+    assertThat(heavySleeperResult.numRejectPackets()).isEqualTo(0);
+
+    logger.info("Payment Sent via sender with sleep = {} : {}", sleepTime.orElse(UnsignedLong.valueOf(100)), heavySleeperResult);
+    return heavySleeperResult;
+  }
+
   /**
    * Multiple calls to {@link SimpleStreamSender#sendMoney(SendMoneyRequest)}} that involves multiple packets in
    * parallel, but using different accounts for each Stream, and thus a different Connection.
@@ -362,8 +412,10 @@ public class SimpleStreamSenderIT {
   public void sendMoneyHonorsTimeout() {
     final UnsignedLong paymentAmount = UnsignedLong.valueOf(10000000);
 
+    // using a sleepy executor here to make sure race condition is handled properly where timeout is reached
+    // after submitting a sendPacketized task to the executor but before the task is executed
     StreamSender streamSender = new SimpleStreamSender(
-        new JavaxStreamEncryptionService(), link
+        new JavaxStreamEncryptionService(), link, new SleepyExecutorService(Executors.newFixedThreadPool(5), 5)
     );
 
     String username = "sendMoneyHonorsTimeout";
@@ -371,29 +423,35 @@ public class SimpleStreamSenderIT {
     RustNodeAccount rustNodeAccount = accountBuilder()
         .username(username)
         .ilpAddress(address)
-        .maxPacketAmount(BigInteger.valueOf(100))
-        .amountPerMinuteLimit(BigInteger.valueOf(1))
-        .packetsPerMinuteLimit(BigInteger.valueOf(1))
         .build();
 
     final StreamConnectionDetails connectionDetails = getStreamConnectionDetails(rustNodeAccount);
 
-    final SendMoneyResult sendMoneyResult = streamSender
-        .sendMoney(
-            SendMoneyRequest.builder()
-                .sourceAddress(SENDER_ADDRESS)
-                .amount(paymentAmount)
-                .denomination(Denominations.XRP)
-                .destinationAddress(connectionDetails.destinationAddress())
-                .sharedSecret(connectionDetails.sharedSecret())
-                .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
-                .timeout(Duration.ofMillis(100))
-                .build()
-        ).join();
-
-    assertThat(sendMoneyResult.successfulPayment()).isFalse();
+    try {
+      // loop to test different timeout amounts
+      for (int i = 0; i < 10; i++) {
+        final SendMoneyResult sendMoneyResult = streamSender
+            .sendMoney(
+                SendMoneyRequest.builder()
+                    .sourceAddress(SENDER_ADDRESS)
+                    .amount(paymentAmount)
+                    .denomination(Denominations.XRP)
+                    .destinationAddress(connectionDetails.destinationAddress())
+                    .sharedSecret(connectionDetails.sharedSecret())
+                    .paymentTracker(new FixedSenderAmountPaymentTracker(paymentAmount, new NoOpExchangeRateCalculator()))
+                    .timeout(Duration.ofMillis(10 + i * 10))
+                    .build()
+            ).get();
+        assertThat(sendMoneyResult.successfulPayment()).isFalse();
+
+        logger.info("Payment Sent: {}", sendMoneyResult);
 
-    logger.info("Payment Sent: {}", sendMoneyResult);
+      }
+    } catch (ExecutionException | InterruptedException e) {
+      logger.error("Error getting completeable future");
+      logger.error("Error getting completeable future: " + e.toString() + " cause: " + e.getCause());
+      fail();
+    }
   }
 
   @Test(expected = NoExchangeRateException.class)
@@ -513,6 +571,7 @@ public class SimpleStreamSenderIT {
     nodeClient.createAccount(accountBuilder()
         .username(connectorAccountUsername)
         .ilpAddress(HOST_ADDRESS.with(connectorAccountUsername))
+        .routingRelation(RustNodeAccount.RoutingRelation.CHILD)
         .build());
 
     final UnsignedLong paymentAmount = UnsignedLong.valueOf(1000);
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderTests.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderTests.java
index 2516272..40aa3ee 100644
--- a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderTests.java
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SimpleStreamSenderTests.java
@@ -14,6 +14,7 @@ import org.junit.rules.ExpectedException;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 
+import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 
 /**
@@ -65,7 +66,13 @@ public class SimpleStreamSenderTests {
   @Test
   public void constructThreeArgWithNullExecutor() {
     expectedException.expect(NullPointerException.class);
-    new SimpleStreamSender(streamEncryptionServiceMock, linkMock, null);
+    new SimpleStreamSender(streamEncryptionServiceMock, linkMock, (ExecutorService) null);
+  }
+
+  @Test
+  public void constructThreeArgWithNullSleep() {
+    expectedException.expect(NullPointerException.class);
+    new SimpleStreamSender(streamEncryptionServiceMock, linkMock, (Optional<UnsignedLong>) null);
   }
 
   @Test
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SleepyExecutorService.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SleepyExecutorService.java
new file mode 100644
index 0000000..608a099
--- /dev/null
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SleepyExecutorService.java
@@ -0,0 +1,121 @@
+package org.interledger.stream.sender;
+
+import org.jetbrains.annotations.NotNull;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * For testing purposes, an executor service that sleeps before submitting or executing tasks to simulate conditions
+ * where there is a delay between when a task is submitted and when it is executed.
+ */
+class SleepyExecutorService implements ExecutorService {
+
+  private final ExecutorService delegate;
+  private final long sleep;
+
+  SleepyExecutorService(ExecutorService delegate, long sleep) {
+    this.delegate = delegate;
+    this.sleep = sleep;
+  }
+
+  @Override
+  public void shutdown() {
+    delegate.shutdown();
+  }
+
+  @NotNull
+  @Override
+  public List<Runnable> shutdownNow() {
+    return delegate.shutdownNow();
+  }
+
+  @Override
+  public boolean isShutdown() {
+    return delegate.isShutdown();
+  }
+
+  @Override
+  public boolean isTerminated() {
+    return delegate.isTerminated();
+  }
+
+  @Override
+  public boolean awaitTermination(long timeout, @NotNull TimeUnit unit) throws InterruptedException {
+    return delegate.awaitTermination(timeout, unit);
+  }
+
+  @NotNull
+  @Override
+  public <T> Future<T> submit(@NotNull Callable<T> task) {
+    return withSleep(() -> delegate.submit(task));
+  }
+
+  @NotNull
+  @Override
+  public <T> Future<T> submit(@NotNull Runnable task, T result) {
+    return withSleep(() -> delegate.submit(task, result));
+  }
+
+  @NotNull
+  @Override
+  public Future<?> submit(@NotNull Runnable task) {
+    return withSleep(() -> delegate.submit(task));
+  }
+
+  @NotNull
+  @Override
+  public <T> List<Future<T>> invokeAll(@NotNull Collection<? extends Callable<T>> tasks) throws InterruptedException {
+    return delegate.invokeAll(tasks);
+  }
+
+  @NotNull
+  @Override
+  public <T> List<Future<T>> invokeAll(@NotNull Collection<? extends Callable<T>> tasks,
+                                       long timeout,
+                                       @NotNull TimeUnit unit) throws InterruptedException {
+    return delegate.invokeAll(tasks, timeout, unit);
+  }
+
+  @NotNull
+  @Override
+  public <T> T invokeAny(@NotNull Collection<? extends Callable<T>> tasks)
+      throws InterruptedException, ExecutionException {
+    return delegate.invokeAny(tasks);
+  }
+
+  @Override
+  public <T> T invokeAny(@NotNull Collection<? extends Callable<T>> tasks, long timeout, @NotNull TimeUnit unit)
+      throws InterruptedException, ExecutionException, TimeoutException {
+    return delegate.invokeAny(tasks, timeout, unit);
+  }
+
+  @Override
+  public void execute(@NotNull Runnable command) {
+    withSleep(() -> delegate.execute(command));
+  }
+
+  private <T> T withSleep(Callable<T> task) {
+    try {
+      Thread.sleep(sleep);
+      return task.call();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void withSleep(Runnable task) {
+    try {
+      Thread.sleep(sleep);
+      task.run();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+}