Fixes #8405 - onAllDataRead() is called twice under h2 if the stream … (#10174)

* Fixes #8405 - onAllDataRead() is called twice under h2 if the stream times out

Per Servlet semantic, HTTP/2 stream timeout should be ignored.

The code was trying to fail the read via `_contentDemander.onTimeout()`, but
then it was still calling `onContentProducible()`, which was returning `true`
because the state of the read was IDLE (all the request content was read) and
the request was suspended.

Now the code checks if the read was really failed; if it is not, then
`onContentProducible()` is not called and so the idle timeout is ignored.

Signed-off-by: Simone Bordet <simone.bordet@gmail.com>

jetty-http2/http2-server/src/main/java/org/eclipse/jetty/http2/server/HttpChannelOverHTTP2.java

@@ -645,7 +645,7 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
     @Override
     public boolean onTimeout(Throwable failure, Consumer<Runnable> consumer)
     {
-        final boolean delayed = _delayedUntilContent;
+        boolean delayed = _delayedUntilContent;
         _delayedUntilContent = false;
 
         boolean reset = isIdle();
@@ -655,10 +655,9 @@ public class HttpChannelOverHTTP2 extends HttpChannel implements Closeable, Writ
         getHttpTransport().onStreamTimeout(failure);
 
         failure.addSuppressed(new Throwable("HttpInput idle timeout"));
-        _contentDemander.onTimeout(failure);
-        boolean needed = getRequest().getHttpInput().onContentProducible();
-
-        if (needed || delayed)
+        boolean readFailed = _contentDemander.onTimeout(failure);
+        boolean handle = readFailed && getRequest().getHttpInput().onContentProducible();
+        if (handle || delayed)
         {
             consumer.accept(this::handleWithContext);
             reset = false;

tests/test-http-client-transport/src/test/java/org/eclipse/jetty/http/client/AsyncIOServletTest.java

@@ -29,6 +29,7 @@ import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -96,7 +97,9 @@ import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -1739,6 +1742,69 @@ public class AsyncIOServletTest extends AbstractTest<AsyncIOServletTest.AsyncTra
         assertThat(failures, empty());
     }
 
+    @ParameterizedTest
+    @ArgumentsSource(TransportProvider.class)
+    public void testOnAllDataReadCalledOnceThenIdleTimeout(Transport transport) throws Exception
+    {
+        init(transport);
+        AtomicInteger allDataReadCount = new AtomicInteger();
+        AtomicReference<Throwable> errorRef = new AtomicReference<>();
+        scenario.start(new HttpServlet()
+        {
+            @Override
+            protected void service(HttpServletRequest request, HttpServletResponse resp) throws IOException
+            {
+                AsyncContext asyncContext = request.startAsync();
+                asyncContext.setTimeout(0);
+
+                ServletInputStream input = request.getInputStream();
+                input.setReadListener(new ReadListener()
+                {
+                    @Override
+                    public void onDataAvailable() throws IOException
+                    {
+                        while (input.isReady())
+                        {
+                            int read = input.read();
+                            if (read < 0)
+                                break;
+                        }
+                    }
+
+                    @Override
+                    public void onAllDataRead()
+                    {
+                        allDataReadCount.incrementAndGet();
+                    }
+
+                    @Override
+                    public void onError(Throwable x)
+                    {
+                        // There should be no errors because request body has
+                        // been successfully read and idle timeouts are ignored.
+                        errorRef.set(x);
+                    }
+                });
+
+                // Never reply to the request, let it idle timeout.
+                // The Servlet semantic is that the idle timeout will
+                // be ignored so the client will timeout the request.
+            }
+        });
+        long idleTimeout = 1000;
+        scenario.setConnectionIdleTimeout(2 * idleTimeout);
+        scenario.setRequestIdleTimeout(idleTimeout);
+
+        assertThrows(TimeoutException.class, () -> scenario.client.newRequest(scenario.newURI())
+            .path(scenario.servletPath)
+            .timeout(2 * idleTimeout, TimeUnit.MILLISECONDS)
+            .send()
+        );
+
+        assertNull(errorRef.get());
+        assertEquals(1, allDataReadCount.get());
+    }
+
     private static class Listener implements ReadListener, WriteListener
     {
         private final Executor executor = Executors.newFixedThreadPool(32);