This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/master by this push:
new d914a6e704 Bound streaming reader timeout by readTimeout instead of a
fixed 30s (#3505)
d914a6e704 is described below
commit d914a6e704ffaacffcf299be8f8b099a9474ba44
Author: Ken Hu <[email protected]>
AuthorDate: Fri Jul 10 12:00:37 2026 -0700
Bound streaming reader timeout by readTimeout instead of a fixed 30s (#3505)
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 | 15 ++++--
.../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, 155 insertions(+), 22 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 d67f16adc0..bc54061b05 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
@@ -239,11 +239,16 @@ public final class ResultSet implements Iterable<Result> {
* @param throwable the error that occurred
*/
public void markError(final Throwable throwable) {
- error.set(throwable);
- // an error may arrive before headers were ever signaled (e.g. a write
failure or a non-streaming error
- // response); fail the headers future too so anyone blocked on it
observes the error rather than hanging.
- this.headersReceived.completeExceptionally(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)) {
+ // an error may arrive before headers were ever signaled (e.g. a
write failure or a non-streaming error
+ // response); fail the headers future too so anyone blocked on it
observes the error rather than hanging.
+ this.headersReceived.completeExceptionally(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 1beb726564..b433bf8bf8 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 BACKSTOP_GRACE_MILLIS = 5000L;
+
private final GraphBinaryReader graphBinaryReader;
private final AtomicReference<ResultSet> pendingResultSet;
private final ExecutorService readerPool;
+ private final long backstopTimeoutMillis;
// 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
#BACKSTOP_GRACE_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.backstopTimeoutMillis = readTimeoutMillis > 0 ? readTimeoutMillis
+ BACKSTOP_GRACE_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(backstopTimeoutMillis);
// Signal that the server's response headers have arrived, before
any body chunk is processed. The full
// round trip to the server has completed, so a caller blocked on
headersReceivedAsync() (e.g. a remote
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 707167a29b..006ce663f0 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
@@ -61,6 +61,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();
