This is an automated email from the ASF dual-hosted git repository. kenhuuu pushed a commit to branch j-flaky in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 4ea917bd9b20a770821f620d004dabf94b5ab66d Author: Ken Hu <[email protected]> AuthorDate: Wed Jul 1 15:16:54 2026 -0700 Remove redundant connection return in gremlin-driver CTR This is code leftover from the WebSocket implementation. Connections are now returned by the handler in streaming mode so a second way of returning the connection actually caused problems in the error path as the connection would be signaled as being available twice after an error. Assisted-by: Kiro:claude-opus-4-8 --- .../org/apache/tinkerpop/gremlin/driver/Connection.java | 13 ++++++++----- .../gremlin/server/GremlinServerIntegrateTest.java | 13 ++++++++++++- 2 files changed, 20 insertions(+), 6 deletions(-) diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java index 942da4d0bc..b416bd51ab 100644 --- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java +++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java @@ -289,12 +289,15 @@ final class Connection { resultSet.getReadCompleted().whenComplete((v, t) -> { if (t != null) { - // the callback for when the read failed. a failed read means the request went to the server - // and came back with a server-side error of some sort. it means the server is responsive - // so this isn't going to be like a potentially dead host situation which is handled above on a failed - // write operation. + // A failed read means the request reached the server and came back with a + // server-side error, so the server is responsive and the connection is healthy. + // The connection is returned to the pool by the wire-level completion path + // (LAST_CONTENT_READ_RESPONSE), so no connection cleanup is performed here. + // Returning it here as well was redundant and, because this callback can run after + // the connection has already been returned and re-borrowed for a later request, + // could return a connection that is still in flight for a different request. + // Dead/dropped connections are handled by the channel close/inactive paths. logger.debug("Error while processing request on the server {}.", this, t); - handleConnectionCleanupOnError(thisConnection); } // While this request was in process, close might have been signaled in closeAsync(). // However, close would be blocked until all pending requests are completed. Attempt 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 a67592c2da..07e8e4ce71 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 @@ -273,7 +273,11 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration } } - @Test + // Backstop timeout: this test floods the server to exercise rate limiting and then joins on the request + // futures. If a response were ever lost the unbounded joins would block forever and stall the entire CI job, + // so this generous cap bounds any such failure to this method rather than the whole build. With the + // connection-return fix in place it should never fire; it exists purely as a safety net. + @Test(timeout = 300000) public void shouldBlowTheWorkQueueSize() throws Exception { final Cluster cluster = TestClientFactory.open(); final Client client = cluster.connect(); @@ -287,6 +291,13 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration final AtomicBoolean errorTriggered = new AtomicBoolean(); final ResultSet r1 = client.submitAsync("Thread.sleep(1000);'test4'", groovyRequestOptions).get(); + // Give r1 a brief head start to reach the server and occupy the single gremlin worker before the + // flood begins. r1 and the flood requests travel on different pooled connections, so without this + // wait the flood could reach the server's single-slot work queue first and rate-limit r1 itself + // (the request that is meant to be holding the worker). Occupying the worker takes only milliseconds + // while r1 sleeps for 1s, so this wait still leaves ample time for the queue to fill during the flood. + Thread.sleep(500); + final List<CompletableFuture<List<Result>>> blockers = new ArrayList<>(); for (int ix = 0; ix < 512 && !errorTriggered.get(); ix++) { blockers.add(client.submit("'test'", groovyRequestOptions).all().exceptionally(t -> {
