This is an automated email from the ASF dual-hosted git repository. kenhuuu pushed a commit to branch java-timeouts in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit bf9c3d8fb5b6e77c032c92c083bb273327c6284e Author: Ken Hu <[email protected]> AuthorDate: Thu Jul 2 20:22:04 2026 -0700 Bound streaming reader timeout by readTimeout instead of a fixed 30s The gremlin-driver streaming reader thread used a hardcoded 30s queue poll that ignored the configured readTimeout — so a client with readTimeout=0 ("no timeout", the default) was still cut off at 30s, and the failure surfaced as a raw IOException rather than a typed timeout. The reader's wait now derives from readTimeout: 0 blocks indefinitely (matching the aggregated path), and a positive value arms a backstop set longer than readTimeout so the pipeline ReadTimeoutHandler fires first and terminates the response with a proper exception. ResultSet.markError is now first-writer-wins so a read-timeout and end-of-stream racing on the same request can't leave the recorded cause and the completed future disagreeing. This surfaced as flakiness in shouldProduceProperExceptionOnTimeout on slow CI runners, where the reader repeatedly stalled on the 30s poll and pushed the test past the build limit. A 5-minute per-test timeout is added so a reader stall fails the test cleanly instead of hanging the build. Assisted-by: Claude Code:claude-opus-4-8 --- .../tinkerpop/gremlin/driver/Channelizer.java | 4 +- .../apache/tinkerpop/gremlin/driver/ResultSet.java | 9 +++- .../handler/HttpStreamingResponseHandler.java | 28 ++++++++++- .../driver/stream/ByteBufQueueInputStream.java | 54 ++++++++++++++++------ .../tinkerpop/gremlin/driver/ResultSetTest.java | 23 +++++++++ .../handler/ByteBufQueueInputStreamTest.java | 51 ++++++++++++++++++++ .../gremlin/server/GremlinServerIntegrateTest.java | 2 +- 7 files changed, 152 insertions(+), 19 deletions(-) diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java index bab50ede53..b0bd88811b 100644 --- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java +++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java @@ -226,8 +226,10 @@ public interface Channelizer extends ChannelHandler { useStreaming = true; final GraphBinaryReader graphBinaryReader = ((GraphBinaryMessageSerializerV4) serializer).getMapper().getReader(); + // Pass the connection's readTimeout so the reader thread can arm a backstop timeout longer than it, + // letting the pipeline ReadTimeoutHandler fire first on a stalled connection (see the handler). streamingResponseHandler = new HttpStreamingResponseHandler( - graphBinaryReader, pending, cluster.streamingReaderPool()); + graphBinaryReader, pending, cluster.streamingReaderPool(), cluster.getReadTimeout()); } else { useStreaming = false; gremlinResponseDecoder = new HttpGremlinResponseDecoder(serializer); diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultSet.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultSet.java index b5d506bc1f..c610ea5395 100644 --- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultSet.java +++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultSet.java @@ -210,8 +210,13 @@ public final class ResultSet implements Iterable<Result> { * @param throwable the error that occurred */ public void markError(final Throwable throwable) { - error.set(throwable); - this.readCompleted.completeExceptionally(throwable); + // First writer wins. Two threads can race to mark the same stream as failed - e.g. the Netty event loop + // firing a read-timeout while the streaming reader thread hits end-of-stream. Gate on the error field with a + // CAS so the recorded cause and the future's completion always agree. The error must be set before the future + // completes because waiting-future draining reads it after observing readCompleted.isDone(). + if (error.compareAndSet(null, throwable)) { + this.readCompleted.completeExceptionally(throwable); + } this.drainAllWaiting(); } diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpStreamingResponseHandler.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpStreamingResponseHandler.java index 7aca5a2e2c..d36880466d 100644 --- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpStreamingResponseHandler.java +++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpStreamingResponseHandler.java @@ -62,9 +62,16 @@ public class HttpStreamingResponseHandler extends MessageToMessageDecoder<HttpOb private static final Logger logger = LoggerFactory.getLogger(HttpStreamingResponseHandler.class); private static final ObjectMapper mapper = new ObjectMapper(); + /** + * Grace period added to {@code readTimeout} when arming the reader thread's backstop timeout, so the pipeline + * {@link ReadTimeoutHandler} always fires first on a stalled connection. + */ + private static final long READER_TIMEOUT_BACKSTOP_MILLIS = 5000L; + private final GraphBinaryReader graphBinaryReader; private final AtomicReference<ResultSet> pendingResultSet; private final ExecutorService readerPool; + private final long readerTimeoutMillis; // Mutable state below is accessed exclusively from the channel's event loop thread. private HttpResponseStatus responseStatus; @@ -76,9 +83,28 @@ public class HttpStreamingResponseHandler extends MessageToMessageDecoder<HttpOb public HttpStreamingResponseHandler(final GraphBinaryReader graphBinaryReader, final AtomicReference<ResultSet> pendingResultSet, final ExecutorService readerPool) { + this(graphBinaryReader, pendingResultSet, readerPool, 0L); + } + + /** + * @param readTimeoutMillis the connection's {@code readTimeout}. When positive, the reader thread's own timeout is + * armed as a backstop {@value #READER_TIMEOUT_BACKSTOP_MILLIS}ms longer, so the pipeline + * {@link ReadTimeoutHandler} fires first on a stalled connection and terminates the + * response with a properly typed exception. A value {@code <= 0} (the default, meaning + * "no timeout") leaves the reader blocking indefinitely. + * <p> + * The translation to the backstop bound lives here rather than in + * {@link ByteBufQueueInputStream}'s constructor so that the stream can be constructed with + * the effective bound directly (small values in tests, without paying the full grace). + */ + public HttpStreamingResponseHandler(final GraphBinaryReader graphBinaryReader, + final AtomicReference<ResultSet> pendingResultSet, + final ExecutorService readerPool, + final long readTimeoutMillis) { this.graphBinaryReader = graphBinaryReader; this.pendingResultSet = pendingResultSet; this.readerPool = readerPool; + this.readerTimeoutMillis = readTimeoutMillis > 0 ? readTimeoutMillis + READER_TIMEOUT_BACKSTOP_MILLIS : 0L; } @Override @@ -93,7 +119,7 @@ public class HttpStreamingResponseHandler extends MessageToMessageDecoder<HttpOb responseStatus = resp.status(); contentType = resp.headers().get(HttpHeaderNames.CONTENT_TYPE); - queueInputStream = new ByteBufQueueInputStream(); + queueInputStream = new ByteBufQueueInputStream(readerTimeoutMillis); // Spawn reader thread for GraphBinary responses if (isGraphBinaryResponse()) { diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/stream/ByteBufQueueInputStream.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/stream/ByteBufQueueInputStream.java index 757b2821cb..c44fe2809a 100644 --- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/stream/ByteBufQueueInputStream.java +++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/stream/ByteBufQueueInputStream.java @@ -37,11 +37,30 @@ public class ByteBufQueueInputStream extends InputStream { private static final ByteBuf END_OF_STREAM = Unpooled.buffer(0); private final BlockingQueue<ByteBuf> queue; + private final long timeoutMillis; private ByteBuf current; private volatile boolean eof; public ByteBufQueueInputStream() { + this(0L); + } + + /** + * @param timeoutMillis the effective maximum time a read blocks waiting for the next chunk. A value {@code <= 0} + * makes the read block indefinitely, deferring the liveness bound to the connection's + * {@code readTimeout} (see {@link org.apache.tinkerpop.gremlin.driver.handler.ReadTimeoutHandler}). + * A positive value acts only as a backstop and is expected to be set longer than + * {@code readTimeout} so the pipeline read-timeout fires first on a stalled connection. + * <p> + * This takes the already-resolved bound rather than a {@code readTimeout} to translate: the + * translation (adding the backstop grace) lives with its caller in + * {@link org.apache.tinkerpop.gremlin.driver.handler.HttpStreamingResponseHandler}. Keeping + * the constructor parameter as the effective bound also lets tests exercise the timeout with + * small values instead of paying the full backstop grace. + */ + public ByteBufQueueInputStream(final long timeoutMillis) { this.queue = new LinkedBlockingQueue<>(); + this.timeoutMillis = timeoutMillis; } /** @@ -72,13 +91,7 @@ public class ByteBufQueueInputStream extends InputStream { while (current == null || !current.isReadable()) { releaseCurrent(); - try { - current = queue.poll(30, TimeUnit.SECONDS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new IOException("Interrupted while waiting for data", e); - } - if (current == null) throw new IOException("Timed out waiting for streaming response data"); + current = awaitNext(); if (current == END_OF_STREAM) { eof = true; current = null; @@ -96,13 +109,7 @@ public class ByteBufQueueInputStream extends InputStream { // Block until at least one byte is available, then return what we have (short read). while (current == null || !current.isReadable()) { releaseCurrent(); - try { - current = queue.poll(30, TimeUnit.SECONDS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new IOException("Interrupted while waiting for data", e); - } - if (current == null) throw new IOException("Timed out waiting for streaming response data"); + current = awaitNext(); if (current == END_OF_STREAM) { eof = true; current = null; @@ -114,6 +121,25 @@ public class ByteBufQueueInputStream extends InputStream { return readable; } + /** + * Waits for the next buffer. When {@code timeoutMillis <= 0} this blocks indefinitely so the wait is bounded only + * by the connection's read-timeout (which, when it fires, closes the channel and delivers {@code END_OF_STREAM} + * here). A positive {@code timeoutMillis} is a backstop: if it elapses without a buffer, the connection's + * read-timeout should already have fired, so reaching this point signals that the normal termination path failed. + */ + private ByteBuf awaitNext() throws IOException { + try { + final ByteBuf next = timeoutMillis <= 0 + ? queue.take() + : queue.poll(timeoutMillis, TimeUnit.MILLISECONDS); + if (next == null) throw new IOException("Timed out waiting for streaming response data"); + return next; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while waiting for data", e); + } + } + @Override public void close() throws IOException { eof = true; diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetTest.java index 4bf3b8c6b7..11ff3fda0f 100644 --- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetTest.java +++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetTest.java @@ -60,6 +60,29 @@ public class ResultSetTest extends AbstractResultSetTest { assertThat(all.isCompletedExceptionally(), is(true)); } + @Test + public void shouldKeepFirstErrorWhenMarkedFailedTwice() throws InterruptedException { + // Two threads can race to fail the same stream (e.g. a read-timeout on the event loop and end-of-stream on + // the reader thread). The first error must win consistently across both the completed future and the error + // observed via getAvailableItemCount(), so the recorded cause and the future's completion never disagree. + final CompletableFuture<Void> all = resultSet.allItemsAvailableAsync(); + final RuntimeException first = new RuntimeException("first"); + final RuntimeException second = new RuntimeException("second"); + + resultSet.markError(first); + resultSet.markError(second); + + pool.awaitTermination(2, TimeUnit.SECONDS); + assertThat(all.isCompletedExceptionally(), is(true)); + + try { + resultSet.getAvailableItemCount(); + fail("Expected the recorded error to be thrown"); + } catch (RuntimeException ex) { + assertEquals(first, ex.getCause()); + } + } + @Test public void shouldHaveAllItemsAvailableOnReadComplete() throws InterruptedException { assertThat(resultSet.allItemsAvailable(), is(false)); diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/ByteBufQueueInputStreamTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/ByteBufQueueInputStreamTest.java index 91ba9cf28a..1650edd5f3 100644 --- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/ByteBufQueueInputStreamTest.java +++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/ByteBufQueueInputStreamTest.java @@ -24,6 +24,12 @@ import io.netty.buffer.Unpooled; import org.apache.tinkerpop.gremlin.driver.stream.ByteBufQueueInputStream; import org.junit.Test; +import java.io.IOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + import static org.junit.Assert.*; public class ByteBufQueueInputStreamTest { @@ -77,6 +83,51 @@ public class ByteBufQueueInputStreamTest { assertEquals(0, buf.refCnt()); } + @Test + public void shouldThrowWhenBoundedReadTimesOut() throws Exception { + // A positive timeout is a backstop - when it elapses with no buffer offered, the read fails rather than + // blocking forever. + final ByteBufQueueInputStream stream = new ByteBufQueueInputStream(50L); + try { + stream.read(); + fail("Expected a timeout since no buffer was ever offered"); + } catch (IOException ex) { + assertEquals("Timed out waiting for streaming response data", ex.getMessage()); + } + } + + @Test(timeout = 10000) + public void shouldBlockIndefinitelyWhenUnboundedUntilBufferArrives() throws Exception { + // A timeout <= 0 means "no timeout" - the read blocks until a buffer is offered rather than giving up. + final ByteBufQueueInputStream stream = new ByteBufQueueInputStream(0L); + final AtomicInteger readValue = new AtomicInteger(-2); + final AtomicReference<Throwable> failure = new AtomicReference<>(); + final CountDownLatch started = new CountDownLatch(1); + + final Thread reader = new Thread(() -> { + started.countDown(); + try { + readValue.set(stream.read()); + } catch (Throwable t) { + failure.set(t); + } + }); + reader.start(); + + // Let the reader block, then confirm it is still waiting well past what any old hardcoded bound would allow + // to elapse in this test, and only unblocks once a buffer is actually offered. + assertTrue(started.await(1, TimeUnit.SECONDS)); + Thread.sleep(200); + assertTrue("reader should still be blocked waiting for data", reader.isAlive()); + + stream.offer(Unpooled.wrappedBuffer(new byte[]{7})); + reader.join(5000); + + assertFalse("reader should have unblocked once data arrived", reader.isAlive()); + assertNull(failure.get()); + assertEquals(7, readValue.get()); + } + @Test public void shouldCleanUpOnClose() throws Exception { final ByteBufQueueInputStream stream = new ByteBufQueueInputStream(); diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java index 07e8e4ce71..427c0b8930 100644 --- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java +++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java @@ -412,7 +412,7 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration g.close(); } - @Test + @Test(timeout = 300000) public void shouldProduceProperExceptionOnTimeout() throws Exception { final Cluster cluster = TestClientFactory.open(); final Client client = cluster.connect();
