core,netty: plumb fake clock into transport tracer tests (#3724)

core/src/main/java/io/grpc/internal/AbstractManagedChannelImplBuilder.java

@@ -124,6 +124,8 @@ public abstract class AbstractManagedChannelImplBuilder
 
   long idleTimeoutMillis = IDLE_MODE_DEFAULT_TIMEOUT_MILLIS;
 
+  protected TransportTracer.Factory transportTracerFactory = TransportTracer.getDefaultFactory();
+
   private int maxInboundMessageSize = GrpcUtil.DEFAULT_MAX_MESSAGE_SIZE;
 
   /**

core/src/main/java/io/grpc/internal/AbstractServerImplBuilder.java

@@ -107,6 +107,8 @@ public abstract class AbstractServerImplBuilder<T extends AbstractServerImplBuil
   private boolean recordFinishedRpcs = true;
   private boolean tracingEnabled = true;
 
+  protected TransportTracer.Factory transportTracerFactory = TransportTracer.getDefaultFactory();
+
   @Override
   public final T directExecutor() {
     return executor(MoreExecutors.directExecutor());

core/src/main/java/io/grpc/internal/TransportTracer.java

@@ -16,6 +16,7 @@
 
 package io.grpc.internal;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import java.util.concurrent.TimeUnit;
 
@@ -24,6 +25,15 @@ import java.util.concurrent.TimeUnit;
  * Can only be called from the transport thread unless otherwise noted.
  */
 public final class TransportTracer {
+  private static final TimeProvider SYSTEM_TIME_PROVIDER = new TimeProvider() {
+    @Override
+    public long currentTimeMillis() {
+      return System.currentTimeMillis();
+    }
+  };
+  private static final Factory DEFAULT_FACTORY = new Factory(SYSTEM_TIME_PROVIDER);
+
+  private final TimeProvider timeProvider;
   private long streamsStarted;
   private long lastStreamCreatedTimeNanos;
   private long streamsSucceeded;
@@ -37,6 +47,14 @@ public final class TransportTracer {
   private final LongCounter messagesReceived = LongCounterFactory.create();
   private volatile long lastMessageReceivedTimeNanos;
 
+  public TransportTracer() {
+    this.timeProvider = SYSTEM_TIME_PROVIDER;
+  }
+
+  private TransportTracer(TimeProvider timeProvider) {
+    this.timeProvider = timeProvider;
+  }
+
   /**
    * Returns a read only set of current stats.
    */
@@ -128,8 +146,8 @@ public final class TransportTracer {
     FlowControlWindows read();
   }
 
-  private static long currentTimeNanos() {
+  private long currentTimeNanos() {
-    return TimeUnit.MILLISECONDS.toNanos(System.currentTimeMillis());
+    return TimeUnit.MILLISECONDS.toNanos(timeProvider.currentTimeMillis());
   }
 
   /**
@@ -178,4 +196,31 @@ public final class TransportTracer {
       }
     }
   }
+
+  /**
+   * Time source representing the current system time in millis. Used to inject a fake clock
+   * into unit tests.
+   */
+  @VisibleForTesting
+  public interface TimeProvider {
+    /** Returns the current milli time. */
+    long currentTimeMillis();
+  }
+
+  public static final class Factory {
+    private TimeProvider timeProvider;
+
+    @VisibleForTesting
+    public Factory(TimeProvider timeProvider) {
+      this.timeProvider = timeProvider;
+    }
+
+    public TransportTracer create() {
+      return new TransportTracer(timeProvider);
+    }
+  }
+
+  public static Factory getDefaultFactory() {
+    return DEFAULT_FACTORY;
+  }
 }

netty/src/main/java/io/grpc/netty/NettyChannelBuilder.java

@@ -38,6 +38,7 @@ import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.KeepAliveManager;
 import io.grpc.internal.ProxyParameters;
 import io.grpc.internal.SharedResourceHolder;
+import io.grpc.internal.TransportTracer;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
@@ -320,7 +321,8 @@ public final class NettyChannelBuilder
   protected ClientTransportFactory buildTransportFactory() {
     return new NettyTransportFactory(dynamicParamsFactory, channelType, channelOptions,
         negotiationType, sslContext, eventLoopGroup, flowControlWindow, maxInboundMessageSize(),
-        maxHeaderListSize, keepAliveTimeNanos, keepAliveTimeoutNanos, keepAliveWithoutCalls);
+        maxHeaderListSize, keepAliveTimeNanos, keepAliveTimeoutNanos, keepAliveWithoutCalls,
+        transportTracerFactory.create());
   }
 
   @Override
@@ -408,6 +410,12 @@ public final class NettyChannelBuilder
     super.setStatsEnabled(value);
   }
 
+  @VisibleForTesting
+  NettyChannelBuilder setTransportTracerFactory(TransportTracer.Factory transportTracerFactory) {
+    this.transportTracerFactory = transportTracerFactory;
+    return this;
+  }
+
   interface TransportCreationParamsFilterFactory {
     @CheckReturnValue
     TransportCreationParamsFilter create(
@@ -445,6 +453,7 @@ public final class NettyChannelBuilder
     private final AtomicBackoff keepAliveTimeNanos;
     private final long keepAliveTimeoutNanos;
     private final boolean keepAliveWithoutCalls;
+    private final TransportTracer transportTracer;
 
     private boolean closed;
 
@@ -452,10 +461,12 @@ public final class NettyChannelBuilder
         Class<? extends Channel> channelType, Map<ChannelOption<?>, ?> channelOptions,
         NegotiationType negotiationType, SslContext sslContext, EventLoopGroup group,
         int flowControlWindow, int maxMessageSize, int maxHeaderListSize,
-        long keepAliveTimeNanos, long keepAliveTimeoutNanos, boolean keepAliveWithoutCalls) {
+        long keepAliveTimeNanos, long keepAliveTimeoutNanos, boolean keepAliveWithoutCalls,
+        TransportTracer transportTracer) {
       this.channelType = channelType;
       this.negotiationType = negotiationType;
       this.channelOptions = new HashMap<ChannelOption<?>, Object>(channelOptions);
+      this.transportTracer = transportTracer;
 
       if (transportCreationParamsFilterFactory == null) {
         transportCreationParamsFilterFactory =
@@ -499,7 +510,7 @@ public final class NettyChannelBuilder
           dparams.getProtocolNegotiator(), flowControlWindow,
           maxMessageSize, maxHeaderListSize, keepAliveTimeNanosState.get(), keepAliveTimeoutNanos,
           keepAliveWithoutCalls, dparams.getAuthority(), dparams.getUserAgent(),
-          tooManyPingsRunnable);
+          tooManyPingsRunnable, transportTracer);
       return transport;
     }
 

netty/src/main/java/io/grpc/netty/NettyClientTransport.java

@@ -86,7 +86,7 @@ class NettyClientTransport implements ConnectionClientTransport {
   /** Since not thread-safe, may only be used from event loop. */
   private ClientTransportLifecycleManager lifecycleManager;
   /** Since not thread-safe, may only be used from event loop. */
-  private final TransportTracer transportTracer = new TransportTracer();
+  private final TransportTracer transportTracer;
 
   NettyClientTransport(
       SocketAddress address, Class<? extends Channel> channelType,
@@ -94,7 +94,7 @@ class NettyClientTransport implements ConnectionClientTransport {
       ProtocolNegotiator negotiator, int flowControlWindow, int maxMessageSize,
       int maxHeaderListSize, long keepAliveTimeNanos, long keepAliveTimeoutNanos,
       boolean keepAliveWithoutCalls, String authority, @Nullable String userAgent,
-      Runnable tooManyPingsRunnable) {
+      Runnable tooManyPingsRunnable, TransportTracer transportTracer) {
     this.negotiator = Preconditions.checkNotNull(negotiator, "negotiator");
     this.address = Preconditions.checkNotNull(address, "address");
     this.group = Preconditions.checkNotNull(group, "group");
@@ -110,6 +110,7 @@ class NettyClientTransport implements ConnectionClientTransport {
     this.userAgent = new AsciiString(GrpcUtil.getGrpcUserAgent("netty", userAgent));
     this.tooManyPingsRunnable =
         Preconditions.checkNotNull(tooManyPingsRunnable, "tooManyPingsRunnable");
+    this.transportTracer = Preconditions.checkNotNull(transportTracer, "transportTracer");
   }
 
   @Override

netty/src/main/java/io/grpc/netty/NettyServer.java

@@ -26,6 +26,7 @@ import io.grpc.internal.InternalServer;
 import io.grpc.internal.ServerListener;
 import io.grpc.internal.ServerTransportListener;
 import io.grpc.internal.SharedResourceHolder;
+import io.grpc.internal.TransportTracer;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
@@ -72,11 +73,13 @@ class NettyServer implements InternalServer {
   private final long permitKeepAliveTimeInNanos;
   private final ReferenceCounted eventLoopReferenceCounter = new EventLoopReferenceCounter();
   private final List<ServerStreamTracer.Factory> streamTracerFactories;
+  private final TransportTracer.Factory transportTracerFactory;
 
   NettyServer(
       SocketAddress address, Class<? extends ServerChannel> channelType,
       @Nullable EventLoopGroup bossGroup, @Nullable EventLoopGroup workerGroup,
       ProtocolNegotiator protocolNegotiator, List<ServerStreamTracer.Factory> streamTracerFactories,
+      TransportTracer.Factory transportTracerFactory,
       int maxStreamsPerConnection, int flowControlWindow, int maxMessageSize, int maxHeaderListSize,
       long keepAliveTimeInNanos, long keepAliveTimeoutInNanos,
       long maxConnectionIdleInNanos,
@@ -90,6 +93,7 @@ class NettyServer implements InternalServer {
     this.streamTracerFactories = checkNotNull(streamTracerFactories, "streamTracerFactories");
     this.usingSharedBossGroup = bossGroup == null;
     this.usingSharedWorkerGroup = workerGroup == null;
+    this.transportTracerFactory = transportTracerFactory;
     this.maxStreamsPerConnection = maxStreamsPerConnection;
     this.flowControlWindow = flowControlWindow;
     this.maxMessageSize = maxMessageSize;
@@ -142,7 +146,8 @@ class NettyServer implements InternalServer {
 
         NettyServerTransport transport =
             new NettyServerTransport(
-                ch, protocolNegotiator, streamTracerFactories, maxStreamsPerConnection,
+                ch, protocolNegotiator, streamTracerFactories, transportTracerFactory.create(),
+                maxStreamsPerConnection,
                 flowControlWindow, maxMessageSize, maxHeaderListSize,
                 keepAliveTimeInNanos, keepAliveTimeoutInNanos,
                 maxConnectionIdleInNanos,

netty/src/main/java/io/grpc/netty/NettyServerBuilder.java

@@ -22,6 +22,7 @@ import static io.grpc.internal.GrpcUtil.DEFAULT_SERVER_KEEPALIVE_TIMEOUT_NANOS;
 import static io.grpc.internal.GrpcUtil.DEFAULT_SERVER_KEEPALIVE_TIME_NANOS;
 import static io.grpc.internal.GrpcUtil.SERVER_KEEPALIVE_TIME_NANOS_DISABLED;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.grpc.ExperimentalApi;
@@ -30,6 +31,7 @@ import io.grpc.ServerStreamTracer;
 import io.grpc.internal.AbstractServerImplBuilder;
 import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.KeepAliveManager;
+import io.grpc.internal.TransportTracer;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.ServerChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
@@ -204,6 +206,12 @@ public final class NettyServerBuilder extends AbstractServerImplBuilder<NettySer
     super.setStatsEnabled(value);
   }
 
+  @VisibleForTesting
+  NettyServerBuilder setTransportTracerFactory(TransportTracer.Factory transportTracerFactory) {
+    this.transportTracerFactory = transportTracerFactory;
+    return this;
+  }
+
   /**
    * The maximum number of concurrent calls permitted for each incoming connection. Defaults to no
    * limit.
@@ -391,7 +399,8 @@ public final class NettyServerBuilder extends AbstractServerImplBuilder<NettySer
 
     return new NettyServer(
         address, channelType, bossEventLoopGroup, workerEventLoopGroup,
-        negotiator, streamTracerFactories, maxConcurrentCallsPerConnection, flowControlWindow,
+        negotiator, streamTracerFactories, transportTracerFactory,
+        maxConcurrentCallsPerConnection, flowControlWindow,
         maxMessageSize, maxHeaderListSize, keepAliveTimeInNanos, keepAliveTimeoutInNanos,
         maxConnectionIdleInNanos,
         maxConnectionAgeInNanos, maxConnectionAgeGraceInNanos,

netty/src/main/java/io/grpc/netty/NettyServerTransport.java

@@ -73,7 +73,8 @@ class NettyServerTransport implements ServerTransport {
 
   NettyServerTransport(
       Channel channel, ProtocolNegotiator protocolNegotiator,
-      List<ServerStreamTracer.Factory> streamTracerFactories, int maxStreams,
+      List<ServerStreamTracer.Factory> streamTracerFactories,
+      TransportTracer transportTracer, int maxStreams,
       int flowControlWindow, int maxMessageSize, int maxHeaderListSize,
       long keepAliveTimeInNanos, long keepAliveTimeoutInNanos,
       long maxConnectionIdleInNanos,
@@ -83,6 +84,7 @@ class NettyServerTransport implements ServerTransport {
     this.protocolNegotiator = Preconditions.checkNotNull(protocolNegotiator, "protocolNegotiator");
     this.streamTracerFactories =
         Preconditions.checkNotNull(streamTracerFactories, "streamTracerFactories");
+    this.transportTracer = Preconditions.checkNotNull(transportTracer, "transportTracer");
     this.maxStreams = maxStreams;
     this.flowControlWindow = flowControlWindow;
     this.maxMessageSize = maxMessageSize;
@@ -94,7 +96,6 @@ class NettyServerTransport implements ServerTransport {
     this.maxConnectionAgeGraceInNanos = maxConnectionAgeGraceInNanos;
     this.permitKeepAliveWithoutCalls = permitKeepAliveWithoutCalls;
     this.permitKeepAliveTimeInNanos = permitKeepAliveTimeInNanos;
-    this.transportTracer = new TransportTracer();
   }
 
   public void start(ServerTransportListener listener) {

netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java

@@ -58,6 +58,7 @@ import io.grpc.internal.ServerStream;
 import io.grpc.internal.ServerStreamListener;
 import io.grpc.internal.ServerTransport;
 import io.grpc.internal.ServerTransportListener;
+import io.grpc.internal.TransportTracer;
 import io.grpc.internal.testing.TestUtils;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelOption;
@@ -170,7 +171,7 @@ public class NettyClientTransportTest {
         address, NioSocketChannel.class, channelOptions, group, newNegotiator(),
         DEFAULT_WINDOW_SIZE, DEFAULT_MAX_MESSAGE_SIZE, GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE,
         KEEPALIVE_TIME_NANOS_DISABLED, 1L, false, authority, null /* user agent */,
-        tooManyPingsRunnable);
+        tooManyPingsRunnable, new TransportTracer());
     transports.add(transport);
     callMeMaybe(transport.start(clientTransportListener));
 
@@ -374,7 +375,7 @@ public class NettyClientTransportTest {
         address, CantConstructChannel.class, new HashMap<ChannelOption<?>, Object>(), group,
         newNegotiator(), DEFAULT_WINDOW_SIZE, DEFAULT_MAX_MESSAGE_SIZE,
         GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE, KEEPALIVE_TIME_NANOS_DISABLED, 1, false, authority,
-        null, tooManyPingsRunnable);
+        null, tooManyPingsRunnable, new TransportTracer());
     transports.add(transport);
 
     // Should not throw
@@ -542,7 +543,8 @@ public class NettyClientTransportTest {
         address, NioSocketChannel.class, new HashMap<ChannelOption<?>, Object>(), group, negotiator,
         DEFAULT_WINDOW_SIZE, maxMsgSize, maxHeaderListSize,
         keepAliveTimeNano, keepAliveTimeoutNano,
-        false, authority, userAgent, tooManyPingsRunnable);
+        false, authority, userAgent, tooManyPingsRunnable,
+        new TransportTracer());
     transports.add(transport);
     return transport;
   }
@@ -555,7 +557,9 @@ public class NettyClientTransportTest {
     server = new NettyServer(
         TestUtils.testServerAddress(0),
         NioServerSocketChannel.class, group, group, negotiator,
-        Collections.<ServerStreamTracer.Factory>emptyList(), maxStreamsPerConnection,
+        Collections.<ServerStreamTracer.Factory>emptyList(),
+        TransportTracer.getDefaultFactory(),
+        maxStreamsPerConnection,
         DEFAULT_WINDOW_SIZE, DEFAULT_MAX_MESSAGE_SIZE, maxHeaderListSize,
         DEFAULT_SERVER_KEEPALIVE_TIME_NANOS, DEFAULT_SERVER_KEEPALIVE_TIMEOUT_NANOS,
         MAX_CONNECTION_IDLE_NANOS_DISABLED,

netty/src/test/java/io/grpc/netty/NettyServerTest.java

@@ -22,6 +22,7 @@ import io.grpc.ServerStreamTracer;
 import io.grpc.internal.ServerListener;
 import io.grpc.internal.ServerTransport;
 import io.grpc.internal.ServerTransportListener;
+import io.grpc.internal.TransportTracer;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import java.net.InetSocketAddress;
 import java.util.Collections;
@@ -42,6 +43,7 @@ public class NettyServerTest {
         null, // no event group
         new ProtocolNegotiators.PlaintextNegotiator(),
         Collections.<ServerStreamTracer.Factory>emptyList(),
+        TransportTracer.getDefaultFactory(),
         1, // ignore
         1, // ignore
         1, // ignore
@@ -77,6 +79,7 @@ public class NettyServerTest {
         null, // no event group
         new ProtocolNegotiators.PlaintextNegotiator(),
         Collections.<ServerStreamTracer.Factory>emptyList(),
+        TransportTracer.getDefaultFactory(),
         1, // ignore
         1, // ignore
         1, // ignore

netty/src/test/java/io/grpc/netty/NettyTransportTest.java

@@ -18,11 +18,14 @@ package io.grpc.netty;
 
 import io.grpc.ServerStreamTracer;
 import io.grpc.internal.ClientTransportFactory;
+import io.grpc.internal.FakeClock;
 import io.grpc.internal.InternalServer;
 import io.grpc.internal.ManagedClientTransport;
+import io.grpc.internal.TransportTracer;
 import io.grpc.internal.testing.AbstractTransportTest;
 import java.net.InetSocketAddress;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -32,14 +35,23 @@ import org.junit.runners.JUnit4;
 /** Unit tests for Netty transport. */
 @RunWith(JUnit4.class)
 public class NettyTransportTest extends AbstractTransportTest {
+  private final FakeClock fakeClock = new FakeClock();
+  private final TransportTracer.Factory fakeClockTransportTracer = new TransportTracer.Factory(
+      new TransportTracer.TimeProvider() {
+        @Override
+        public long currentTimeMillis() {
+          return fakeClock.currentTimeMillis();
+        }
+      });
   // Avoid LocalChannel for testing because LocalChannel can fail with
   // io.netty.channel.ChannelException instead of java.net.ConnectException which breaks
   // serverNotListening test.
-  private ClientTransportFactory clientFactory = NettyChannelBuilder
+  private final ClientTransportFactory clientFactory = NettyChannelBuilder
       // Although specified here, address is ignored because we never call build.
       .forAddress("localhost", 0)
       .flowControlWindow(65 * 1024)
       .negotiationType(NegotiationType.PLAINTEXT)
+      .setTransportTracerFactory(fakeClockTransportTracer)
       .buildTransportFactory();
 
   @Override
@@ -57,6 +69,7 @@ public class NettyTransportTest extends AbstractTransportTest {
     return NettyServerBuilder
         .forPort(0)
         .flowControlWindow(65 * 1024)
+        .setTransportTracerFactory(fakeClockTransportTracer)
         .buildTransportServer(streamTracerFactories);
   }
 
@@ -67,6 +80,7 @@ public class NettyTransportTest extends AbstractTransportTest {
     return NettyServerBuilder
         .forPort(port)
         .flowControlWindow(65 * 1024)
+        .setTransportTracerFactory(fakeClockTransportTracer)
         .buildTransportServer(streamTracerFactories);
   }
 
@@ -75,6 +89,16 @@ public class NettyTransportTest extends AbstractTransportTest {
     return "localhost:" + server.getPort();
   }
 
+  @Override
+  protected void advanceClock(long offset, TimeUnit unit) {
+    fakeClock.forwardNanos(unit.toNanos(offset));
+  }
+
+  @Override
+  protected long currentTimeMillis() {
+    return fakeClock.currentTimeMillis();
+  }
+
   @Override
   protected ManagedClientTransport newClientTransport(InternalServer server) {
     int port = server.getPort();

testing/src/main/java/io/grpc/internal/testing/AbstractTransportTest.java

@@ -190,6 +190,21 @@ public abstract class AbstractTransportTest {
     }
   }
 
+  /**
+   * Moves the clock forward, for tests that require moving the clock forward. It is the transport
+   * subclass's responsibility to implement this method.
+   */
+  protected void advanceClock(long offset, TimeUnit unit) {
+    throw new UnsupportedOperationException();
+  }
+
+  /**
+   * Returns the current time, for tests that rely on the clock.
+   */
+  protected long currentTimeMillis() {
+    throw new UnsupportedOperationException();
+  }
+
   // TODO(ejona):
   //   multiple streams on same transport
   //   multiple client transports to same server
@@ -1409,23 +1424,23 @@ public abstract class AbstractTransportTest {
           serverTransportListener.transport.getTransportStats().get();
       assertEquals(1, serverAfter.streamsStarted);
       serverFirstTimestampNanos = serverAfter.lastStreamCreatedTimeNanos;
-      assertThat(System.currentTimeMillis()
+      assertEquals(
-          - TimeUnit.NANOSECONDS.toMillis(serverAfter.lastStreamCreatedTimeNanos)).isAtMost(50L);
+          currentTimeMillis(),
+          TimeUnit.NANOSECONDS.toMillis(serverAfter.lastStreamCreatedTimeNanos));
 
       TransportTracer.Stats clientAfter = client.getTransportStats().get();
       assertEquals(1, clientAfter.streamsStarted);
       clientFirstTimestampNanos = clientAfter.lastStreamCreatedTimeNanos;
-      assertThat(System.currentTimeMillis()
+      assertEquals(
-          - TimeUnit.NANOSECONDS.toMillis(clientFirstTimestampNanos)).isAtMost(50L);
+          currentTimeMillis(),
+          TimeUnit.NANOSECONDS.toMillis(clientFirstTimestampNanos));
 
       ServerStream serverStream = serverStreamCreation.stream;
       serverStream.close(Status.OK, new Metadata());
     }
 
-    // lastStreamCreatedTimeNanos is converted from the system milli clock. Sleep a bit to ensure
+    final long elapsedMillis = 100;
-    // it has moved forward in time.
+    advanceClock(100, TimeUnit.MILLISECONDS);
-    // TODO(zpencer): plumb in a fake clock instead
-    Thread.sleep(5);
 
     // start second stream
     {
@@ -1444,17 +1459,21 @@ public abstract class AbstractTransportTest {
       TransportTracer.Stats serverAfter =
           serverTransportListener.transport.getTransportStats().get();
       assertEquals(2, serverAfter.streamsStarted);
-      assertTrue(serverAfter.lastStreamCreatedTimeNanos > serverFirstTimestampNanos);
+      assertEquals(
+          TimeUnit.MILLISECONDS.toNanos(elapsedMillis),
+          serverAfter.lastStreamCreatedTimeNanos - serverFirstTimestampNanos);
       long serverSecondTimestamp =
           TimeUnit.NANOSECONDS.toMillis(serverAfter.lastStreamCreatedTimeNanos);
-      assertThat(System.currentTimeMillis() - serverSecondTimestamp).isAtMost(50L);
+      assertEquals(currentTimeMillis(), serverSecondTimestamp);
 
       TransportTracer.Stats clientAfter = client.getTransportStats().get();
       assertEquals(2, clientAfter.streamsStarted);
-      assertTrue(clientAfter.lastStreamCreatedTimeNanos > clientFirstTimestampNanos);
+      assertEquals(
+          TimeUnit.MILLISECONDS.toNanos(elapsedMillis),
+          clientAfter.lastStreamCreatedTimeNanos - clientFirstTimestampNanos);
       long clientSecondTimestamp =
           TimeUnit.NANOSECONDS.toMillis(clientAfter.lastStreamCreatedTimeNanos);
-      assertThat(System.currentTimeMillis() - clientSecondTimestamp).isAtMost(50L);
+      assertEquals(currentTimeMillis(), clientSecondTimestamp);
 
       ServerStream serverStream = serverStreamCreation.stream;
       serverStream.close(Status.OK, new Metadata());
@@ -1618,12 +1637,12 @@ public abstract class AbstractTransportTest {
     assertEquals(1, serverAfter.messagesReceived);
     long serverTimestamp =
         TimeUnit.NANOSECONDS.toMillis(serverAfter.lastMessageReceivedTimeNanos);
-    assertThat(System.currentTimeMillis() - serverTimestamp).isAtMost(50L);
+    assertEquals(currentTimeMillis(), serverTimestamp);
     TransportTracer.Stats clientAfter = client.getTransportStats().get();
     assertEquals(1, clientAfter.messagesSent);
     long clientTimestamp =
         TimeUnit.NANOSECONDS.toMillis(clientAfter.lastMessageSentTimeNanos);
-    assertThat(System.currentTimeMillis() - clientTimestamp).isAtMost(50L);
+    assertEquals(currentTimeMillis(), clientTimestamp);
 
     serverStream.close(Status.OK, new Metadata());
   }
@@ -1663,12 +1682,13 @@ public abstract class AbstractTransportTest {
         serverTransportListener.transport.getTransportStats().get();
     assertEquals(1, serverAfter.messagesSent);
     long serverTimestmap = TimeUnit.NANOSECONDS.toMillis(serverAfter.lastMessageSentTimeNanos);
-    assertThat(System.currentTimeMillis() - serverTimestmap).isAtMost(50L);
+    assertEquals(currentTimeMillis(), serverTimestmap);
     TransportTracer.Stats clientAfter = client.getTransportStats().get();
     assertEquals(1, clientAfter.messagesReceived);
     long clientTimestmap =
         TimeUnit.NANOSECONDS.toMillis(clientAfter.lastMessageReceivedTimeNanos);
-    assertThat(System.currentTimeMillis() - clientTimestmap).isAtMost(50L);
+    assertEquals(currentTimeMillis(), clientTimestmap);
+
 
     serverStream.close(Status.OK, new Metadata());
   }