okhttp: make flow control window size configurable (#4959)

Make flow control window size in OkHttp configurable.

okhttp/src/main/java/io/grpc/okhttp/OkHttpChannelBuilder.java

@@ -61,6 +61,8 @@ import javax.net.ssl.TrustManagerFactory;
 public class OkHttpChannelBuilder extends
         AbstractManagedChannelImplBuilder<OkHttpChannelBuilder> {
 
+  public static final int DEFAULT_FLOW_CONTROL_WINDOW = 65535;
+
   /** Identifies the negotiation used for starting up HTTP/2. */
   private enum NegotiationType {
     /** Uses TLS ALPN/NPN negotiation, assumes an SSL connection. */
@@ -157,6 +159,7 @@ public class OkHttpChannelBuilder extends
   private NegotiationType negotiationType = NegotiationType.TLS;
   private long keepAliveTimeNanos = KEEPALIVE_TIME_NANOS_DISABLED;
   private long keepAliveTimeoutNanos = DEFAULT_KEEPALIVE_TIMEOUT_NANOS;
+  private int flowControlWindow = DEFAULT_FLOW_CONTROL_WINDOW;
   private boolean keepAliveWithoutCalls;
 
   protected OkHttpChannelBuilder(String host, int port) {
@@ -273,6 +276,16 @@ public class OkHttpChannelBuilder extends
     return this;
   }
 
+  /**
+   * Sets the flow control window in bytes. If not called, the default value
+   * is {@link #DEFAULT_FLOW_CONTROL_WINDOW}).
+   */
+  public OkHttpChannelBuilder flowControlWindow(int flowControlWindow) {
+    Preconditions.checkState(flowControlWindow > 0, "flowControlWindow must be positive");
+    this.flowControlWindow = flowControlWindow;
+    return this;
+  }
+
   /**
    * {@inheritDoc}
    *
@@ -398,8 +411,8 @@ public class OkHttpChannelBuilder extends
     boolean enableKeepAlive = keepAliveTimeNanos != KEEPALIVE_TIME_NANOS_DISABLED;
     return new OkHttpTransportFactory(transportExecutor, scheduledExecutorService,
         createSocketFactory(), hostnameVerifier, connectionSpec, maxInboundMessageSize(),
-        enableKeepAlive, keepAliveTimeNanos, keepAliveTimeoutNanos, keepAliveWithoutCalls,
+        enableKeepAlive, keepAliveTimeNanos, keepAliveTimeoutNanos, flowControlWindow,
-        transportTracerFactory);
+        keepAliveWithoutCalls, transportTracerFactory);
   }
 
   @Override
@@ -476,6 +489,7 @@ public class OkHttpChannelBuilder extends
     private final boolean enableKeepAlive;
     private final AtomicBackoff keepAliveTimeNanos;
     private final long keepAliveTimeoutNanos;
+    private final int flowControlWindow;
     private final boolean keepAliveWithoutCalls;
     private final ScheduledExecutorService timeoutService;
     private boolean closed;
@@ -489,6 +503,7 @@ public class OkHttpChannelBuilder extends
         boolean enableKeepAlive,
         long keepAliveTimeNanos,
         long keepAliveTimeoutNanos,
+        int flowControlWindow,
         boolean keepAliveWithoutCalls,
         TransportTracer.Factory transportTracerFactory) {
       usingSharedScheduler = timeoutService == null;
@@ -501,6 +516,7 @@ public class OkHttpChannelBuilder extends
       this.enableKeepAlive = enableKeepAlive;
       this.keepAliveTimeNanos = new AtomicBackoff("keepalive time nanos", keepAliveTimeNanos);
       this.keepAliveTimeoutNanos = keepAliveTimeoutNanos;
+      this.flowControlWindow = flowControlWindow;
       this.keepAliveWithoutCalls = keepAliveWithoutCalls;
 
       usingSharedExecutor = executor == null;
@@ -537,6 +553,7 @@ public class OkHttpChannelBuilder extends
           hostnameVerifier,
           connectionSpec,
           maxMessageSize,
+          flowControlWindow,
           options.getProxyParameters(),
           tooManyPingsRunnable,
           transportTracerFactory.create());

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java

@@ -44,8 +44,6 @@ import okio.Buffer;
  */
 class OkHttpClientStream extends AbstractClientStream {
 
-  private static final int WINDOW_UPDATE_THRESHOLD = Utils.DEFAULT_WINDOW_SIZE / 2;
-
   private static final Buffer EMPTY_BUFFER = new Buffer();
 
   public static final int ABSENT_ID = -1;
@@ -71,6 +69,7 @@ class OkHttpClientStream extends AbstractClientStream {
       OutboundFlowController outboundFlow,
       Object lock,
       int maxMessageSize,
+      int initialWindowSize,
       String authority,
       String userAgent,
       StatsTraceContext statsTraceCtx,
@@ -91,8 +90,15 @@ class OkHttpClientStream extends AbstractClientStream {
     // so it is safe to read the transport attributes.
     // We make a copy here for convenience, even though we can ask the transport.
     this.attributes = transport.getAttributes();
-    this.state = new TransportState(maxMessageSize, statsTraceCtx, lock, frameWriter, outboundFlow,
+    this.state =
-        transport);
+        new TransportState(
+            maxMessageSize,
+            statsTraceCtx,
+            lock,
+            frameWriter,
+            outboundFlow,
+            transport,
+            initialWindowSize);
   }
 
   @Override
@@ -184,6 +190,7 @@ class OkHttpClientStream extends AbstractClientStream {
   }
 
   class TransportState extends Http2ClientStreamTransportState {
+    private final int initialWindowSize;
     private final Object lock;
     @GuardedBy("lock")
     private List<Header> requestHeaders;
@@ -196,9 +203,9 @@ class OkHttpClientStream extends AbstractClientStream {
     @GuardedBy("lock")
     private boolean cancelSent = false;
     @GuardedBy("lock")
-    private int window = Utils.DEFAULT_WINDOW_SIZE;
+    private int window;
     @GuardedBy("lock")
-    private int processedWindow = Utils.DEFAULT_WINDOW_SIZE;
+    private int processedWindow;
     @GuardedBy("lock")
     private final AsyncFrameWriter frameWriter;
     @GuardedBy("lock")
@@ -212,12 +219,16 @@ class OkHttpClientStream extends AbstractClientStream {
         Object lock,
         AsyncFrameWriter frameWriter,
         OutboundFlowController outboundFlow,
-        OkHttpClientTransport transport) {
+        OkHttpClientTransport transport,
+        int initialWindowSize) {
       super(maxMessageSize, statsTraceCtx, OkHttpClientStream.this.getTransportTracer());
       this.lock = checkNotNull(lock, "lock");
       this.frameWriter = frameWriter;
       this.outboundFlow = outboundFlow;
       this.transport = transport;
+      this.window = initialWindowSize;
+      this.processedWindow = initialWindowSize;
+      this.initialWindowSize = initialWindowSize;
     }
 
     @GuardedBy("lock")
@@ -270,8 +281,8 @@ class OkHttpClientStream extends AbstractClientStream {
     @GuardedBy("lock")
     public void bytesRead(int processedBytes) {
       processedWindow -= processedBytes;
-      if (processedWindow <= WINDOW_UPDATE_THRESHOLD) {
+      if (processedWindow <= initialWindowSize * Utils.DEFAULT_WINDOW_UPDATE_RATIO) {
-        int delta = Utils.DEFAULT_WINDOW_SIZE - processedWindow;
+        int delta = initialWindowSize - processedWindow;
         window += delta;
         processedWindow += delta;
         frameWriter.windowUpdate(id(), delta);

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java

@@ -18,6 +18,7 @@ package io.grpc.okhttp;
 
 import static com.google.common.base.Preconditions.checkState;
 import static io.grpc.internal.GrpcUtil.TIMER_SERVICE;
+import static io.grpc.okhttp.Utils.DEFAULT_WINDOW_UPDATE_RATIO;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
@@ -140,6 +141,7 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
   private final Random random = new Random();
   // Returns new unstarted stopwatches
   private final Supplier<Stopwatch> stopwatchFactory;
+  private final int initialWindowSize;
   private Listener listener;
   private FrameReader testFrameReader;
   private AsyncFrameWriter frameWriter;
@@ -220,11 +222,12 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
   OkHttpClientTransport(InetSocketAddress address, String authority, @Nullable String userAgent,
       Executor executor, @Nullable SSLSocketFactory sslSocketFactory,
       @Nullable HostnameVerifier hostnameVerifier, ConnectionSpec connectionSpec,
-      int maxMessageSize, @Nullable ProxyParameters proxy, Runnable tooManyPingsRunnable,
+      int maxMessageSize, int initialWindowSize, @Nullable ProxyParameters proxy,
-      TransportTracer transportTracer) {
+      Runnable tooManyPingsRunnable, TransportTracer transportTracer) {
     this.address = Preconditions.checkNotNull(address, "address");
     this.defaultAuthority = authority;
     this.maxMessageSize = maxMessageSize;
+    this.initialWindowSize = initialWindowSize;
     this.executor = Preconditions.checkNotNull(executor, "executor");
     serializingExecutor = new SerializingExecutor(executor);
     // Client initiated streams are odd, server initiated ones are even. Server should not need to
@@ -257,10 +260,12 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
       @Nullable Runnable connectingCallback,
       SettableFuture<Void> connectedFuture,
       int maxMessageSize,
+      int initialWindowSize,
       Runnable tooManyPingsRunnable,
       TransportTracer transportTracer) {
     address = null;
     this.maxMessageSize = maxMessageSize;
+    this.initialWindowSize = initialWindowSize;
     defaultAuthority = "notarealauthority:80";
     this.userAgent = GrpcUtil.getGrpcUserAgent("okhttp", userAgent);
     this.executor = Preconditions.checkNotNull(executor, "executor");
@@ -358,6 +363,7 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
         outboundFlow,
         lock,
         maxMessageSize,
+        initialWindowSize,
         defaultAuthority,
         userAgent,
         statsTraceCtx,
@@ -437,7 +443,7 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
     }
 
     frameWriter = new AsyncFrameWriter(this, serializingExecutor);
-    outboundFlow = new OutboundFlowController(this, frameWriter);
+    outboundFlow = new OutboundFlowController(this, frameWriter, initialWindowSize);
     // Connecting in the serializingExecutor, so that some stream operations like synStream
     // will be executed after connected.
     serializingExecutor.execute(new Runnable() {
@@ -1044,7 +1050,7 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
 
       // connection window update
       connectionUnacknowledgedBytesRead += length;
-      if (connectionUnacknowledgedBytesRead >= Utils.DEFAULT_WINDOW_SIZE / 2) {
+      if (connectionUnacknowledgedBytesRead >= initialWindowSize * DEFAULT_WINDOW_UPDATE_RATIO) {
         frameWriter.windowUpdate(0, connectionUnacknowledgedBytesRead);
         connectionUnacknowledgedBytesRead = 0;
       }

okhttp/src/main/java/io/grpc/okhttp/OutboundFlowController.java

@@ -17,7 +17,6 @@
 package io.grpc.okhttp;
 
 import static io.grpc.okhttp.Utils.CONNECTION_STREAM_ID;
-import static io.grpc.okhttp.Utils.DEFAULT_WINDOW_SIZE;
 import static java.lang.Math.ceil;
 import static java.lang.Math.max;
 import static java.lang.Math.min;
@@ -37,12 +36,15 @@ import okio.Buffer;
 class OutboundFlowController {
   private final OkHttpClientTransport transport;
   private final FrameWriter frameWriter;
-  private int initialWindowSize = DEFAULT_WINDOW_SIZE;
+  private int initialWindowSize;
-  private final OutboundFlowState connectionState = new OutboundFlowState(CONNECTION_STREAM_ID);
+  private final OutboundFlowState connectionState;
 
-  OutboundFlowController(OkHttpClientTransport transport, FrameWriter frameWriter) {
+  OutboundFlowController(
+      OkHttpClientTransport transport, FrameWriter frameWriter, int initialWindowSize) {
     this.transport = Preconditions.checkNotNull(transport, "transport");
     this.frameWriter = Preconditions.checkNotNull(frameWriter, "frameWriter");
+    this.initialWindowSize = initialWindowSize;
+    connectionState = new OutboundFlowState(CONNECTION_STREAM_ID, initialWindowSize);
   }
 
   /**
@@ -65,7 +67,7 @@ class OutboundFlowController {
       OutboundFlowState state = (OutboundFlowState) stream.getOutboundFlowState();
       if (state == null) {
         // Create the OutboundFlowState with the new window size.
-        state = new OutboundFlowState(stream);
+        state = new OutboundFlowState(stream, initialWindowSize);
         stream.setOutboundFlowState(state);
       } else {
         state.incrementStreamWindow(delta);
@@ -158,7 +160,7 @@ class OutboundFlowController {
   private OutboundFlowState state(OkHttpClientStream stream) {
     OutboundFlowState state = (OutboundFlowState) stream.getOutboundFlowState();
     if (state == null) {
-      state = new OutboundFlowState(stream);
+      state = new OutboundFlowState(stream, initialWindowSize);
       stream.setOutboundFlowState(state);
     }
     return state;
@@ -229,17 +231,18 @@ class OutboundFlowController {
     final Queue<Frame> pendingWriteQueue;
     final int streamId;
     int queuedBytes;
-    int window = initialWindowSize;
+    int window;
     int allocatedBytes;
     OkHttpClientStream stream;
 
-    OutboundFlowState(int streamId) {
+    OutboundFlowState(int streamId, int initialWindowSize) {
       this.streamId = streamId;
-      pendingWriteQueue = new ArrayDeque<Frame>(2);
+      window = initialWindowSize;
+      pendingWriteQueue = new ArrayDeque<>(2);
     }
 
-    OutboundFlowState(OkHttpClientStream stream) {
+    OutboundFlowState(OkHttpClientStream stream, int initialWindowSize) {
-      this(stream.id());
+      this(stream.id(), initialWindowSize);
       this.stream = stream;
     }
 

okhttp/src/main/java/io/grpc/okhttp/Utils.java

@@ -36,7 +36,11 @@ import java.util.logging.Logger;
 class Utils {
   private static final Logger log = Logger.getLogger(Utils.class.getName());
 
-  static final int DEFAULT_WINDOW_SIZE = 65535;
+  /**
+   * The default ratio of window size to initial window size below which a {@code WINDOW_UPDATE}
+   * is sent to expand the window.
+   */
+  static final float DEFAULT_WINDOW_UPDATE_RATIO = 0.5f;
   static final int CONNECTION_STREAM_ID = 0;
 
   public static Metadata convertHeaders(List<Header> http2Headers) {

okhttp/src/test/java/io/grpc/okhttp/OkHttpClientStreamTest.java

@@ -57,6 +57,7 @@ import org.mockito.stubbing.Answer;
 @RunWith(JUnit4.class)
 public class OkHttpClientStreamTest {
   private static final int MAX_MESSAGE_SIZE = 100;
+  private static final int INITIAL_WINDOW_SIZE = 65535;
 
   @Mock private MethodDescriptor.Marshaller<Void> marshaller;
   @Mock private AsyncFrameWriter frameWriter;
@@ -88,6 +89,7 @@ public class OkHttpClientStreamTest {
         flowController,
         lock,
         MAX_MESSAGE_SIZE,
+        INITIAL_WINDOW_SIZE,
         "localhost",
         "userAgent",
         StatsTraceContext.NOOP,
@@ -150,8 +152,8 @@ public class OkHttpClientStreamTest {
     Metadata metaData = new Metadata();
     metaData.put(GrpcUtil.USER_AGENT_KEY, "misbehaving-application");
     stream = new OkHttpClientStream(methodDescriptor, metaData, frameWriter, transport,
-        flowController, lock, MAX_MESSAGE_SIZE, "localhost", "good-application",
+        flowController, lock, MAX_MESSAGE_SIZE, INITIAL_WINDOW_SIZE, "localhost",
-        StatsTraceContext.NOOP, transportTracer, CallOptions.DEFAULT);
+        "good-application", StatsTraceContext.NOOP, transportTracer, CallOptions.DEFAULT);
     stream.start(new BaseClientStreamListener());
     stream.transportState().start(3);
 
@@ -165,8 +167,8 @@ public class OkHttpClientStreamTest {
     Metadata metaData = new Metadata();
     metaData.put(GrpcUtil.USER_AGENT_KEY, "misbehaving-application");
     stream = new OkHttpClientStream(methodDescriptor, metaData, frameWriter, transport,
-        flowController, lock, MAX_MESSAGE_SIZE, "localhost", "good-application",
+        flowController, lock, MAX_MESSAGE_SIZE, INITIAL_WINDOW_SIZE, "localhost",
-        StatsTraceContext.NOOP, transportTracer, CallOptions.DEFAULT);
+        "good-application", StatsTraceContext.NOOP, transportTracer, CallOptions.DEFAULT);
     stream.start(new BaseClientStreamListener());
     stream.transportState().start(3);
 
@@ -193,8 +195,8 @@ public class OkHttpClientStreamTest {
         .setResponseMarshaller(marshaller)
         .build();
     stream = new OkHttpClientStream(getMethod, new Metadata(), frameWriter, transport,
-        flowController, lock, MAX_MESSAGE_SIZE, "localhost", "good-application",
+        flowController, lock, MAX_MESSAGE_SIZE, INITIAL_WINDOW_SIZE, "localhost",
-        StatsTraceContext.NOOP, transportTracer, CallOptions.DEFAULT);
+        "good-application", StatsTraceContext.NOOP, transportTracer, CallOptions.DEFAULT);
     stream.start(new BaseClientStreamListener());
 
     // GET streams send headers after halfClose is called.

okhttp/src/test/java/io/grpc/okhttp/OkHttpClientTransportTest.java

@@ -125,6 +125,7 @@ import org.mockito.MockitoAnnotations;
 @RunWith(JUnit4.class)
 public class OkHttpClientTransportTest {
   private static final int TIME_OUT_MS = 2000;
+  private static final int INITIAL_WINDOW_SIZE = 65535;
   private static final String NETWORK_ISSUE_MESSAGE = "network issue";
   private static final String ERROR_MESSAGE = "simulated error";
   // The gRPC header length, which includes 1 byte compression flag and 4 bytes message length.
@@ -174,21 +175,28 @@ public class OkHttpClientTransportTest {
   }
 
   private void initTransport() throws Exception {
-    startTransport(DEFAULT_START_STREAM_ID, null, true, DEFAULT_MAX_MESSAGE_SIZE, null);
+    startTransport(
+        DEFAULT_START_STREAM_ID, null, true, DEFAULT_MAX_MESSAGE_SIZE, INITIAL_WINDOW_SIZE, null);
   }
 
   private void initTransport(int startId) throws Exception {
-    startTransport(startId, null, true, DEFAULT_MAX_MESSAGE_SIZE, null);
+    startTransport(startId, null, true, DEFAULT_MAX_MESSAGE_SIZE, INITIAL_WINDOW_SIZE, null);
   }
 
   private void initTransportAndDelayConnected() throws Exception {
     delayConnectedCallback = new DelayConnectedCallback();
     startTransport(
-        DEFAULT_START_STREAM_ID, delayConnectedCallback, false, DEFAULT_MAX_MESSAGE_SIZE, null);
+        DEFAULT_START_STREAM_ID,
+        delayConnectedCallback,
+        false,
+        DEFAULT_MAX_MESSAGE_SIZE,
+        INITIAL_WINDOW_SIZE,
+        null);
   }
 
   private void startTransport(int startId, @Nullable Runnable connectingCallback,
-      boolean waitingForConnected, int maxMessageSize, String userAgent) throws Exception {
+      boolean waitingForConnected, int maxMessageSize, int initialWindowSize, String userAgent)
+      throws Exception {
     connectedFuture = SettableFuture.create();
     final Ticker ticker = new Ticker() {
       @Override
@@ -213,6 +221,7 @@ public class OkHttpClientTransportTest {
         connectingCallback,
         connectedFuture,
         maxMessageSize,
+        initialWindowSize,
         tooManyPingsRunnable,
         new TransportTracer());
     clientTransport.start(transportListener);
@@ -233,6 +242,7 @@ public class OkHttpClientTransportTest {
         hostnameVerifier,
         OkHttpChannelBuilder.INTERNAL_DEFAULT_CONNECTION_SPEC,
         DEFAULT_MAX_MESSAGE_SIZE,
+        INITIAL_WINDOW_SIZE,
         NO_PROXY,
         tooManyPingsRunnable,
         transportTracer);
@@ -244,7 +254,7 @@ public class OkHttpClientTransportTest {
   @Test
   public void maxMessageSizeShouldBeEnforced() throws Exception {
     // Allow the response payloads of up to 1 byte.
-    startTransport(3, null, true, 1, null);
+    startTransport(3, null, true, 1, INITIAL_WINDOW_SIZE, null);
 
     MockStreamListener listener = new MockStreamListener();
     OkHttpClientStream stream =
@@ -502,7 +512,7 @@ public class OkHttpClientTransportTest {
 
   @Test
   public void overrideDefaultUserAgent() throws Exception {
-    startTransport(3, null, true, DEFAULT_MAX_MESSAGE_SIZE, "fakeUserAgent");
+    startTransport(3, null, true, DEFAULT_MAX_MESSAGE_SIZE, INITIAL_WINDOW_SIZE, "fakeUserAgent");
     MockStreamListener listener = new MockStreamListener();
     OkHttpClientStream stream =
         clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
@@ -557,7 +567,7 @@ public class OkHttpClientTransportTest {
   public void transportTracer_windowSizeDefault() throws Exception {
     initTransport();
     TransportStats stats = getTransportStats(clientTransport);
-    assertEquals(Utils.DEFAULT_WINDOW_SIZE, stats.remoteFlowControlWindow);
+    assertEquals(INITIAL_WINDOW_SIZE, stats.remoteFlowControlWindow);
     // okhttp does not track local window sizes
     assertEquals(-1, stats.localFlowControlWindow);
   }
@@ -566,13 +576,13 @@ public class OkHttpClientTransportTest {
   public void transportTracer_windowSize_remote() throws Exception {
     initTransport();
     TransportStats before = getTransportStats(clientTransport);
-    assertEquals(Utils.DEFAULT_WINDOW_SIZE, before.remoteFlowControlWindow);
+    assertEquals(INITIAL_WINDOW_SIZE, before.remoteFlowControlWindow);
     // okhttp does not track local window sizes
     assertEquals(-1, before.localFlowControlWindow);
 
     frameHandler().windowUpdate(0, 1000);
     TransportStats after = getTransportStats(clientTransport);
-    assertEquals(Utils.DEFAULT_WINDOW_SIZE + 1000, after.remoteFlowControlWindow);
+    assertEquals(INITIAL_WINDOW_SIZE + 1000, after.remoteFlowControlWindow);
     // okhttp does not track local window sizes
     assertEquals(-1, after.localFlowControlWindow);
   }
@@ -598,7 +608,7 @@ public class OkHttpClientTransportTest {
     frameHandler().headers(false, false, 3, 0, grpcResponseHeaders(), HeadersMode.HTTP_20_HEADERS);
     frameHandler().headers(false, false, 5, 0, grpcResponseHeaders(), HeadersMode.HTTP_20_HEADERS);
 
-    int messageLength = Utils.DEFAULT_WINDOW_SIZE / 4;
+    int messageLength = INITIAL_WINDOW_SIZE / 4;
     byte[] fakeMessage = new byte[messageLength];
 
     // Stream 1 receives a message
@@ -651,7 +661,7 @@ public class OkHttpClientTransportTest {
     OkHttpClientStream stream =
         clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
     stream.start(listener);
-    int messageLength = Utils.DEFAULT_WINDOW_SIZE / 2 + 1;
+    int messageLength = INITIAL_WINDOW_SIZE / 2 + 1;
     byte[] fakeMessage = new byte[messageLength];
 
     frameHandler().headers(false, false, 3, 0, grpcResponseHeaders(), HeadersMode.HTTP_20_HEADERS);
@@ -679,13 +689,18 @@ public class OkHttpClientTransportTest {
 
   @Test
   public void outboundFlowControl() throws Exception {
-    initTransport();
+    outboundFlowControl(INITIAL_WINDOW_SIZE);
+  }
+
+  private void outboundFlowControl(int windowSize) throws Exception {
+    startTransport(
+        DEFAULT_START_STREAM_ID, null, true, DEFAULT_MAX_MESSAGE_SIZE, windowSize, null);
     MockStreamListener listener = new MockStreamListener();
     OkHttpClientStream stream =
         clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
     stream.start(listener);
     // The first message should be sent out.
-    int messageLength = Utils.DEFAULT_WINDOW_SIZE / 2 + 1;
+    int messageLength = windowSize / 2 + 1;
     InputStream input = new ByteArrayInputStream(new byte[messageLength]);
     stream.writeMessage(input);
     stream.flush();
@@ -698,13 +713,13 @@ public class OkHttpClientTransportTest {
     stream.writeMessage(input);
     stream.flush();
     int partiallySentSize =
-        Utils.DEFAULT_WINDOW_SIZE - messageLength - HEADER_LENGTH;
+        windowSize - messageLength - HEADER_LENGTH;
     verify(frameWriter, timeout(TIME_OUT_MS))
         .data(eq(false), eq(3), any(Buffer.class), eq(partiallySentSize));
 
     // Get more credit, the rest data should be sent out.
-    frameHandler().windowUpdate(3, Utils.DEFAULT_WINDOW_SIZE);
+    frameHandler().windowUpdate(3, windowSize);
-    frameHandler().windowUpdate(0, Utils.DEFAULT_WINDOW_SIZE);
+    frameHandler().windowUpdate(0, windowSize);
     verify(frameWriter, timeout(TIME_OUT_MS)).data(
         eq(false), eq(3), any(Buffer.class),
         eq(messageLength + HEADER_LENGTH - partiallySentSize));
@@ -714,6 +729,16 @@ public class OkHttpClientTransportTest {
     shutdownAndVerify();
   }
 
+  @Test
+  public void outboundFlowControl_smallWindowSize() throws Exception {
+    outboundFlowControl(100);
+  }
+
+  @Test
+  public void outboundFlowControl_bigWindowSize() throws Exception {
+    outboundFlowControl(INITIAL_WINDOW_SIZE * 2);
+  }
+
   @Test
   public void outboundFlowControlWithInitialWindowSizeChange() throws Exception {
     initTransport();
@@ -1076,7 +1101,7 @@ public class OkHttpClientTransportTest {
 
     frameHandler().headers(false, false, 3, 0, grpcResponseHeaders(), HeadersMode.HTTP_20_HEADERS);
 
-    int messageLength = Utils.DEFAULT_WINDOW_SIZE + 1;
+    int messageLength = INITIAL_WINDOW_SIZE + 1;
     byte[] fakeMessage = new byte[messageLength];
     Buffer buffer = createMessageFrame(fakeMessage);
     int messageFrameLength = (int) buffer.size();
@@ -1209,13 +1234,13 @@ public class OkHttpClientTransportTest {
     stream.cancel(Status.CANCELLED);
 
     Buffer buffer = createMessageFrame(
-        new byte[Utils.DEFAULT_WINDOW_SIZE / 2 + 1]);
+        new byte[INITIAL_WINDOW_SIZE / 2 + 1]);
     frameHandler().data(false, 3, buffer, (int) buffer.size());
     // Should still update the connection window even stream 3 is gone.
     verify(frameWriter, timeout(TIME_OUT_MS)).windowUpdate(0,
-        HEADER_LENGTH + Utils.DEFAULT_WINDOW_SIZE / 2 + 1);
+        HEADER_LENGTH + INITIAL_WINDOW_SIZE / 2 + 1);
     buffer = createMessageFrame(
-        new byte[Utils.DEFAULT_WINDOW_SIZE / 2 + 1]);
+        new byte[INITIAL_WINDOW_SIZE / 2 + 1]);
 
     // This should kill the connection, since we never created stream 5.
     frameHandler().data(false, 5, buffer, (int) buffer.size());
@@ -1490,6 +1515,7 @@ public class OkHttpClientTransportTest {
         hostnameVerifier,
         ConnectionSpec.CLEARTEXT,
         DEFAULT_MAX_MESSAGE_SIZE,
+        INITIAL_WINDOW_SIZE,
         NO_PROXY,
         tooManyPingsRunnable,
         transportTracer);
@@ -1512,6 +1538,7 @@ public class OkHttpClientTransportTest {
         hostnameVerifier,
         ConnectionSpec.CLEARTEXT,
         DEFAULT_MAX_MESSAGE_SIZE,
+        INITIAL_WINDOW_SIZE,
         NO_PROXY,
         tooManyPingsRunnable,
         new TransportTracer());
@@ -1542,6 +1569,7 @@ public class OkHttpClientTransportTest {
         hostnameVerifier,
         ConnectionSpec.CLEARTEXT,
         DEFAULT_MAX_MESSAGE_SIZE,
+        INITIAL_WINDOW_SIZE,
         new ProxyParameters(
             (InetSocketAddress) serverSocket.getLocalSocketAddress(), NO_USER, NO_PW),
         tooManyPingsRunnable,
@@ -1592,6 +1620,7 @@ public class OkHttpClientTransportTest {
         hostnameVerifier,
         ConnectionSpec.CLEARTEXT,
         DEFAULT_MAX_MESSAGE_SIZE,
+        INITIAL_WINDOW_SIZE,
         new ProxyParameters(
             (InetSocketAddress) serverSocket.getLocalSocketAddress(), NO_USER, NO_PW),
         tooManyPingsRunnable,
@@ -1641,6 +1670,7 @@ public class OkHttpClientTransportTest {
         hostnameVerifier,
         ConnectionSpec.CLEARTEXT,
         DEFAULT_MAX_MESSAGE_SIZE,
+        INITIAL_WINDOW_SIZE,
         new ProxyParameters(
             (InetSocketAddress) serverSocket.getLocalSocketAddress(), NO_USER, NO_PW),
         tooManyPingsRunnable,