This is an automated email from the ASF dual-hosted git repository.
Cole-Greer 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 1e0cb6f267 Behavioral test fixes and error-handling consistency across
GLVs (#3519)
1e0cb6f267 is described below
commit 1e0cb6f267ea85c7592ec9fe811c3600a5b980a0
Author: Cole Greer <[email protected]>
AuthorDate: Sun Jul 19 19:14:24 2026 -0700
Behavioral test fixes and error-handling consistency across GLVs (#3519)
This is a followup to the audit findings from
https://github.com/apache/tinkerpop/pull/3436.
Improves error-handling consistency for adversarial HTTP scenarios (empty
responses, malformed responses, dead connections) across the Gremlin drivers,
and reconciles a few timeout-related test gaps against master's
connection-options standardization work.
### Java (gremlin-driver)
- **Problem:** An empty HTTP response body surfaced as a bare EOFException
with no context.
- **Fix:** GraphBinaryStreamResponseReader now detects this case and raises
a clear "Server returned an empty response body" error.
### Go (gremlin-go)
- **Problem:** Same empty-body issue as Java — the driver silently returned
an empty result set instead of an error.
- **Fix:** Empty response bodies now produce a clear error.
- **Problem:** TestShouldTimeoutWhenServerNeverResponds was skipped
outright since the driver had no client-side read/request timeout at the time.
- **Fix:** Un-skipped, now uses the standardized ReadTimeout option
(confirmed it correctly bounds this case, since Go's per-read deadline is armed
before the first read).
### Python (gremlin_python)
- **Problem:** Transport-level failures (server closes connection, partial
response, empty body) leaked raw aiohttp exception types
(ClientConnectionError, ClientPayloadError, ServerDisconnectedError) with no
Gremlin-level context.
- **Fix:** Wraps these in a new GremlinConnectionError with an actionable
message.
- **Problem:** A half-closed connection wasn't evicted from the pool after
an empty response, so the client could keep reusing a dead connection.
- **Fix:** Releases the response on error so the connection is evicted and
the pool recovers.
### .NET (Gremlin.Net)
- **Problem:** A malformed response yielded a non-deterministic exception
type (IOException or KeyNotFoundException depending on where deserialization
failed), forcing the test to accept either type.
- **Fix:** Deserialization failures are now wrapped consistently. Reuses
the existing ResponseException (via a new constructor overload with a
NoStatusCode sentinel) rather than introducing a separate exception type,
keeping one exception type per driver for response-related failures.
### JavaScript (gremlin-javascript)
- **Problem:** No client-side timeout existed for a server that never
responds at all.
- **Fix:** Investigated using the newly standardized readTimeoutMillis
option, but confirmed it only maps to undici's bodyTimeout, which doesn't start
ticking until response parsing begins — so it never fires for a server that
sends nothing. The test is left skipped, but with updated context.
---
.../src/Gremlin.Net/Driver/Connection.cs | 8 ++++++
.../Driver/Exceptions/ResponseException.cs | 21 +++++++++++++-
.../Driver/ClientBehaviorIntegrationTests.cs | 16 +++++------
.../stream/GraphBinaryStreamResponseReader.java | 19 +++++++++++--
.../driver/ClientBehaviorIntegrateTest.java | 4 +--
.../GraphBinaryStreamResponseReaderTest.java | 23 ++++++++++++++++
gremlin-go/driver/client_behavior_test.go | 32 ++++++++++++++--------
gremlin-go/driver/connection.go | 6 +++-
.../gremlin_python/driver/aiohttp/transport.py | 8 ++++++
.../python/gremlin_python/driver/connection.py | 32 ++++++++++++++++++++++
.../integration/driver/test_client_behavior.py | 21 +++++++-------
11 files changed, 152 insertions(+), 38 deletions(-)
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
b/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
index 13de107654..ef9084f97b 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
@@ -30,6 +30,7 @@ using System.Net.Http.Headers;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
+using Gremlin.Net.Driver.Exceptions;
using Gremlin.Net.Driver.Messages;
using Gremlin.Net.Process;
@@ -377,6 +378,13 @@ namespace Gremlin.Net.Driver
}
channel.Writer.Complete();
}
+ catch (Exception ex) when (ex is not ResponseException
+ and not
OperationCanceledException
+ and not HttpIOException)
+ {
+ channel.Writer.Complete(
+ new ResponseException(ex));
+ }
catch (Exception ex)
{
channel.Writer.Complete(ex);
diff --git
a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ResponseException.cs
b/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ResponseException.cs
index 0dbac9c12a..b334dffdd9 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ResponseException.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ResponseException.cs
@@ -44,7 +44,26 @@ namespace Gremlin.Net.Driver.Exceptions
}
/// <summary>
- /// Gets the status code from the GraphBinary status footer.
+ /// Initializes a new instance of the <see
cref="ResponseException" /> class for a response that
+ /// could not be deserialized. There is no status code from the
server in this case since the
+ /// failure occurred locally while reading the response.
+ /// </summary>
+ /// <param name="innerException">The exception that caused the
deserialization failure.</param>
+ public ResponseException(Exception innerException)
+ : base("Failed to deserialize the response received from Gremlin
Server.", innerException)
+ {
+ StatusCode = NoStatusCode;
+ }
+
+ /// <summary>
+ /// The <see cref="StatusCode" /> value used when the exception
was not raised from a status
+ /// code reported by the server (e.g. a local deserialization
failure).
+ /// </summary>
+ public const int NoStatusCode = -1;
+
+ /// <summary>
+ /// Gets the status code from the GraphBinary status footer, or
<see cref="NoStatusCode" /> if this
+ /// exception represents a local deserialization failure rather
than a server-reported error.
/// </summary>
public int StatusCode { get; }
diff --git
a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ClientBehaviorIntegrationTests.cs
b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ClientBehaviorIntegrationTests.cs
index dca18329ad..e00f2bd48d 100644
---
a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ClientBehaviorIntegrationTests.cs
+++
b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ClientBehaviorIntegrationTests.cs
@@ -173,17 +173,13 @@ namespace Gremlin.Net.IntegrationTest.Driver
{
SkipIfServerUnavailable();
- // NOTE: the driver surfaces a low-level exception (no
Gremlin-aware wrapping,).
- // The exact type is non-deterministic for malformed bytes: either
an IOException
- // at the stream layer or a KeyNotFoundException from the
GraphBinary deserializer,
- // depending on how the chunk is read.
- var ex = await Assert.ThrowsAnyAsync<Exception>(async () =>
+ var ex = await Assert.ThrowsAsync<ResponseException>(async () =>
{
var resultSet = await
_client!.SubmitAsync<dynamic>(SocketServerConstants.GremlinMalformedResponse);
await resultSet.ToListAsync();
});
- Assert.True(ex is System.IO.IOException or KeyNotFoundException,
- $"Unexpected exception type: {ex.GetType().FullName}");
+ Assert.Equal(ResponseException.NoStatusCode, ex.StatusCode);
+ Assert.NotNull(ex.InnerException);
// Recovery
var resultSet = await
_client!.SubmitAsync<dynamic>(SocketServerConstants.GremlinSingleVertex);
@@ -196,12 +192,14 @@ namespace Gremlin.Net.IntegrationTest.Driver
{
SkipIfServerUnavailable();
- var ex = await Assert.ThrowsAsync<System.IO.IOException>(async ()
=>
+ var ex = await Assert.ThrowsAsync<ResponseException>(async () =>
{
var resultSet = await
_client!.SubmitAsync<dynamic>(SocketServerConstants.GremlinEmptyBody);
await resultSet.ToListAsync();
});
- Assert.Contains("Unexpected end of stream", ex.Message);
+ Assert.Equal(ResponseException.NoStatusCode, ex.StatusCode);
+ Assert.IsType<System.IO.IOException>(ex.InnerException);
+ Assert.Contains("Unexpected end of stream",
ex.InnerException!.Message);
// Recovery
var resultSet = await
_client!.SubmitAsync<dynamic>(SocketServerConstants.GremlinSingleVertex);
diff --git
a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/stream/GraphBinaryStreamResponseReader.java
b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/stream/GraphBinaryStreamResponseReader.java
index 8432d01562..63766145da 100644
---
a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/stream/GraphBinaryStreamResponseReader.java
+++
b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/stream/GraphBinaryStreamResponseReader.java
@@ -29,6 +29,7 @@ import
org.apache.tinkerpop.gremlin.structure.io.binary.Marker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.EOFException;
import java.util.concurrent.atomic.AtomicReference;
/**
@@ -95,11 +96,25 @@ public class GraphBinaryStreamResponseReader implements
Runnable {
resultSet.markError(ResponseException.create(HttpResponseStatus.valueOf(statusCode),
message, exception));
}
} catch (Throwable t) {
- logger.warn("Error reading streaming response", t);
- resultSet.markError(t);
+ if (buffer.readerIndex() == 0 && hasCause(t, EOFException.class)) {
+ final RuntimeException empty = new RuntimeException(
+ "Server returned an empty response body", t);
+ logger.warn("Error reading streaming response", empty);
+ resultSet.markError(empty);
+ } else {
+ logger.warn("Error reading streaming response", t);
+ resultSet.markError(t);
+ }
} finally {
pendingResultSet.compareAndSet(resultSet, null);
buffer.release();
}
}
+
+ private static boolean hasCause(final Throwable t, final Class<? extends
Throwable> type) {
+ for (Throwable cause = t; cause != null; cause = cause.getCause()) {
+ if (type.isInstance(cause)) return true;
+ }
+ return false;
+ }
}
diff --git
a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientBehaviorIntegrateTest.java
b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientBehaviorIntegrateTest.java
index 56f3da64e9..59dc53a4e4 100644
---
a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientBehaviorIntegrateTest.java
+++
b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientBehaviorIntegrateTest.java
@@ -244,8 +244,8 @@ public class ClientBehaviorIntegrateTest {
} catch (ExecutionException e) {
// Empty body causes the stream reader to hit EOF immediately
assertThat(e.getCause(), instanceOf(RuntimeException.class));
- assertThat(e.getCause().getCause(),
instanceOf(java.io.EOFException.class));
- assertTrue(e.getCause().getMessage().contains("EOFException"));
+ assertTrue(e.getCause().getMessage().contains("empty response
body"));
+ assertThat(ExceptionHelper.getRootCause(e.getCause()),
instanceOf(java.io.EOFException.class));
} catch (TimeoutException e) {
fail("Driver hung on empty response body instead of throwing an
error");
}
diff --git
a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/GraphBinaryStreamResponseReaderTest.java
b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/GraphBinaryStreamResponseReaderTest.java
index 484ec7d524..06adcbe80d 100644
---
a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/GraphBinaryStreamResponseReaderTest.java
+++
b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/GraphBinaryStreamResponseReaderTest.java
@@ -223,4 +223,27 @@ public class GraphBinaryStreamResponseReaderTest {
assertTrue(results.isEmpty());
assertNull(pending.get());
}
+
+ @Test
+ public void shouldProduceClearErrorOnEmptyBody() {
+ final ResultSet rs = new ResultSet(executor,
RequestMessage.build("g.V()").create(), null);
+ final AtomicReference<ResultSet> pending = new AtomicReference<>(rs);
+
+ // Simulate an empty HTTP response body — no bytes offered before
end-of-stream
+ final ByteBufQueueInputStream stream = new ByteBufQueueInputStream();
+ stream.signalEndOfStream();
+
+ final InputStreamBuffer buffer = new InputStreamBuffer(stream);
+ new GraphBinaryStreamResponseReader(buffer, reader, rs, pending).run();
+
+ assertTrue(rs.allItemsAvailable());
+ try {
+ rs.all().get();
+ fail("Expected exception");
+ } catch (Exception e) {
+ assertTrue(e.getCause() instanceof RuntimeException);
+ assertEquals("Server returned an empty response body",
e.getCause().getMessage());
+ }
+ assertNull(pending.get());
+ }
}
diff --git a/gremlin-go/driver/client_behavior_test.go
b/gremlin-go/driver/client_behavior_test.go
index b358b06404..5359b29781 100644
--- a/gremlin-go/driver/client_behavior_test.go
+++ b/gremlin-go/driver/client_behavior_test.go
@@ -262,14 +262,10 @@ func TestShouldHandleEmptyResponseBody(t *testing.T) {
done <- submitExpectErr(client, gremlinEmptyBody)
}()
- // The key requirement is that an empty response body does not hang.
- // NOTE: Unlike the Java/Python/JS drivers (which raise an error), the
Go
- // driver currently treats an empty body as an empty (successful) result
- // set rather than an error. This driver gap is flagged in the cross-GLV
- // error-message audit (tinkerpop-8lw.6) for further consideration.
select {
- case <-done:
- // completed without hanging - acceptable for now
+ case submitErr := <-done:
+ require.Error(t, submitErr)
+ assert.Contains(t, submitErr.Error(), "empty response body")
case <-ctx.Done():
t.Fatal("request hung on empty response body")
}
@@ -289,11 +285,23 @@ func TestShouldHandleSlowResponse(t *testing.T) {
}
func TestShouldTimeoutWhenServerNeverResponds(t *testing.T) {
- // The Go driver's ConnectionTimeout only governs connection
establishment,
- // not how long to wait for a response. With no client-side request/read
- // timeout, a server that never responds causes an indefinite hang.
Skipped
- // until the driver supports a request timeout (flagged in
tinkerpop-8lw.6).
- t.Skip("Go driver lacks a client-side request/read timeout")
+ url := socketServerURL()
+ client, err := NewClient(url, func(settings *ClientSettings) {
+ settings.ReadTimeout = 2 * time.Second
+ })
+ if err != nil {
+ t.Skip("Socket server not available")
+ }
+ defer client.Close()
+
+ // Verify connectivity before testing the no-response scenario
+ if err := submitExpectErr(client, gremlinSingleVertex); err != nil {
+ t.Skip("Socket server not available")
+ }
+
+ err = submitExpectErr(client, gremlinNoResponse)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "timeout")
}
func TestShouldHandleAsyncRequestsDuringConnectionClose(t *testing.T) {
diff --git a/gremlin-go/driver/connection.go b/gremlin-go/driver/connection.go
index b00852fb65..93502910c3 100644
--- a/gremlin-go/driver/connection.go
+++ b/gremlin-go/driver/connection.go
@@ -395,7 +395,11 @@ func (c *connection) streamToResultSet(reader io.Reader,
rs ResultSet) {
d = NewGraphBinaryDeserializer(reader)
}
if err := d.ReadHeader(); err != nil {
- if err != io.EOF {
+ if err == io.EOF {
+ emptyBodyErr := fmt.Errorf("server returned an empty
response body")
+ c.logHandler.logf(Error, failedToReceiveResponse,
emptyBodyErr.Error())
+ rs.setError(emptyBodyErr)
+ } else {
c.logHandler.logf(Error, failedToReceiveResponse,
err.Error())
rs.setError(err)
}
diff --git
a/gremlin-python/src/main/python/gremlin_python/driver/aiohttp/transport.py
b/gremlin-python/src/main/python/gremlin_python/driver/aiohttp/transport.py
index 447069ffd3..37e464856a 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/aiohttp/transport.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/aiohttp/transport.py
@@ -325,6 +325,14 @@ class AiohttpHTTPTransport:
"""Read the entire HTTP response body as bytes."""
return _run_read(self._loop, self._read_timeout,
self._http_req_resp.read())
+ def evict_response(self):
+ """Close the current HTTP response and its underlying connection,
evicting it from
+ aiohttp's connection pool. Used after a transport failure so a
dead/half-closed
+ connection is discarded rather than returned to the pool for reuse."""
+ if self._http_req_resp is not None:
+ self._http_req_resp.close()
+ self._http_req_resp = None
+
def close(self):
# Inner function to perform async close.
async def async_close():
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/connection.py
b/gremlin-python/src/main/python/gremlin_python/driver/connection.py
index 7a46a53a44..b482abc4c9 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/connection.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/connection.py
@@ -14,15 +14,28 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+import asyncio
import queue
from concurrent.futures import Future
+from aiohttp.client_exceptions import (
+ ClientOSError,
+ ClientPayloadError,
+ ServerDisconnectedError,
+)
+
from gremlin_python.driver import resultset, useragent
from gremlin_python.driver.aiohttp.transport import AiohttpHTTPTransport
from gremlin_python.driver.http_request import HttpRequest
__author__ = 'David M. Brown ([email protected])'
+_TRANSPORT_ERRORS = (ClientOSError, ClientPayloadError,
ServerDisconnectedError, asyncio.IncompleteReadError)
+_CONNECTION_ERROR_MSG = (
+ "Connection to server closed unexpectedly. "
+ "Ensure that the server is still reachable and the connection has not been
closed by the server or a network device."
+)
+
class GremlinServerError(Exception):
def __init__(self, status):
@@ -32,6 +45,11 @@ class GremlinServerError(Exception):
self.status_exception = status['exception']
+class GremlinConnectionError(Exception):
+ """Raised when a transport-level failure occurs communicating with the
server."""
+ pass
+
+
class Connection:
def __init__(self, url, traversal_source,
@@ -127,6 +145,11 @@ class Connection:
def cb(f):
try:
f.result()
+ except _TRANSPORT_ERRORS as e:
+ wrapped = GremlinConnectionError(_CONNECTION_ERROR_MSG)
+ wrapped.__cause__ = e
+ future.set_exception(wrapped)
+ self._pool.put_nowait(self)
except Exception as e:
future.set_exception(e)
self._pool.put_nowait(self)
@@ -164,6 +187,15 @@ class Connection:
stream = self._transport.get_stream()
for obj in
self._response_serializer.deserialize_response_stream(stream):
self._result_set.stream.put_nowait(obj)
+ except _TRANSPORT_ERRORS as err:
+ # Evict the dead connection from aiohttp's internal pool, ensuring
subsequent
+ # requests get a fresh connection.
+ self._transport.evict_response()
+ msg = 'Server returned an empty response body' if isinstance(err,
asyncio.IncompleteReadError) and not err.partial else _CONNECTION_ERROR_MSG
+ raise GremlinConnectionError(msg) from err
+ except Exception:
+ self._transport.evict_response()
+ raise
finally:
self._pool.put_nowait(self)
diff --git
a/gremlin-python/src/main/python/tests/integration/driver/test_client_behavior.py
b/gremlin-python/src/main/python/tests/integration/driver/test_client_behavior.py
index 18eef5edf8..491c106a97 100644
---
a/gremlin-python/src/main/python/tests/integration/driver/test_client_behavior.py
+++
b/gremlin-python/src/main/python/tests/integration/driver/test_client_behavior.py
@@ -25,10 +25,9 @@ import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
import pytest
-from aiohttp.client_exceptions import ClientPayloadError,
ServerDisconnectedError
from gremlin_python.driver.client import Client
-from gremlin_python.driver.connection import GremlinServerError
+from gremlin_python.driver.connection import GremlinConnectionError,
GremlinServerError
from gremlin_python.driver.serializer import GraphBinarySerializersV4
from .socket_server_constants import (
@@ -74,7 +73,7 @@ def test_should_receive_single_vertex(socket_server_client):
def
test_should_handle_server_closing_connection_before_response(socket_server_client):
- with pytest.raises(ServerDisconnectedError, match="Server disconnected"):
+ with pytest.raises(GremlinConnectionError, match="Connection to server
closed unexpectedly"):
socket_server_client.submit(GREMLIN_CLOSE_CONNECTION).all().result()
# Recovery
@@ -101,7 +100,7 @@ def
test_should_handle_server_error_after_delay(socket_server_client):
def test_should_handle_partial_content_close(socket_server_client):
- with pytest.raises(ClientPayloadError, match="payload is not completed"):
+ with pytest.raises(GremlinConnectionError, match="Connection to server
closed unexpectedly"):
socket_server_client.submit(GREMLIN_PARTIAL_CONTENT_CLOSE).all().result()
# Recovery
@@ -121,15 +120,15 @@ def
test_should_handle_malformed_response(socket_server_client):
def test_should_handle_empty_response_body(fresh_client):
- # An empty HTTP response body should surface as an error rather than hang.
- with pytest.raises(asyncio.IncompleteReadError):
+ # An empty HTTP response body should surface as a GremlinConnectionError
+ # wrapping the underlying IncompleteReadError.
+ with pytest.raises(GremlinConnectionError, match="Server returned an empty
response body"):
fresh_client.submit(GREMLIN_EMPTY_BODY).all().result()
- # NOTE: Unlike the Java driver, the Python (aiohttp) driver does not
recover
- # on the same client after an empty response body - the half-closed
connection
- # is not evicted from the pool and a subsequent request fails with
- # 'Cannot write to closing transport'. This driver gap is flagged in the
- # cross-GLV error-message audit (tinkerpop-8lw.6) for further
consideration.
+ # Recovery on the same client - the dead connection should be evicted from
+ # aiohttp's internal pool so subsequent requests get a fresh connection.
+ result = fresh_client.submit(GREMLIN_SINGLE_VERTEX).all().result()
+ assert len(result) == 1
def test_should_handle_slow_response(socket_server_client):
