retries:Remove early commit for transparent retries with none remaining. (#10066) (#10082)

* retries:Remove early commit for transparent retries when no retries or no hedging remain.

Fixes #10011

Author: larry-safran

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

@@ -882,29 +882,15 @@ public void run() {
           // transparent retry
           final Substream newSubstream = createSubstream(substream.previousAttemptCount, true);
           if (isHedging) {
-            boolean commit = false;
             synchronized (lock) {
               // Although this operation is not done atomically with
               // noMoreTransparentRetry.compareAndSet(false, true), it does not change the size() of
               // activeHedges, so neither does it affect the commitment decision of other threads,
               // nor do the commitment decision making threads affect itself.
               state = state.replaceActiveHedge(substream, newSubstream);
-
-              // optimization for early commit
-              if (!hasPotentialHedging(state)
-                  && state.activeHedges.size() == 1) {
-                commit = true;
-              }
-            }
-            if (commit) {
-              commitAndRun(newSubstream);
-            }
-          } else {
-            if (retryPolicy == null || retryPolicy.maxAttempts == 1) {
-              // optimization for early commit
-              commitAndRun(newSubstream);
             }
           }
+
           callExecutor.execute(new Runnable() {
             @Override
             public void run() {

core/src/test/java/io/grpc/internal/RetriableStreamTest.java

@@ -1870,7 +1870,7 @@ public void normalRetry_thenNoTransparentRetry_andNoMoreRetry() {
   }
 
   @Test
-  public void noRetry_transparentRetry_earlyCommit() {
+  public void noRetry_transparentRetry_noEarlyCommit() {
     ClientStream mockStream1 = mock(ClientStream.class);
     ClientStream mockStream2 = mock(ClientStream.class);
     InOrder inOrder = inOrder(retriableStreamRecorder, mockStream1, mockStream2);
@@ -1898,7 +1898,7 @@ method, new Metadata(), channelBufferUsed, PER_RPC_BUFFER_LIMIT, CHANNEL_BUFFER_
     inOrder.verify(retriableStreamRecorder).newSubstream(0);
     ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
         ArgumentCaptor.forClass(ClientStreamListener.class);
-    inOrder.verify(retriableStreamRecorder).postCommit();
+    verify(retriableStreamRecorder, never()).postCommit();
     inOrder.verify(mockStream2).start(sublistenerCaptor2.capture());
     inOrder.verify(mockStream2).isReady();
     inOrder.verifyNoMoreInteractions();