You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ke...@apache.org on 2026/03/20 23:15:40 UTC
(tinkerpop) branch master-tx-client updated (f4d9d325aa -> 77732f4643)
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a change to branch master-tx-client
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
omit f4d9d325aa Add HTTP transaction support to gremlin-driver
omit 0c049375e8 Merge ResultQueue into ResultSet to simplify result handling
new 584495c713 Merge ResultQueue into ResultSet to simplify result handling
new 77732f4643 Add HTTP transaction support to gremlin-driver
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (f4d9d325aa)
\
N -- N -- N refs/heads/master-tx-client (77732f4643)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGELOG.asciidoc | 1 -
.../apache/tinkerpop/gremlin/driver/Client.java | 177 ++++++++++++++++++---
.../apache/tinkerpop/gremlin/driver/Cluster.java | 129 ++++-----------
.../tinkerpop/gremlin/driver/Connection.java | 7 +
.../tinkerpop/gremlin/driver/ConnectionPool.java | 23 +--
.../driver/handler/HttpGremlinRequestEncoder.java | 2 +-
.../driver/remote/HttpRemoteTransaction.java | 15 +-
.../driver/remote/TransactionRemoteConnection.java | 7 -
.../tinkerpop/gremlin/driver/ClientTest.java | 159 ------------------
.../tinkerpop/gremlin/driver/ClusterTest.java | 117 --------------
.../gremlin/driver/ConnectionPoolTest.java | 4 +-
.../driver/ClientConnectionIntegrateTest.java | 67 ++++++--
12 files changed, 273 insertions(+), 435 deletions(-)
delete mode 100644 gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientTest.java
(tinkerpop) 01/02: Merge ResultQueue into ResultSet to simplify result handling
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch master-tx-client
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 584495c71315efb16cccce62c0052d793ab6fe6d
Author: Ken Hu <10...@users.noreply.github.com>
AuthorDate: Fri Mar 20 09:42:37 2026 -0700
Merge ResultQueue into ResultSet to simplify result handling
The ResultQueue class was an unnecessary indirection layer between Netty
and the ResultSet. This was probably more useful when the protocol was
WebSocket-based and there were multiple requests on the same connection
but it doesn't seem to have much of a purpose anymore.
This new setup also potentially works better for HTTP/2 if the high
level, stream-based API is used. If the frame-based API is used then
changes might be required in the future.
---
CHANGELOG.asciidoc | 1 +
.../tinkerpop/gremlin/driver/Channelizer.java | 2 +-
.../tinkerpop/gremlin/driver/Connection.java | 29 ++---
.../tinkerpop/gremlin/driver/ResultQueue.java | 139 ---------------------
.../apache/tinkerpop/gremlin/driver/ResultSet.java | 120 ++++++++++++++++--
.../driver/handler/GremlinResponseHandler.java | 35 +++---
...ltQueueTest.java => AbstractResultSetTest.java} | 33 +++--
...ultQueueTest.java => ResultSetQueuingTest.java} | 126 +++++++++----------
.../tinkerpop/gremlin/driver/ResultSetTest.java | 37 +++---
9 files changed, 232 insertions(+), 290 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 0fe49eda9c..fe123a3fae 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -37,6 +37,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Added `RequestInterceptor` to `gremlin-go` with `auth` reference implementations to replace `authInfo`.
* Refactored GraphBinary serializers to use `io.Writer` and `io.Reader` instead of `*bytes.Buffer` for streaming capacities.
* Refactored `httpProtocol` and `httpTransport` in `gremlin-go` into single `connection.go` that handles HTTP request and response.
+* Refactored result handling in `gremlin-driver` by merging `ResultQueue` into `ResultSet`.
* Bumped ANTLR to 4.13.2
* Bumped SLF4j to 2.0.16.
* Bumped GMavenPlus to 4.1.1.
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 fd7d585baf..427f76c056 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
@@ -92,7 +92,7 @@ public interface Channelizer extends ChannelHandler {
protected Connection connection;
protected Cluster cluster;
protected SslHandler sslHandler;
- private AtomicReference<ResultQueue> pending;
+ private AtomicReference<ResultSet> pending;
protected static final String PIPELINE_GREMLIN_HANDLER = "gremlin-handler";
protected static final String PIPELINE_SSL_HANDLER = "gremlin-ssl-handler";
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 5f9c1325d3..cb94b39b87 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
@@ -32,7 +32,6 @@ import io.netty.channel.socket.nio.NioSocketChannel;
import java.net.URI;
import java.time.Instant;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -56,7 +55,7 @@ final class Connection {
private final Channel channel;
private final URI uri;
- private final AtomicReference<ResultQueue> pending = new AtomicReference<>();
+ private final AtomicReference<ResultSet> pending = new AtomicReference<>();
private final Cluster cluster;
private final Client client;
private final ConnectionPool pool;
@@ -155,7 +154,7 @@ final class Connection {
return cluster;
}
- AtomicReference<ResultQueue> getPending() {
+ AtomicReference<ResultSet> getPending() {
return pending;
}
@@ -184,9 +183,9 @@ final class Connection {
return future;
}
- public ChannelPromise write(final RequestMessage requestMessage, final CompletableFuture<ResultSet> resultQueueSetup) {
- // once there is a completed write, then create a traverser for the result set and complete
- // the promise so that the client knows that that it can start checking for results.
+ public ChannelPromise write(final RequestMessage requestMessage, final CompletableFuture<ResultSet> resultSetFuture) {
+ // once there is a completed write, then create a ResultSet and complete
+ // the promise so that the client knows that it can start checking for results.
final Connection thisConnection = this;
final ChannelPromise requestPromise = channel.newPromise()
@@ -198,12 +197,11 @@ final class Connection {
handleConnectionCleanupOnError(thisConnection);
- cluster.executor().submit(() -> resultQueueSetup.completeExceptionally(f.cause()));
+ cluster.executor().submit(() -> resultSetFuture.completeExceptionally(f.cause()));
} else {
- final LinkedBlockingQueue<Result> resultLinkedBlockingQueue = new LinkedBlockingQueue<>();
- final CompletableFuture<Void> readCompleted = new CompletableFuture<>();
+ final ResultSet resultSet = new ResultSet(cluster.executor(), requestMessage, pool.host);
- readCompleted.whenCompleteAsync((v, t) -> {
+ resultSet.getReadCompleted().whenCompleteAsync((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
@@ -212,7 +210,7 @@ final class Connection {
logger.debug("Error while processing request on the server {}.", this, t);
handleConnectionCleanupOnError(thisConnection);
} else {
- // the callback for when the read was successful, meaning that ResultQueue.markComplete()
+ // the callback for when the read was successful, meaning that ResultSet.markComplete()
// was called
thisConnection.returnToPool();
}
@@ -223,15 +221,12 @@ final class Connection {
tryShutdown();
}, cluster.executor());
- final ResultQueue handler = new ResultQueue(resultLinkedBlockingQueue, readCompleted);
- // pending.put(requestMessage.getRequestId(), handler);
- pending.set(handler);
+ pending.set(resultSet);
- // resultQueueSetup should only be completed by a worker since the application code might have sync
+ // resultSetFuture should only be completed by a worker since the application code might have sync
// completion stages attached to it which and we do not want the event loop threads to process those
// stages.
- cluster.executor().submit(() -> resultQueueSetup.complete(
- new ResultSet(handler, cluster.executor(), readCompleted, requestMessage, pool.host)));
+ cluster.executor().submit(() -> resultSetFuture.complete(resultSet));
}
});
channel.writeAndFlush(requestMessage, requestPromise);
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultQueue.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultQueue.java
deleted file mode 100644
index e91e833b94..0000000000
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultQueue.java
+++ /dev/null
@@ -1,139 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.tinkerpop.gremlin.driver;
-
-import org.apache.tinkerpop.gremlin.driver.handler.GremlinResponseHandler;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.javatuples.Pair;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Queue;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * A queue of incoming {@link Result} objects. The queue is updated by the {@link GremlinResponseHandler}
- * until a response terminator is identified.
- *
- * @author Stephen Mallette (http://stephen.genoprime.com)
- */
-@SuppressWarnings("ThrowableResultOfMethodCallIgnored")
-public final class ResultQueue {
-
- private final LinkedBlockingQueue<Result> resultLinkedBlockingQueue;
-
- private final AtomicReference<Throwable> error = new AtomicReference<>();
-
- private final CompletableFuture<Void> readComplete;
-
- private final Queue<Pair<CompletableFuture<List<Result>>,Integer>> waiting = new ConcurrentLinkedQueue<>();
-
- public ResultQueue(final LinkedBlockingQueue<Result> resultLinkedBlockingQueue, final CompletableFuture<Void> readComplete) {
- this.resultLinkedBlockingQueue = resultLinkedBlockingQueue;
- this.readComplete = readComplete;
- }
-
- /**
- * Adds a {@link Result} to the queue which will be later read by the {@link ResultSet}.
- *
- * @param result a return value from the {@link Traversal} or script submitted for execution
- */
- public void add(final Result result) {
- this.resultLinkedBlockingQueue.offer(result);
- tryDrainNextWaiting(false);
- }
-
- public CompletableFuture<List<Result>> await(final int items) {
- final CompletableFuture<List<Result>> result = new CompletableFuture<>();
- waiting.add(Pair.with(result, items));
-
- tryDrainNextWaiting(false);
-
- return result;
- }
-
- public int size() {
- if (error.get() != null) throw new RuntimeException(error.get());
- return this.resultLinkedBlockingQueue.size();
- }
-
- public boolean isEmpty() {
- if (error.get() != null) throw new RuntimeException(error.get());
- return this.size() == 0;
- }
-
- public boolean isComplete() {
- return readComplete.isDone();
- }
-
- void drainTo(final Collection<Result> collection) {
- if (error.get() != null) throw new RuntimeException(error.get());
- resultLinkedBlockingQueue.drainTo(collection);
- }
-
- public void markComplete() {
- this.readComplete.complete(null);
-
- this.drainAllWaiting();
- }
-
- public void markError(final Throwable throwable) {
- error.set(throwable);
- this.readComplete.completeExceptionally(throwable);
- this.drainAllWaiting();
- }
-
- /**
- * Completes the next waiting future if there is one.
- */
- private synchronized void tryDrainNextWaiting(final boolean force) {
- // need to peek because the number of available items needs to be >= the expected size for that future. if not
- // it needs to keep waiting
- final Pair<CompletableFuture<List<Result>>, Integer> nextWaiting = waiting.peek();
- if (nextWaiting != null && (force || (resultLinkedBlockingQueue.size() >= nextWaiting.getValue1() || readComplete.isDone()))) {
- final int items = nextWaiting.getValue1();
- final CompletableFuture<List<Result>> future = nextWaiting.getValue0();
- final List<Result> results = new ArrayList<>(items);
- resultLinkedBlockingQueue.drainTo(results, items);
-
- // it's important to check for error here because a future may have already been queued in "waiting" prior
- // to the first response back from the server. if that happens, any "waiting" futures should be completed
- // exceptionally otherwise it will look like success.
- if (null == error.get())
- future.complete(results);
- else
- future.completeExceptionally(error.get());
-
- waiting.remove(nextWaiting);
- }
- }
-
- /**
- * Completes all remaining futures.
- */
- private void drainAllWaiting() {
- while (!waiting.isEmpty()) {
- tryDrainNextWaiting(true);
- }
- }
-}
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 0656586706..b5d506bc1f 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
@@ -19,22 +19,28 @@
package org.apache.tinkerpop.gremlin.driver;
import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
+import org.javatuples.Pair;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
+import java.util.Queue;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
/**
* A {@code ResultSet} is returned from the submission of a Gremlin script to the server and represents the
* results provided by the server. The results from the server are streamed into the {@code ResultSet} and
- * therefore may not be available immediately. As such, {@code ResultSet} provides access to a a number
+ * therefore may not be available immediately. As such, {@code ResultSet} provides access to a number
* of functions that help to work with the asynchronous nature of the data streaming back. Data from results
* is stored in an {@link Result} which can be used to retrieve the item once it is on the client side.
* <p/>
@@ -48,20 +54,18 @@ import java.util.stream.StreamSupport;
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
public final class ResultSet implements Iterable<Result> {
- private final ResultQueue resultQueue;
+ private final LinkedBlockingQueue<Result> resultQueue = new LinkedBlockingQueue<>();
+ private final Queue<Pair<CompletableFuture<List<Result>>, Integer>> waiting = new ConcurrentLinkedQueue<>();
+ private final AtomicReference<Throwable> error = new AtomicReference<>();
+ private final CompletableFuture<Void> readCompleted = new CompletableFuture<>();
+
private final ExecutorService executor;
private final RequestMessage originalRequestMessage;
private final Host host;
- private final CompletableFuture<Void> readCompleted;
-
- public ResultSet(final ResultQueue resultQueue, final ExecutorService executor,
- final CompletableFuture<Void> readCompleted, final RequestMessage originalRequestMessage,
- final Host host) {
+ public ResultSet(final ExecutorService executor, final RequestMessage originalRequestMessage, final Host host) {
this.executor = executor;
this.host = host;
- this.resultQueue = resultQueue;
- this.readCompleted = readCompleted;
this.originalRequestMessage = originalRequestMessage;
}
@@ -94,6 +98,7 @@ public final class ResultSet implements Iterable<Result> {
* Gets the number of items available on the client.
*/
public int getAvailableItemCount() {
+ if (error.get() != null) throw new RuntimeException(error.get());
return resultQueue.size();
}
@@ -114,7 +119,10 @@ public final class ResultSet implements Iterable<Result> {
* completed and there are less than that number specified available.
*/
public CompletableFuture<List<Result>> some(final int items) {
- return resultQueue.await(items);
+ final CompletableFuture<List<Result>> result = new CompletableFuture<>();
+ waiting.add(Pair.with(result, items));
+ tryDrainNextWaiting(false);
+ return result;
}
/**
@@ -125,6 +133,7 @@ public final class ResultSet implements Iterable<Result> {
*/
public CompletableFuture<List<Result>> all() {
return readCompleted.thenApplyAsync(unusedInput -> {
+ if (error.get() != null) throw new RuntimeException(error.get());
final List<Result> list = new ArrayList<>();
resultQueue.drainTo(list);
return list;
@@ -174,4 +183,95 @@ public final class ResultSet implements Iterable<Result> {
}
};
}
+
+ // ==================== Methods called by Handlers ====================
+
+ /**
+ * Adds a {@link Result} to the queue which will be later read by consumers.
+ *
+ * @param result a return value from the traversal or script submitted for execution
+ */
+ public void add(final Result result) {
+ this.resultQueue.offer(result);
+ tryDrainNextWaiting(false);
+ }
+
+ /**
+ * Marks the result stream as complete.
+ */
+ public void markComplete() {
+ this.readCompleted.complete(null);
+ this.drainAllWaiting();
+ }
+
+ /**
+ * Marks the result stream as failed with an error.
+ *
+ * @param throwable the error that occurred
+ */
+ public void markError(final Throwable throwable) {
+ error.set(throwable);
+ this.readCompleted.completeExceptionally(throwable);
+ this.drainAllWaiting();
+ }
+
+ /**
+ * Returns the future that completes when reading is done. Used internally
+ * for connection lifecycle management.
+ */
+ CompletableFuture<Void> getReadCompleted() {
+ return readCompleted;
+ }
+
+ // ==================== Internal queue management ====================
+
+ /**
+ * Checks if the queue is empty.
+ */
+ boolean isEmpty() {
+ if (error.get() != null) throw new RuntimeException(error.get());
+ return resultQueue.isEmpty();
+ }
+
+ /**
+ * Drains results to the provided collection.
+ */
+ void drainTo(final Collection<Result> collection) {
+ if (error.get() != null) throw new RuntimeException(error.get());
+ resultQueue.drainTo(collection);
+ }
+
+ /**
+ * Completes the next waiting future if there is one and enough results are available.
+ */
+ private synchronized void tryDrainNextWaiting(final boolean force) {
+ // need to peek because the number of available items needs to be >= the expected size for that future. if not
+ // it needs to keep waiting
+ final Pair<CompletableFuture<List<Result>>, Integer> nextWaiting = waiting.peek();
+ if (nextWaiting != null && (force || (resultQueue.size() >= nextWaiting.getValue1() || readCompleted.isDone()))) {
+ final int items = nextWaiting.getValue1();
+ final CompletableFuture<List<Result>> future = nextWaiting.getValue0();
+ final List<Result> results = new ArrayList<>(items);
+ resultQueue.drainTo(results, items);
+
+ // it's important to check for error here because a future may have already been queued in "waiting" prior
+ // to the first response back from the server. if that happens, any "waiting" futures should be completed
+ // exceptionally otherwise it will look like success.
+ if (null == error.get())
+ future.complete(results);
+ else
+ future.completeExceptionally(error.get());
+
+ waiting.remove(nextWaiting);
+ }
+ }
+
+ /**
+ * Completes all remaining futures.
+ */
+ private void drainAllWaiting() {
+ while (!waiting.isEmpty()) {
+ tryDrainNextWaiting(true);
+ }
+ }
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
index a94f17bdb2..d4cfb167d2 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
@@ -25,7 +25,7 @@ import io.netty.util.AttributeKey;
import javax.net.ssl.SSLException;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.tinkerpop.gremlin.driver.Result;
-import org.apache.tinkerpop.gremlin.driver.ResultQueue;
+import org.apache.tinkerpop.gremlin.driver.ResultSet;
import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
import org.apache.tinkerpop.gremlin.process.remote.traversal.DefaultRemoteTraverser;
import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
@@ -42,17 +42,16 @@ import java.util.concurrent.atomic.AtomicReference;
import static org.apache.tinkerpop.gremlin.driver.Channelizer.HttpChannelizer.LAST_CONTENT_READ_RESPONSE;
/**
- * Takes a map of requests pending responses and writes responses to the {@link ResultQueue} of a request
- * as the {@link ResponseMessage} objects are deserialized.
+ * Writes responses to the {@link ResultSet} of a request as the {@link ResponseMessage} objects are deserialized.
*/
public class GremlinResponseHandler extends SimpleChannelInboundHandler<ResponseMessage> {
public static final AttributeKey<Throwable> INBOUND_SSL_EXCEPTION = AttributeKey.valueOf("inboundSslException");
private static final Logger logger = LoggerFactory.getLogger(GremlinResponseHandler.class);
private static final AttributeKey<ResponseException> CAUGHT_EXCEPTION = AttributeKey.valueOf("caughtException");
- private final AtomicReference<ResultQueue> pending;
+ private final AtomicReference<ResultSet> pendingResultSet;
- public GremlinResponseHandler(final AtomicReference<ResultQueue> pending) {
- this.pending = pending;
+ public GremlinResponseHandler(final AtomicReference<ResultSet> pending) {
+ this.pendingResultSet = pending;
}
@Override
@@ -61,7 +60,7 @@ public class GremlinResponseHandler extends SimpleChannelInboundHandler<Response
// should fire off a close message which will properly release the driver.
super.channelInactive(ctx);
- final ResultQueue current = pending.getAndSet(null);
+ final ResultSet current = pendingResultSet.getAndSet(null);
if (current != null) {
current.markError(new IllegalStateException("Connection to server is no longer active"));
}
@@ -70,28 +69,28 @@ public class GremlinResponseHandler extends SimpleChannelInboundHandler<Response
@Override
protected void channelRead0(final ChannelHandlerContext channelHandlerContext, final ResponseMessage response) {
final HttpResponseStatus statusCode = response.getStatus() == null ? null : response.getStatus().getCode();
- final ResultQueue queue = pending.get();
+ final ResultSet resultSet = pendingResultSet.get();
if ((null == statusCode) || (statusCode == HttpResponseStatus.OK)) {
final List<Object> data = response.getResult().getData();
final boolean bulked = channelHandlerContext.channel().attr(HttpGremlinResponseStreamDecoder.IS_BULKED).get();
- // unrolls the collection into individual results to be handled by the queue.
+ // unrolls the collection into individual results to be handled by the ResultSet.
if (bulked) {
for (Iterator<Object> iter = data.iterator(); iter.hasNext(); ) {
final Object obj = iter.next();
final long bulk = (long) iter.next();
DefaultRemoteTraverser<Object> item = new DefaultRemoteTraverser<>(obj, bulk);
- queue.add(new Result(item));
+ resultSet.add(new Result(item));
}
} else {
- data.forEach(item -> queue.add(new Result(item)));
+ data.forEach(item -> resultSet.add(new Result(item)));
}
} else {
// this is a "success" but represents no results otherwise it is an error
if (statusCode != HttpResponseStatus.NO_CONTENT) {
// Save the error because there could be a subsequent HttpContent coming (probably just trailers). All
- // content should be read first before marking the queue or else this channel might get reused too early.
+ // content should be read first before marking the ResultSet or else this channel might get reused too early.
channelHandlerContext.channel().attr(CAUGHT_EXCEPTION).set(
new ResponseException(response.getStatus().getCode(), response.getStatus().getMessage(),
response.getStatus().getException())
@@ -101,12 +100,12 @@ public class GremlinResponseHandler extends SimpleChannelInboundHandler<Response
// Stream is done when the last content signaling response message is read.
if (LAST_CONTENT_READ_RESPONSE == response) {
- final ResultQueue resultQueue = pending.getAndSet(null);
- if (resultQueue != null) {
+ final ResultSet rs = pendingResultSet.getAndSet(null);
+ if (rs != null) {
if (null == channelHandlerContext.channel().attr(CAUGHT_EXCEPTION).get()) {
- resultQueue.markComplete();
+ rs.markComplete();
} else {
- resultQueue.markError(channelHandlerContext.channel().attr(CAUGHT_EXCEPTION).getAndSet(null));
+ rs.markError(channelHandlerContext.channel().attr(CAUGHT_EXCEPTION).getAndSet(null));
}
}
}
@@ -119,8 +118,8 @@ public class GremlinResponseHandler extends SimpleChannelInboundHandler<Response
// there are that many failures someone would take notice and hopefully stop the client.
logger.error("Could not process the response", cause);
- final ResultQueue pendingQueue = pending.getAndSet(null);
- if (pendingQueue != null) pendingQueue.markError(cause);
+ final ResultSet resultSet = this.pendingResultSet.getAndSet(null);
+ if (resultSet != null) resultSet.markError(cause);
if (ExceptionHelper.getRootCause(cause) instanceof SSLException) {
// inbound ssl error can happen with tls 1.3 because client certification auth can fail after the handshake completes
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/AbstractResultQueueTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/AbstractResultSetTest.java
similarity index 78%
rename from gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/AbstractResultQueueTest.java
rename to gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/AbstractResultSetTest.java
index 33b9287999..c2054f2761 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/AbstractResultQueueTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/AbstractResultSetTest.java
@@ -18,31 +18,26 @@
*/
package org.apache.tinkerpop.gremlin.driver;
+import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
import org.junit.After;
import org.junit.Before;
-import java.util.Collections;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
/**
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
-public abstract class AbstractResultQueueTest {
+public abstract class AbstractResultSetTest {
protected ExecutorService pool;
- protected ResultQueue resultQueue;
- protected CompletableFuture<Void> readCompleted;
+ protected ResultSet resultSet;
@Before
public void setup() {
- final LinkedBlockingQueue<Result> resultLinkedBlockingQueue = new LinkedBlockingQueue<>();
- readCompleted = new CompletableFuture<>();
- resultQueue = new ResultQueue(resultLinkedBlockingQueue, readCompleted);
- pool = Executors.newCachedThreadPool();
+ pool = Executors.newCachedThreadPool();
+ resultSet = new ResultSet(pool, RequestMessage.build("traversal").create(), null);
}
@After
@@ -63,15 +58,15 @@ public abstract class AbstractResultQueueTest {
}
/**
- * Adds some test items to the {@link ResultQueue}. This method has the potential to block if the
+ * Adds some test items to the {@link ResultSet}. This method has the potential to block if the
* {@code itemsToWaitFor > 0}.
*
* @param numberOfItemsToAdd the number of items to add in total
- * @param pauseBetweenItemsInMillis amount of time to wait between additions to the {@link ResultQueue}
+ * @param pauseBetweenItemsInMillis amount of time to wait between additions to the {@link ResultSet}
* @param start set to {@code true} to start the thread
- * @param markDone force mark the queue as complete
- * @param itemsToWaitFor block until this many items have been added to the {@link ResultQueue}
- * @return the thread that is doing the work of adding to the queue
+ * @param markDone force mark the result set as complete
+ * @param itemsToWaitFor block until this many items have been added to the {@link ResultSet}
+ * @return the thread that is doing the work of adding to the result set
* @throws Exception
*/
protected Thread addToQueue(final int numberOfItemsToAdd, final long pauseBetweenItemsInMillis,
@@ -85,7 +80,7 @@ public abstract class AbstractResultQueueTest {
final int currentIndex = ix;
pool.submit(() -> {
final Result result = new Result("test-" + currentIndex);
- resultQueue.add(result);
+ resultSet.add(result);
latch.countDown();
}).get();
TimeUnit.MILLISECONDS.sleep(pauseBetweenItemsInMillis);
@@ -94,13 +89,13 @@ public abstract class AbstractResultQueueTest {
}
}
- if (markDone) resultQueue.markComplete();
+ if (markDone) resultSet.markComplete();
- }, "ResultQueueTest-job-submitter");
+ }, "ResultSetTest-job-submitter");
if (start) t.start();
- // wait for the number of items requested to be added to the queue
+ // wait for the number of items requested to be added to the result set
latch.await();
return t;
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultQueueTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetQueuingTest.java
similarity index 67%
rename from gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultQueueTest.java
rename to gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetQueuingTest.java
index f540ddc123..4bc0f5b5b8 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultQueueTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetQueuingTest.java
@@ -22,9 +22,7 @@ import org.apache.commons.lang3.exception.ExceptionUtils;
import org.junit.Test;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -37,23 +35,25 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
/**
+ * Tests for the internal queuing functionality of {@link ResultSet}.
+ *
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
-public class ResultQueueTest extends AbstractResultQueueTest {
+public class ResultSetQueuingTest extends AbstractResultSetTest {
@Test
public void shouldGetSizeUntilError() throws Exception {
final Thread t = addToQueue(100, 10, true, false, 1);
try {
- assertThat(resultQueue.size(), is(greaterThan(0)));
- assertThat(readCompleted.isDone(), is(false));
+ assertThat(resultSet.getAvailableItemCount(), is(greaterThan(0)));
+ assertThat(resultSet.allItemsAvailable(), is(false));
final Exception theProblem = new Exception();
- resultQueue.markError(theProblem);
- assertThat(readCompleted.isDone(), is(true));
+ resultSet.markError(theProblem);
+ assertThat(resultSet.allItemsAvailable(), is(true));
try {
- resultQueue.size();
+ resultSet.getAvailableItemCount();
fail("Should have thrown an exception");
} catch (Exception ex) {
assertEquals(theProblem, ex.getCause());
@@ -65,24 +65,24 @@ public class ResultQueueTest extends AbstractResultQueueTest {
@Test
public void shouldBeEmptyThenNotEmpty() {
- assertThat(resultQueue.isEmpty(), is(true));
- resultQueue.add(new Result("test"));
- assertThat(resultQueue.isEmpty(), is(false));
+ assertThat(resultSet.isEmpty(), is(true));
+ resultSet.add(new Result("test"));
+ assertThat(resultSet.isEmpty(), is(false));
}
@Test
public void shouldNotBeEmptyUntilError() throws Exception {
final Thread t = addToQueue(100, 10, true, false, 1);
try {
- assertThat(resultQueue.isEmpty(), is(false));
- assertThat(readCompleted.isDone(), is(false));
+ assertThat(resultSet.isEmpty(), is(false));
+ assertThat(resultSet.allItemsAvailable(), is(false));
final Exception theProblem = new Exception();
- resultQueue.markError(theProblem);
- assertThat(readCompleted.isDone(), is(true));
+ resultSet.markError(theProblem);
+ assertThat(resultSet.allItemsAvailable(), is(true));
try {
- resultQueue.isEmpty();
+ resultSet.isEmpty();
fail("Should have thrown an exception");
} catch (Exception ex) {
assertEquals(theProblem, ex.getCause());
@@ -96,24 +96,24 @@ public class ResultQueueTest extends AbstractResultQueueTest {
public void shouldDrainUntilError() throws Exception {
final Thread t = addToQueue(100, 10, true, false, 1);
try {
- assertThat(resultQueue.isEmpty(), is(false));
+ assertThat(resultSet.isEmpty(), is(false));
final List<Result> drain = new ArrayList<>();
- resultQueue.drainTo(drain);
+ resultSet.drainTo(drain);
assertThat(drain.size(), is(greaterThan(0)));
- assertThat(readCompleted.isDone(), is(false));
+ assertThat(resultSet.allItemsAvailable(), is(false));
- // make sure some more items get added to the queue before assert
+ // make sure some more items get added to the result set before assert
TimeUnit.MILLISECONDS.sleep(100);
- assertThat(resultQueue.isEmpty(), is(false));
- assertThat(readCompleted.isDone(), is(false));
+ assertThat(resultSet.isEmpty(), is(false));
+ assertThat(resultSet.allItemsAvailable(), is(false));
final Exception theProblem = new Exception();
- resultQueue.markError(theProblem);
- assertThat(readCompleted.isDone(), is(true));
+ resultSet.markError(theProblem);
+ assertThat(resultSet.allItemsAvailable(), is(true));
try {
- resultQueue.drainTo(new ArrayList<>());
+ resultSet.drainTo(new ArrayList<>());
fail("Should have thrown an exception");
} catch (Exception ex) {
assertEquals(theProblem, ex.getCause());
@@ -125,13 +125,13 @@ public class ResultQueueTest extends AbstractResultQueueTest {
@Test
public void shouldAwaitEverythingAndFlushOnMarkCompleted() throws Exception {
- final CompletableFuture<List<Result>> future = resultQueue.await(4);
- resultQueue.add(new Result("test1"));
- resultQueue.add(new Result("test2"));
- resultQueue.add(new Result("test3"));
+ final CompletableFuture<List<Result>> future = resultSet.some(4);
+ resultSet.add(new Result("test1"));
+ resultSet.add(new Result("test2"));
+ resultSet.add(new Result("test3"));
assertThat(future.isDone(), is(false));
- resultQueue.markComplete();
+ resultSet.markComplete();
assertThat(future.isDone(), is(true));
final List<Result> results = future.get();
@@ -140,18 +140,18 @@ public class ResultQueueTest extends AbstractResultQueueTest {
assertEquals("test3", results.get(2).getString());
assertEquals(3, results.size());
- assertThat(resultQueue.isEmpty(), is(true));
+ assertThat(resultSet.isEmpty(), is(true));
}
@Test
public void shouldAwaitFailTheFutureOnMarkError() {
- final CompletableFuture<List<Result>> future = resultQueue.await(4);
- resultQueue.add(new Result("test1"));
- resultQueue.add(new Result("test2"));
- resultQueue.add(new Result("test3"));
+ final CompletableFuture<List<Result>> future = resultSet.some(4);
+ resultSet.add(new Result("test1"));
+ resultSet.add(new Result("test2"));
+ resultSet.add(new Result("test3"));
assertThat(future.isDone(), is(false));
- resultQueue.markError(new Exception("no worky"));
+ resultSet.markError(new Exception("no worky"));
assertThat(future.isDone(), is(true));
try {
@@ -164,14 +164,14 @@ public class ResultQueueTest extends AbstractResultQueueTest {
@Test
public void shouldAwaitToExpectedValueAndDrainOnAdd() throws Exception {
- final CompletableFuture<List<Result>> future = resultQueue.await(3);
- resultQueue.add(new Result("test1"));
- resultQueue.add(new Result("test2"));
+ final CompletableFuture<List<Result>> future = resultSet.some(3);
+ resultSet.add(new Result("test1"));
+ resultSet.add(new Result("test2"));
// shouldn't complete until the third item is in play
assertThat(future.isDone(), is(false));
- resultQueue.add(new Result("test3"));
+ resultSet.add(new Result("test3"));
final List<Result> results = future.get();
assertEquals("test1", results.get(0).getString());
@@ -179,21 +179,21 @@ public class ResultQueueTest extends AbstractResultQueueTest {
assertEquals("test3", results.get(2).getString());
assertEquals(3, results.size());
- assertThat(resultQueue.isEmpty(), is(true));
+ assertThat(resultSet.isEmpty(), is(true));
}
@Test
public void shouldAwaitMultipleToExpectedValueAndDrainOnAdd() throws Exception {
- final CompletableFuture<List<Result>> future1 = resultQueue.await(3);
- final CompletableFuture<List<Result>> future2 = resultQueue.await(1);
- resultQueue.add(new Result("test1"));
- resultQueue.add(new Result("test2"));
+ final CompletableFuture<List<Result>> future1 = resultSet.some(3);
+ final CompletableFuture<List<Result>> future2 = resultSet.some(1);
+ resultSet.add(new Result("test1"));
+ resultSet.add(new Result("test2"));
// shouldn't complete the first future until the third item is in play
assertThat(future1.isDone(), is(false));
assertThat(future2.isDone(), is(false));
- resultQueue.add(new Result("test3"));
+ resultSet.add(new Result("test3"));
final List<Result> results1 = future1.get();
assertEquals("test1", results1.get(0).getString());
@@ -203,7 +203,7 @@ public class ResultQueueTest extends AbstractResultQueueTest {
assertThat(future1.isDone(), is(true));
assertThat(future2.isDone(), is(false));
- resultQueue.add(new Result("test4"));
+ resultSet.add(new Result("test4"));
assertThat(future1.isDone(), is(true));
assertThat(future2.isDone(), is(true));
@@ -211,16 +211,16 @@ public class ResultQueueTest extends AbstractResultQueueTest {
assertEquals("test4", results2.get(0).getString());
assertEquals(1, results2.size());
- assertThat(resultQueue.isEmpty(), is(true));
+ assertThat(resultSet.isEmpty(), is(true));
}
@Test
public void shouldAwaitToExpectedValueAndDrainOnAwait() throws Exception {
- resultQueue.add(new Result("test1"));
- resultQueue.add(new Result("test2"));
- resultQueue.add(new Result("test3"));
+ resultSet.add(new Result("test1"));
+ resultSet.add(new Result("test2"));
+ resultSet.add(new Result("test3"));
- final CompletableFuture<List<Result>> future = resultQueue.await(3);
+ final CompletableFuture<List<Result>> future = resultSet.some(3);
assertThat(future.isDone(), is(true));
final List<Result> results = future.get();
@@ -229,19 +229,19 @@ public class ResultQueueTest extends AbstractResultQueueTest {
assertEquals("test3", results.get(2).getString());
assertEquals(3, results.size());
- assertThat(resultQueue.isEmpty(), is(true));
+ assertThat(resultSet.isEmpty(), is(true));
}
@Test
public void shouldAwaitToReadCompletedAndDrainOnAwait() throws Exception {
- resultQueue.add(new Result("test1"));
- resultQueue.add(new Result("test2"));
- resultQueue.add(new Result("test3"));
+ resultSet.add(new Result("test1"));
+ resultSet.add(new Result("test2"));
+ resultSet.add(new Result("test3"));
- resultQueue.markComplete();
+ resultSet.markComplete();
// you might want 30 but there are only three
- final CompletableFuture<List<Result>> future = resultQueue.await(30);
+ final CompletableFuture<List<Result>> future = resultSet.some(30);
assertThat(future.isDone(), is(true));
final List<Result> results = future.get();
@@ -250,7 +250,7 @@ public class ResultQueueTest extends AbstractResultQueueTest {
assertEquals("test3", results.get(2).getString());
assertEquals(3, results.size());
- assertThat(resultQueue.isEmpty(), is(true));
+ assertThat(resultSet.isEmpty(), is(true));
}
@Test
@@ -262,17 +262,17 @@ public class ResultQueueTest extends AbstractResultQueueTest {
final AtomicInteger count3 = new AtomicInteger(0);
final CountDownLatch latch = new CountDownLatch(3);
- resultQueue.await(500).thenAcceptAsync(r -> {
+ resultSet.some(500).thenAcceptAsync(r -> {
count1.set(r.size());
latch.countDown();
});
- resultQueue.await(150).thenAcceptAsync(r -> {
+ resultSet.some(150).thenAcceptAsync(r -> {
count2.set(r.size());
latch.countDown();
});
- resultQueue.await(350).thenAcceptAsync(r -> {
+ resultSet.some(350).thenAcceptAsync(r -> {
count3.set(r.size());
latch.countDown();
});
@@ -283,7 +283,7 @@ public class ResultQueueTest extends AbstractResultQueueTest {
assertEquals(150, count2.get());
assertEquals(350, count3.get());
- assertThat(resultQueue.isEmpty(), is(true));
+ assertThat(resultSet.isEmpty(), is(true));
} finally {
t.interrupt();
}
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 855d9466a9..4bf3b8c6b7 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
@@ -18,8 +18,6 @@
*/
package org.apache.tinkerpop.gremlin.driver;
-import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
-import org.junit.Before;
import org.junit.Test;
import java.util.Iterator;
@@ -39,20 +37,13 @@ import static org.junit.Assert.fail;
/**
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
-public class ResultSetTest extends AbstractResultQueueTest {
-
- private ResultSet resultSet;
-
- @Before
- public void setupThis() {
- resultSet = new ResultSet(resultQueue, pool, readCompleted, RequestMessage.build("traversal").create(), null);
- }
+public class ResultSetTest extends AbstractResultSetTest {
@Test
public void shouldHaveAllItemsAvailableAsynchronouslyOnReadComplete() throws InterruptedException {
final CompletableFuture<Void> all = resultSet.allItemsAvailableAsync();
assertThat(all.isDone(), is(false));
- readCompleted.complete(null);
+ resultSet.markComplete();
// flush all tasks in pool
pool.awaitTermination(2, TimeUnit.SECONDS);
assertThat(all.isDone(), is(true));
@@ -62,7 +53,7 @@ public class ResultSetTest extends AbstractResultQueueTest {
public void shouldHaveAllItemsAvailableAsynchronouslyOnReadCompleteExceptionally() throws InterruptedException {
final CompletableFuture<Void> all = resultSet.allItemsAvailableAsync();
assertThat(all.isDone(), is(false));
- readCompleted.completeExceptionally(new RuntimeException());
+ resultSet.markError(new RuntimeException());
// flush all tasks in pool
pool.awaitTermination(2, TimeUnit.SECONDS);
assertThat(all.isDone(), is(true));
@@ -72,7 +63,7 @@ public class ResultSetTest extends AbstractResultQueueTest {
@Test
public void shouldHaveAllItemsAvailableOnReadComplete() throws InterruptedException {
assertThat(resultSet.allItemsAvailable(), is(false));
- readCompleted.complete(null);
+ resultSet.markComplete();
// flush all tasks in pool
pool.awaitTermination(2, TimeUnit.SECONDS);
assertThat(resultSet.allItemsAvailable(), is(true));
@@ -85,7 +76,7 @@ public class ResultSetTest extends AbstractResultQueueTest {
final AtomicBoolean atLeastOnce = new AtomicBoolean(false);
addToQueue(1000, 1, true, true);
- while (!readCompleted.isDone()) {
+ while (!resultSet.allItemsAvailable()) {
atLeastOnce.set(true);
if (!atLeastOnce.get())
assertThat(all.isDone(), is(false));
@@ -104,7 +95,7 @@ public class ResultSetTest extends AbstractResultQueueTest {
final AtomicBoolean atLeastOnce = new AtomicBoolean(false);
addToQueue(1000, 1, true, true);
- while (!readCompleted.isDone()) {
+ while (!resultSet.allItemsAvailable()) {
atLeastOnce.set(true);
if (!atLeastOnce.get())
assertThat(resultSet.allItemsAvailable(), is(false));
@@ -117,12 +108,12 @@ public class ResultSetTest extends AbstractResultQueueTest {
@Test
public void shouldAwaitEverythingAndFlushOnMarkCompleted() throws Exception {
final CompletableFuture<List<Result>> future = resultSet.some(4);
- resultQueue.add(new Result("test1"));
- resultQueue.add(new Result("test2"));
- resultQueue.add(new Result("test3"));
+ resultSet.add(new Result("test1"));
+ resultSet.add(new Result("test2"));
+ resultSet.add(new Result("test3"));
assertThat(future.isDone(), is(false));
- resultQueue.markComplete();
+ resultSet.markComplete();
assertThat(future.isDone(), is(true));
final List<Result> results = future.get();
@@ -138,12 +129,12 @@ public class ResultSetTest extends AbstractResultQueueTest {
@Test
public void shouldGetAllOnlyOnComplete() throws Exception {
final CompletableFuture<List<Result>> future = resultSet.all();
- resultQueue.add(new Result("test1"));
- resultQueue.add(new Result("test2"));
- resultQueue.add(new Result("test3"));
+ resultSet.add(new Result("test1"));
+ resultSet.add(new Result("test2"));
+ resultSet.add(new Result("test3"));
assertThat(future.isDone(), is(false));
- resultQueue.markComplete();
+ resultSet.markComplete();
final List<Result> results = future.get();
assertEquals("test1", results.get(0).getString());
(tinkerpop) 02/02: Add HTTP transaction support to gremlin-driver
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch master-tx-client
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 77732f46432383cc636020704deacacc5f53de79
Author: Ken Hu <10...@users.noreply.github.com>
AuthorDate: Fri Mar 20 14:38:38 2026 -0700
Add HTTP transaction support to gremlin-driver
Introduce HttpRemoteTransaction and TransactionRemoteConnection to
manage transaction lifecycle. RequestSubmitter and RequestSubmitterAsync
interfaces were extracted out of Client to ensure transactions would
have the same submission interface. Since Clients and Transactions now have
a similar submission interface, they are both obtained from a Cluster.
Transactions are meant to be short-lived then closed and Clients remain
long-lived and preferrably re-used.
---
CHANGELOG.asciidoc | 1 +
.../traversal/dsl/graph/GraphTraversalSource.java | 13 +-
.../apache/tinkerpop/gremlin/driver/Client.java | 69 ++-
.../apache/tinkerpop/gremlin/driver/Cluster.java | 82 +--
.../gremlin/driver/RemoteTransaction.java | 36 ++
.../tinkerpop/gremlin/driver/RequestOptions.java | 31 ++
.../tinkerpop/gremlin/driver/RequestSubmitter.java | 61 ++
.../gremlin/driver/RequestSubmitterAsync.java | 67 +++
.../driver/handler/HttpGremlinRequestEncoder.java | 7 +
.../driver/remote/DriverRemoteConnection.java | 15 +-
.../driver/remote/HttpRemoteTransaction.java | 317 +++++++++++
.../driver/remote/TransactionRemoteConnection.java | 106 ++++
.../gremlin/server/GremlinDriverIntegrateTest.java | 4 -
.../GremlinDriverTransactionIntegrateTest.java | 617 +++++++++++++++++++++
.../gremlin/server/HttpDriverIntegrateTest.java | 16 -
15 files changed, 1369 insertions(+), 73 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index fe123a3fae..5624866604 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -28,6 +28,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Added `__contains__` and `keys()` to `Element` in `gremlin-python`.
* Added `subgraph()` support for `gremlin-python` so that results are stored in a detached `Graph` object.
* Added support for remote transactions to the `gremlin-server` through `TransactionManager` and `UnmanagedTransaction`.
+* Added support for transactions to `gremlin-driver` using the new `HttpRemoteTransaction`.
* Modified grammar to make `discard()` usage more consistent as a filter step where it can now be used to chain additional traversal steps and be used anonymously.
* Removed `Meta` field from `ResponseResult` struct in `gremlin-go`.
* Removed deprecated elements of the Java-based process testing suite: `ProcessStandardSuite`, `ProcessComputerSuite`, `ProcessLimitedSuite` and associated tests.
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/dsl/graph/GraphTraversalSource.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/dsl/graph/GraphTraversalSource.java
index b1d417ee02..684d9e8609 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/dsl/graph/GraphTraversalSource.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/dsl/graph/GraphTraversalSource.java
@@ -703,14 +703,19 @@ public class GraphTraversalSource implements TraversalSource {
/**
* Proxies calls through to the underlying {@link Graph#tx()} or to the {@link RemoteConnection#tx()}.
+ * <p>
+ * When a remote connection is present, this method delegates to the connection's
+ * {@link RemoteConnection#tx()} method, which returns an appropriate transaction
+ * implementation for the remote connection type (e.g., {@code HttpRemoteTransaction}
+ * for HTTP-based connections).
+ *
+ * @return A {@link Transaction} for managing transactional operations
*/
public Transaction tx() {
if (null == this.connection)
return this.graph.tx();
- else {
- throw new UnsupportedOperationException("TinkerPop 4 does not yet support remote transactions");
- }
-
+ else
+ return this.connection.tx();
}
/**
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
index 95af4270f5..26c24f2da2 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
@@ -53,7 +53,7 @@ import java.util.stream.Collectors;
*
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
-public abstract class Client {
+public abstract class Client implements RequestSubmitter, RequestSubmitterAsync {
private static final Logger logger = LoggerFactory.getLogger(Client.class);
public static final String TOO_MANY_IN_FLIGHT_REQUESTS = "Number of active requests (%s) exceeds pool size (%s). " +
@@ -239,6 +239,7 @@ public abstract class Client {
options.getLanguage().ifPresent(lang -> request.addLanguage(lang));
options.getMaterializeProperties().ifPresent(mp -> request.addMaterializeProperties(mp));
options.getBulkResults().ifPresent(bulked -> request.addBulkResults(Boolean.parseBoolean(bulked)));
+ options.getTransactionId().ifPresent(transactionId -> request.addTransactionId(transactionId));
return submitAsync(request.create());
}
@@ -286,6 +287,13 @@ public abstract class Client {
return cluster;
}
+ protected Host chooseRandomHost() {
+ cluster.init();
+ final List<Host> hosts = new ArrayList<>(cluster.allHosts());
+ final int ix = random.nextInt(hosts.size());
+ return hosts.get(ix);
+ }
+
/**
* A {@code Client} implementation. Requests are sent to multiple servers given a {@link LoadBalancingStrategy}.
* Transactions are automatically committed (or rolled-back on error) after each request.
@@ -358,12 +366,6 @@ public abstract class Client {
return pool.borrowConnection(cluster.connectionPoolSettings().maxWaitForConnection, TimeUnit.MILLISECONDS);
}
- private Host chooseRandomHost() {
- final List<Host> hosts = new ArrayList<>(cluster.allHosts());
- final int ix = random.nextInt(hosts.size());
- return hosts.get(ix);
- }
-
/**
* Initializes the connection pools on all hosts.
*/
@@ -561,5 +563,58 @@ public abstract class Client {
if (close.isDone()) throw new IllegalStateException("Client is closed");
return new AliasClusteredClient(client, graphOrTraversalSource);
}
+
+ }
+
+ /**
+ * A {@link Client} that pins all requests to a single {@link Host}. Used internally by transactions
+ * to ensure all requests within a transaction go to the same server.
+ * <p>
+ * This client is not intended to be used directly — obtain a {@link org.apache.tinkerpop.gremlin.structure.Transaction}
+ * via {@link Cluster#transact()} or {@link Cluster#transact(String)} instead.
+ */
+ public static class PinnedClient extends Client {
+ private final ClusteredClient clusteredClient;
+ private final Host pinnedHost;
+ private final AtomicReference<CompletableFuture<Void>> closing = new AtomicReference<>(null);
+
+ PinnedClient(final Cluster cluster) {
+ super(cluster);
+ this.pinnedHost = chooseRandomHost();
+ this.clusteredClient = cluster.connect();
+ }
+
+ public Host getPinnedHost() {
+ return pinnedHost;
+ }
+
+ @Override
+ protected void initializeImplementation() {
+ this.clusteredClient.init();
+ initialized = true;
+ }
+
+ @Override
+ protected Connection chooseConnection(final RequestMessage msg) throws TimeoutException, ConnectionException {
+ final ConnectionPool pool = clusteredClient.hostConnectionPools.get(pinnedHost);
+ if (pool == null) throw new NoHostAvailableException();
+ return pool.borrowConnection(cluster.connectionPoolSettings().maxWaitForConnection, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public boolean isClosing() {
+ return closing.get() != null;
+ }
+
+ /**
+ * Marks this client as closed. The underlying pool is owned by {@link ClusteredClient} and is not closed here.
+ */
+ @Override
+ public synchronized CompletableFuture<Void> closeAsync() {
+ if (closing.get() != null) return closing.get();
+
+ closing.set(clusteredClient.closeAsync());
+ return closing.get();
+ }
}
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
index e080da8d51..cac193814c 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
@@ -31,7 +31,10 @@ import org.apache.commons.configuration2.Configuration;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.tinkerpop.gremlin.driver.auth.Auth;
+import org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
import org.apache.tinkerpop.gremlin.driver.interceptor.PayloadSerializingInterceptor;
+import org.apache.tinkerpop.gremlin.driver.remote.HttpRemoteTransaction;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
import org.apache.tinkerpop.gremlin.util.MessageSerializer;
import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
import org.apache.tinkerpop.gremlin.util.ser.GraphBinaryMessageSerializerV4;
@@ -63,6 +66,7 @@ import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -94,49 +98,31 @@ public final class Cluster {
}
/**
- * Creates a SessionedClient instance to this {@code Cluster}, meaning requests will be routed to
- * a single server (randomly selected from the cluster), where the same bindings will be available on each request.
- * Requests are bound to the same thread on the server and thus transactions may extend beyond the bounds of a
- * single request. The transactions are managed by the user and must be committed or rolled-back manually.
- * <p/>
- * Note that calling this method does not imply that a connection is made to the server itself at this point.
- * Therefore, if there is only one server specified in the {@code Cluster} and that server is not available an
- * error will not be raised at this point. Connections get initialized in the {@link Client} when a request is
- * submitted or can be directly initialized via {@link Client#init()}.
- *
- * @param sessionId user supplied id for the session which should be unique (a UUID is ideal).
+ * Creates a new {@link Client} based on the settings provided.
*/
- public <T extends Client> T connect(final String sessionId) {
- throw new UnsupportedOperationException("not implemented");
+ public <T extends Client> T connect() {
+ final Client client = new Client.ClusteredClient(this);
+ manager.trackClient(client);
+ return (T) client;
}
/**
- * Creates a SessionedClient instance to this {@code Cluster}, meaning requests will be routed to
- * a single server (randomly selected from the cluster), where the same bindings will be available on each request.
- * Requests are bound to the same thread on the server and thus transactions may extend beyond the bounds of a
- * single request. If {@code manageTransactions} is set to {@code false} then transactions are managed by the
- * user and must be committed or rolled-back manually. When set to {@code true} the transaction is committed or
- * rolled-back at the end of each request.
- * <p/>
- * Note that calling this method does not imply that a connection is made to the server itself at this point.
- * Therefore, if there is only one server specified in the {@code Cluster} and that server is not available an
- * error will not be raised at this point. Connections get initialized in the {@link Client} when a request is
- * submitted or can be directly initialized via {@link Client#init()}.
- *
- * @param sessionId user supplied id for the session which should be unique (a UUID is ideal).
- * @param manageTransactions enables auto-transactions when set to true
+ * Creates a new {@link Transaction} using the server's default traversal source.
+ * The server will bind to "g" by default when no traversal source is specified.
*/
- public <T extends Client> T connect(final String sessionId, final boolean manageTransactions) {
- throw new UnsupportedOperationException("not implemented");
+ public RemoteTransaction transact() {
+ return transact(null);
}
/**
- * Creates a new {@link Client} based on the settings provided.
+ * Creates a new {@link Transaction} bound to the specified graph or traversal source.
+ *
+ * @param graphOrTraversalSource the graph/traversal source alias, or null to use the server default
*/
- public <T extends Client> T connect() {
- final Client client = new Client.ClusteredClient(this);
- manager.trackClient(client);
- return (T) client;
+ public RemoteTransaction transact(final String graphOrTraversalSource) {
+ final Client.PinnedClient pinnedClient = new Client.PinnedClient(this);
+ manager.trackClient(pinnedClient);
+ return new HttpRemoteTransaction(pinnedClient, graphOrTraversalSource);
}
@Override
@@ -365,6 +351,14 @@ public final class Cluster {
return Collections.unmodifiableCollection(manager.allHosts());
}
+ public void trackTransaction(final HttpRemoteTransaction tx) {
+ manager.trackTransaction(tx);
+ }
+
+ public void untrackTransaction(final HttpRemoteTransaction tx) {
+ manager.untrackTransaction(tx);
+ }
+
Factory getFactory() {
return manager.factory;
}
@@ -936,6 +930,7 @@ public final class Cluster {
class Manager {
private final ConcurrentMap<InetSocketAddress, Host> hosts = new ConcurrentHashMap<>();
+ private final Set<HttpRemoteTransaction> openTransactions = ConcurrentHashMap.newKeySet();
private boolean initialized;
private final List<InetSocketAddress> contactPoints;
private final Factory factory;
@@ -1081,6 +1076,14 @@ public final class Cluster {
openedClients.add(new WeakReference<>(client));
}
+ void trackTransaction(final HttpRemoteTransaction tx) {
+ openTransactions.add(tx);
+ }
+
+ void untrackTransaction(final HttpRemoteTransaction tx) {
+ openTransactions.remove(tx);
+ }
+
public Host add(final InetSocketAddress address) {
final Host newHost = new Host(address, Cluster.this);
final Host previous = hosts.putIfAbsent(address, newHost);
@@ -1096,6 +1099,17 @@ public final class Cluster {
if (closeFuture.get() != null)
return closeFuture.get();
+ // best-effort rollback of any open transactions before closing
+ // snapshot to avoid concurrent modification since rollback() calls untrackTransaction()
+ new ArrayList<>(openTransactions).forEach(tx -> {
+ try {
+ tx.rollback();
+ } catch (Exception e) {
+ logger.warn("Failed to rollback transaction on cluster close", e);
+ }
+ });
+ openTransactions.clear();
+
final List<CompletableFuture<Void>> clientCloseFutures = new ArrayList<>(openedClients.size());
for (WeakReference<Client> openedClient : openedClients) {
final Client client = openedClient.get();
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RemoteTransaction.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RemoteTransaction.java
new file mode 100644
index 0000000000..d4505f6d8c
--- /dev/null
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RemoteTransaction.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.driver;
+
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+
+/**
+ * A {@link Transaction} interface for remote connections that combine lifecycle management and synchronous submission.
+ *
+ * Note: Implementations of this interface are generally <b>NOT</b> thread-safe.
+ */
+public interface RemoteTransaction extends Transaction, RequestSubmitter {
+ /**
+ * Returns the server-generated transaction ID, or {@code null} if the transaction has not yet been started via
+ * {@link #begin(Class)}.
+ *
+ * @return the transaction ID, or null if not yet begun
+ */
+ String getTransactionId();
+}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestOptions.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestOptions.java
index a43e186164..c51f2b8516 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestOptions.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestOptions.java
@@ -50,6 +50,7 @@ public final class RequestOptions {
private final String language;
private final String materializeProperties;
private final String bulkResults;
+ private final String transactionId;
private RequestOptions(final Builder builder) {
this.graphOrTraversalSource = builder.graphOrTraversalSource;
@@ -59,6 +60,7 @@ public final class RequestOptions {
this.language = builder.language;
this.materializeProperties = builder.materializeProperties;
this.bulkResults = builder.bulkResults;
+ this.transactionId = builder.transactionId;
}
public Optional<String> getG() {
@@ -85,6 +87,8 @@ public final class RequestOptions {
public Optional<String> getBulkResults() { return Optional.ofNullable(bulkResults); }
+ public Optional<String> getTransactionId() { return Optional.ofNullable(transactionId); }
+
public static Builder build() {
return new Builder();
}
@@ -125,6 +129,25 @@ public final class RequestOptions {
private String materializeProperties = null;
private String language = null;
private String bulkResults = null;
+ private String transactionId = null;
+
+ /**
+ * Creates a {@link Builder} populated with the values from the provided {@link RequestOptions}.
+ * @param options the options to copy from
+ * @return a {@link Builder} with the copied options
+ */
+ public static Builder from(final RequestOptions options) {
+ final Builder builder = build();
+ builder.graphOrTraversalSource = options.graphOrTraversalSource;
+ builder.parameters = options.parameters;
+ builder.batchSize = options.batchSize;
+ builder.timeout = options.timeout;
+ builder.materializeProperties = options.materializeProperties;
+ builder.language = options.language;
+ builder.bulkResults = options.bulkResults;
+ builder.transactionId = options.transactionId;
+ return builder;
+ }
/**
* The aliases to set on the request.
@@ -196,6 +219,14 @@ public final class RequestOptions {
return this;
}
+ /**
+ * Sets the transactionId value to be sent on the request.
+ */
+ public Builder transactionId(final String id) {
+ this.transactionId = id;
+ return this;
+ }
+
public RequestOptions create() {
return new RequestOptions(this);
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestSubmitter.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestSubmitter.java
new file mode 100644
index 0000000000..3afc8e099d
--- /dev/null
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestSubmitter.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.driver;
+
+import java.util.Map;
+
+/**
+ * Defines the synchronous request submission contract for Gremlin requests.
+ * <p>
+ * This interface is implemented by both {@link Client} and transaction classes to ensure a consistent API for
+ * submitting Gremlin scripts. The synchronous nature of these methods means they block until the request completes.
+ * <p>
+ * For asynchronous submission, see {@link RequestSubmitterAsync}.
+ */
+public interface RequestSubmitter {
+
+ /**
+ * Submits a Gremlin script and blocks until the response is received.
+ *
+ * @param gremlin the Gremlin script to execute
+ * @return the results of the script execution
+ */
+ ResultSet submit(String gremlin);
+
+ /**
+ * Submits a Gremlin script with bound parameters and blocks until the response is received.
+ * <p>
+ * Prefer this method over string concatenation when executing scripts with variable
+ * arguments, as parameterized scripts perform better.
+ *
+ * @param gremlin the Gremlin script to execute
+ * @param parameters a map of parameters that will be bound to the script on execution
+ * @return the results of the script execution
+ */
+ ResultSet submit(String gremlin, Map<String, Object> parameters);
+
+ /**
+ * Submits a Gremlin script with request options and blocks until the response is received.
+ *
+ * @param gremlin the Gremlin script to execute
+ * @param options the options to supply for this request
+ * @return the results of the script execution
+ */
+ ResultSet submit(String gremlin, RequestOptions options);
+}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestSubmitterAsync.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestSubmitterAsync.java
new file mode 100644
index 0000000000..c94ed5bb72
--- /dev/null
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestSubmitterAsync.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.driver;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Defines the asynchronous request submission contract for Gremlin requests.
+ * <p>
+ * This interface is implemented by {@link Client} to provide non-blocking request submission. The returned
+ * {@link CompletableFuture} completes when the write of the request is complete, not when the response is received.
+ * <p>
+ * Note: Transaction classes intentionally do not implement this interface because transactional operations require
+ * sequential execution to maintain ordering guarantees over HTTP.
+ * <p>
+ * For synchronous submission, see {@link RequestSubmitter}.
+ */
+public interface RequestSubmitterAsync {
+
+ /**
+ * Submits a Gremlin script asynchronously.
+ * <p>
+ * The returned future completes when the write of the request is complete.
+ *
+ * @param gremlin the Gremlin script to execute
+ * @return a future that completes with the results when the request write is complete
+ */
+ CompletableFuture<ResultSet> submitAsync(String gremlin);
+
+ /**
+ * Submits a Gremlin script with bound parameters asynchronously.
+ * <p>
+ * Prefer this method over string concatenation when executing scripts with variable
+ * arguments, as parameterized scripts perform better.
+ *
+ * @param gremlin the Gremlin script to execute
+ * @param parameters a map of parameters that will be bound to the script on execution
+ * @return a future that completes with the results when the request write is complete
+ */
+ CompletableFuture<ResultSet> submitAsync(String gremlin, Map<String, Object> parameters);
+
+ /**
+ * Submits a Gremlin script with request options asynchronously.
+ *
+ * @param gremlin the Gremlin script to execute
+ * @param options the options to supply for this request
+ * @return a future that completes with the results when the request write is complete
+ */
+ CompletableFuture<ResultSet> submitAsync(String gremlin, RequestOptions options);
+}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java
index c94d74264d..0f8ede4c7f 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java
@@ -91,6 +91,13 @@ public final class HttpGremlinRequestEncoder extends MessageToMessageEncoder<Req
if (bulkResults) {
headersMap.put(Tokens.BULK_RESULTS, "true");
}
+
+ // Add X-Transaction-Id header to comply with specification's dual transmission (header and body)
+ final String transactionId = requestMessage.getField(Tokens.ARGS_TRANSACTION_ID);
+ if (transactionId != null) {
+ headersMap.put(Tokens.Headers.TRANSACTION_ID, transactionId);
+ }
+
HttpRequest gremlinRequest = new HttpRequest(headersMap, requestMessage, uri);
for (final Pair<String, ? extends RequestInterceptor> interceptor : interceptors) {
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
index c8d56814c4..c909e48604 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
@@ -248,15 +248,14 @@ public class DriverRemoteConnection implements RemoteConnection {
}
/**
- * Constructs a new {@link DriverRemoteTransaction}. Not yet supported in TinkerPop 4.
+ * Creates a new {@link HttpRemoteTransaction} for executing transactional operations.
+ *
+ * @return A new {@link HttpRemoteTransaction}
*/
-// @Override
-// public Transaction tx() {
-// // todo: not implemented
-// final DriverRemoteConnection session = new DriverRemoteConnection(
-// client.getCluster().connect(), remoteTraversalSourceName, true);
-// return new DriverRemoteTransaction(session);
-// }
+ @Override
+ public Transaction tx() {
+ return client.getCluster().transact(remoteTraversalSourceName);
+ }
@Override
public String toString() {
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/HttpRemoteTransaction.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/HttpRemoteTransaction.java
new file mode 100644
index 0000000000..a801fd197a
--- /dev/null
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/HttpRemoteTransaction.java
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.driver.remote;
+
+import org.apache.tinkerpop.gremlin.driver.Client;
+import org.apache.tinkerpop.gremlin.driver.Cluster;
+import org.apache.tinkerpop.gremlin.driver.Host;
+import org.apache.tinkerpop.gremlin.driver.RemoteTransaction;
+import org.apache.tinkerpop.gremlin.driver.RequestOptions;
+import org.apache.tinkerpop.gremlin.driver.Result;
+import org.apache.tinkerpop.gremlin.driver.ResultSet;
+import org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
+import org.apache.tinkerpop.gremlin.process.remote.RemoteConnection;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+import org.apache.tinkerpop.gremlin.structure.util.TransactionException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+/**
+ * A {@link Transaction} implementation for HTTP-based remote connections.
+ * <p>
+ * This class provides synchronous, sequential request execution within a transaction context.
+ * All requests are pinned to a single host and include the transaction ID (after begin()).
+ * <p>
+ * Key characteristics:
+ * <ul>
+ * <li>Synchronous API only - no submitAsync() methods</li>
+ * <li>Host pinning - all requests go to the same server</li>
+ * <li>Sequential execution - requests block until complete</li>
+ * </ul>
+ * <p>
+ * Usage:
+ * <pre>
+ * Transaction tx = cluster.transact("g");
+ * GraphTraversalSource gtx = tx.begin();
+ * gtx.addV("person").property("name", "alice").iterate();
+ * tx.commit();
+ * </pre>
+ *
+ * This class is <b>NOT</b> thread-safe.
+ */
+public class HttpRemoteTransaction implements RemoteTransaction {
+ private static final Logger logger = LoggerFactory.getLogger(HttpRemoteTransaction.class);
+ private static final long CLOSING_MAX_WAIT_MS = 10000;
+
+ protected Consumer<Transaction> closeConsumer = CLOSE_BEHAVIOR.COMMIT;
+ private final Client.PinnedClient pinnedClient;
+ private final Cluster cluster;
+ private final Host pinnedHost;
+ private final String graphAlias;
+ private String transactionId; // null until begin(), set from server response
+ private TransactionState state = TransactionState.NOT_STARTED;
+
+ private enum TransactionState {
+ NOT_STARTED, OPEN, CLOSED
+ }
+
+ /**
+ * Creates a new HTTP transaction.
+ * <p>
+ * The transaction is not started until {@link #begin(Class)} is called.
+ * A host is selected at creation time and all requests will be pinned to it.
+ *
+ * @param pinnedClient the underlying client for connection access
+ * @param graphAlias the graph/traversal source alias (e.g., "g")
+ * @throws NoHostAvailableException if no hosts are available in the cluster
+ */
+ public HttpRemoteTransaction(final Client.PinnedClient pinnedClient, final String graphAlias) {
+ this.pinnedClient = pinnedClient;
+ this.graphAlias = graphAlias;
+ this.pinnedHost = pinnedClient.getPinnedHost();
+ this.cluster = pinnedClient.getCluster();
+ }
+
+ /**
+ * Not supported for remote transactions. Use {@link #begin(Class)} instead.
+ *
+ * @throws UnsupportedOperationException always
+ */
+ @Override
+ public void open() {
+ begin();
+ }
+
+ /**
+ * Starts a transaction and returns a traversal source bound to it.
+ * <p>
+ * This method sends {@code g.tx().begin()} to the server, which returns
+ * the transaction ID. All subsequent requests will include this ID.
+ *
+ * @param traversalSourceClass the class of the traversal source to create
+ * @param <T> the type of the traversal source
+ * @return a new traversal source bound to this transaction
+ * @throws IllegalStateException if the transaction is already started
+ * @throws RuntimeException if the transaction fails to begin
+ */
+ @Override
+ public <T extends TraversalSource> T begin(final Class<T> traversalSourceClass) {
+ if (state != TransactionState.NOT_STARTED) {
+ throw new IllegalStateException("Transaction already started");
+ }
+ cluster.trackTransaction(this);
+
+ try {
+ // Send begin - no txId attached yet
+ final ResultSet rs = submitInternal("g.tx().begin()");
+
+ // Server returns the transaction ID
+ this.transactionId = extractTransactionId(rs);
+ this.state = TransactionState.OPEN;
+ } catch (Exception e) {
+ cleanUp();
+ throw new RuntimeException("Failed to begin transaction: " + e.getMessage(), e);
+ }
+
+ // Create RemoteConnection for the traversal source
+ final TransactionRemoteConnection txConnection = new TransactionRemoteConnection(this);
+
+ try {
+ return traversalSourceClass.getConstructor(RemoteConnection.class).newInstance(txConnection);
+ } catch (Exception e) {
+ rollback();
+ throw new IllegalStateException("Failed to create TraversalSource", e);
+ }
+ }
+
+ /**
+ * Extracts the transaction ID from the begin() response.
+ * <p>
+ * The server returns the transaction ID as part of the response to g.tx().begin().
+ *
+ * @param rs the result set from the begin request
+ * @return the transaction ID
+ */
+ private String extractTransactionId(final ResultSet rs) {
+ // Wait for all results and extract the transaction ID
+ final List<Result> results = rs.all().join();
+ if (results.isEmpty()) {
+ throw new IllegalStateException("Server did not return transaction ID");
+ }
+ try {
+ final Object id = results.get(0).get(Map.class).get("transactionId");
+ if (id == null) throw new IllegalStateException("Server did not return transaction ID");
+
+ final String idStr = id.toString();
+ if (idStr.isBlank()) throw new IllegalStateException("Server returned empty transaction ID");
+
+ return idStr;
+ } catch (Exception e) {
+ throw new IllegalStateException("Server did not return transaction ID");
+ }
+ }
+
+ /**
+ * Commits the transaction.
+ * <p>
+ * Sends {@code g.tx().commit()} to the server and closes the transaction.
+ *
+ * @throws IllegalStateException if the transaction is not open
+ * @throws RuntimeException if the commit fails
+ */
+ @Override
+ public void commit() {
+ closeRemoteTransaction("g.tx().commit()");
+ }
+
+ /**
+ * Rolls back the transaction.
+ * <p>
+ * Sends {@code g.tx().rollback()} to the server and closes the transaction.
+ * This is best-effort - errors are logged but not thrown.
+ */
+ @Override
+ public void rollback() {
+ closeRemoteTransaction("g.tx().rollback()");
+ }
+
+ private void closeRemoteTransaction(final String closeScript) {
+ if (state != TransactionState.OPEN) throw new IllegalStateException("Transaction is not open");
+
+ try {
+ submitInternal(closeScript).all().get(CLOSING_MAX_WAIT_MS, TimeUnit.MILLISECONDS);
+ cleanUp();
+ } catch (Exception e) {
+ logger.warn("Failed to {} transaction on {}", closeScript, pinnedHost);
+ throw new TransactionException("Failed to " + closeScript, e);
+ }
+ }
+
+ private void cleanUp() {
+ state = TransactionState.CLOSED;
+ cluster.untrackTransaction(this);
+ }
+
+ /**
+ * Returns the server-generated transaction ID, or {@code null} if the transaction
+ * has not yet been started via {@link #begin(Class)}.
+ *
+ * @return the transaction ID, or null if not yet begun
+ */
+ public String getTransactionId() {
+ return transactionId;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return state == TransactionState.OPEN;
+ }
+
+ @Override
+ public void readWrite() {
+ throw new UnsupportedOperationException("Remote transaction behaviors are not configurable - they are always manually controlled");
+ }
+
+ @Override
+ public void close() {
+ closeConsumer.accept(this);
+
+ // this is just for safety in case of custom closeConsumer but should normally be handled by commit/rollback
+ cleanUp();
+ }
+
+ @Override
+ public Transaction onReadWrite(final Consumer<Transaction> consumer) {
+ throw new UnsupportedOperationException("Remote transaction behaviors are not configurable - they are always manually controlled");
+ }
+
+ @Override
+ public Transaction onClose(final Consumer<Transaction> consumer) {
+ this.closeConsumer = consumer;
+ return this;
+ }
+
+ @Override
+ public void addTransactionListener(final Consumer<Status> listener) {
+ throw new UnsupportedOperationException("Remote transactions cannot have listeners attached");
+ }
+
+ @Override
+ public void removeTransactionListener(final Consumer<Status> listener) {
+ throw new UnsupportedOperationException("Remote transactions cannot have listeners attached");
+ }
+
+ @Override
+ public void clearTransactionListeners() {
+ throw new UnsupportedOperationException("Remote transactions cannot have listeners attached");
+ }
+
+ @Override
+ public ResultSet submit(final String gremlin) {
+ return submit(gremlin, RequestOptions.EMPTY);
+ }
+
+ @Override
+ public ResultSet submit(final String gremlin, final Map<String, Object> parameters) {
+ final RequestOptions.Builder builder = RequestOptions.build();
+ if (parameters != null && !parameters.isEmpty()) {
+ parameters.forEach(builder::addParameter);
+ }
+ return submit(gremlin, builder.create());
+ }
+
+ @Override
+ public ResultSet submit(final String gremlin, final RequestOptions options) {
+ if (state != TransactionState.OPEN) {
+ throw new IllegalStateException("Transaction is not open");
+ }
+ return submitInternal(gremlin, options);
+ }
+
+ private ResultSet submitInternal(final String gremlin) {
+ return submitInternal(gremlin, RequestOptions.EMPTY);
+ }
+
+ // synchronized here is a bit defensive but ensures that even if a user accidentally uses this in different threads,
+ // the server will still receive the requests in the correct order
+ private synchronized ResultSet submitInternal(final String gremlin, final RequestOptions options) {
+ final RequestOptions.Builder builder = RequestOptions.Builder.from(options);
+ if (graphAlias != null) {
+ // Don't allow per-request override of "g" as transactions should only target a single Graph instance.
+ builder.addG(graphAlias);
+ }
+
+ // Attach txId if we have one (not present for begin())
+ if (transactionId != null) {
+ builder.transactionId(transactionId);
+ }
+
+ try {
+ return pinnedClient.submit(gremlin, builder.create());
+ } catch (Exception e) {
+ throw new RuntimeException("Transaction request failed: " + e.getMessage(), e);
+ }
+ }
+}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/TransactionRemoteConnection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/TransactionRemoteConnection.java
new file mode 100644
index 0000000000..6ea9ce5e28
--- /dev/null
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/TransactionRemoteConnection.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.driver.remote;
+
+import org.apache.tinkerpop.gremlin.driver.RequestOptions;
+import org.apache.tinkerpop.gremlin.driver.ResultSet;
+import org.apache.tinkerpop.gremlin.process.remote.RemoteConnection;
+import org.apache.tinkerpop.gremlin.process.remote.RemoteConnectionException;
+import org.apache.tinkerpop.gremlin.process.remote.traversal.RemoteTraversal;
+import org.apache.tinkerpop.gremlin.process.traversal.GremlinLang;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.tinkerpop.gremlin.driver.RequestOptions.getRequestOptions;
+
+/**
+ * A {@link RemoteConnection} that routes all submissions through an {@link HttpRemoteTransaction}.
+ * <p>
+ * This connection adapts the synchronous transaction API to the async {@link RemoteConnection}
+ * interface required by {@link org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource}.
+ * <p>
+ */
+class TransactionRemoteConnection implements RemoteConnection {
+
+ private final HttpRemoteTransaction transaction;
+
+ /**
+ * Creates a new connection bound to the specified transaction.
+ *
+ * @param transaction the transaction that owns this connection
+ */
+ TransactionRemoteConnection(final HttpRemoteTransaction transaction) {
+ this.transaction = transaction;
+ }
+
+ /**
+ * Submits a traversal through the transaction.
+ * <p>
+ * The submission is synchronous internally but returns a completed future
+ * to satisfy the {@link RemoteConnection} interface.
+ *
+ * @param gremlinLang the traversal to submit
+ * @return a completed future with the traversal results
+ * @throws RemoteConnectionException if the transaction is not open or submission fails
+ */
+ @Override
+ public <E> CompletableFuture<RemoteTraversal<?, E>> submitAsync(final GremlinLang gremlinLang)
+ throws RemoteConnectionException {
+ if (!transaction.isOpen()) {
+ throw new RemoteConnectionException("Transaction is not open");
+ }
+
+ try {
+ // Synchronous submission through transaction
+ final ResultSet rs = transaction.submit(gremlinLang.getGremlin(), getRequestOptions(gremlinLang));
+
+ final RemoteTraversal<?, E> traversal = new DriverRemoteTraversal<>(rs,
+ null, // client not needed for iteration
+ false, // attachElements
+ Optional.empty());
+
+ return CompletableFuture.completedFuture(traversal);
+ } catch (Exception e) {
+ throw new RemoteConnectionException(e);
+ }
+ }
+
+ /**
+ * Returns the owning transaction.
+ *
+ * @return the transaction that owns this connection
+ */
+ @Override
+ public Transaction tx() {
+ return transaction;
+ }
+
+ /**
+ * No-op close implementation.
+ * <p>
+ * The transaction manages its own lifecycle - users should call
+ * {@link Transaction#commit()} or {@link Transaction#rollback()} explicitly.
+ */
+ @Override
+ public void close() {
+ // Transaction manages its own lifecycle - don't close it here
+ }
+}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
index 7b051812c7..89ad207135 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
@@ -1094,13 +1094,11 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
public void shouldCloseAllClientsOnCloseOfCluster() throws Exception {
final Cluster cluster = TestClientFactory.open();
final Client sessionlessOne = cluster.connect();
- final Client session = cluster.connect("session");
final Client sessionlessTwo = cluster.connect();
final Client sessionlessThree = cluster.connect();
final Client sessionlessFour = cluster.connect();
assertEquals(2, sessionlessOne.submit("g.inject(2)").all().get().get(0).getInt());
- assertEquals(2, session.submit("g.inject(2)").all().get().get(0).getInt());
assertEquals(2, sessionlessTwo.submit("g.inject(2)").all().get().get(0).getInt());
assertEquals(2, sessionlessThree.submit("g.inject(2)").all().get().get(0).getInt());
// dont' send anything on the 4th client
@@ -1119,7 +1117,6 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
}
try {
- session.submit("g.inject(2)").all().get();
fail("Should have tossed an exception because cluster was closed");
} catch (Exception ex) {
final Throwable root = ExceptionHelper.getRootCause(ex);
@@ -1156,7 +1153,6 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
// allow call to close() even though closed through cluster
sessionlessOne.close();
- session.close();
sessionlessTwo.close();
cluster.close();
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverTransactionIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverTransactionIntegrateTest.java
new file mode 100644
index 0000000000..2198db5525
--- /dev/null
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverTransactionIntegrateTest.java
@@ -0,0 +1,617 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.server;
+
+import org.apache.tinkerpop.gremlin.driver.Client;
+import org.apache.tinkerpop.gremlin.driver.Cluster;
+import org.apache.tinkerpop.gremlin.driver.RemoteTransaction;
+import org.apache.tinkerpop.gremlin.driver.RequestOptions;
+import org.apache.tinkerpop.gremlin.driver.Result;
+import org.apache.tinkerpop.gremlin.driver.remote.DriverRemoteConnection;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.server.channel.HttpChannelizer;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
+import org.apache.tinkerpop.gremlin.util.Tokens;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.tinkerpop.gremlin.process.traversal.AnonymousTraversalSource.traversal;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.StringContains.containsString;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Integration tests for HTTP remote transactions using the driver API ({@code Cluster.transact()}).
+ *
+ * Tests exercise {@link RemoteTransaction} directly via its {@code submit()} methods.
+ *
+ * Server-side verification tests (raw HTTP) are in {@link GremlinServerHttpTransactionIntegrateTest}.
+ */
+public class GremlinDriverTransactionIntegrateTest extends AbstractGremlinServerIntegrationTest {
+ private static final String GTX = "gtx";
+ private static final int MAX_GET_WAIT = 5000;
+
+ private Cluster cluster;
+
+ @Before
+ public void openCluster() {
+ cluster = TestClientFactory.open();
+ }
+
+ @After
+ public void closeCluster() throws Exception {
+ if (cluster != null) cluster.close();
+ }
+
+ @Override
+ public Settings overrideSettings(final Settings settings) {
+ settings.channelizer = HttpChannelizer.class.getName();
+ final String nameOfTest = name.getMethodName();
+ switch (nameOfTest) {
+ case "shouldEnforceMaxConcurrentTransactions":
+ settings.maxConcurrentTransactions = 1;
+ break;
+ case "shouldTimeoutIdleTransaction":
+ case "shouldTimeoutIdleTransactionWithNoOperations":
+ case "shouldRejectLateCommitAfterTimeout":
+ settings.transactionTimeout = 1000;
+ break;
+ case "shouldTimeoutOnlyIdleTransactionNotActiveOne":
+ settings.transactionTimeout = 2000;
+ break;
+ }
+ return settings;
+ }
+
+ /**
+ * Begin a transaction, add a vertex, verify isolation, commit, verify isOpen transitions, and verify data persists
+ * after commit.
+ */
+ @Test
+ public void shouldCommitTransaction() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+ // #4: isOpen true after begin
+ assertTrue(tx.isOpen());
+
+ tx.submit("g.addV('person').property('name','alice')");
+
+ // #6: uncommitted data not visible outside the transaction
+ assertEquals(0L, client.submit("g.V().hasLabel('person').count()").one().getLong());
+
+ tx.commit();
+ // #4: isOpen false after commit
+ assertFalse(tx.isOpen());
+ // #1, #7: committed data visible to non-transactional reads
+ assertEquals(1L, client.submit("g.V().hasLabel('person').count()").one().getLong());
+
+ client.close();
+ }
+
+ @Test
+ public void shouldRollbackTransaction() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+ assertTrue(tx.isOpen());
+
+ tx.submit("g.addV('person').property('name','bob')");
+
+ tx.rollback();
+ // #5: isOpen false after rollback
+ assertFalse(tx.isOpen());
+ // #2: data discarded after rollback
+ assertEquals(0L, client.submit("g.V().hasLabel('person').count()").one().getLong());
+
+ client.close();
+ }
+
+ @Test
+ public void shouldSupportIntraTransactionConsistency() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+ tx.submit("g.addV('test').property('name','A')");
+ // #8: read-your-own-writes — vertex A visible within the transaction
+ assertEquals(1L, tx.submit("g.V().hasLabel('test').count()").all().get().get(0).getLong());
+
+ tx.submit("g.addV('test').property('name','B')");
+ tx.submit("g.V().has('name','A').addE('knows').to(V().has('name','B'))");
+
+ // verify the full subgraph is visible within the transaction before commit
+ assertEquals(2L, tx.submit("g.V().hasLabel('test').count()").all().get().get(0).getLong());
+ assertEquals(1L, tx.submit("g.V().outE('knows').count()").all().get().get(0).getLong());
+
+ tx.commit();
+
+ // #3: vertices and edges persist after commit
+ assertEquals(2L, client.submit("g.V().hasLabel('test').count()").all().get().get(0).getLong());
+ assertEquals(1L, client.submit("g.E().hasLabel('knows').count()").all().get().get(0).getLong());
+ }
+
+ @Test
+ public void shouldThrowOnSubmitAfterCommit() throws Exception {
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+
+ tx.submit("g.addV()");
+ tx.commit();
+
+ try {
+ tx.submit("g.V().count()");
+ fail("Expected exception on submit after commit");
+ } catch (IllegalStateException ex) {
+ assertThat(ex.getMessage(), containsString("Transaction is not open"));
+ }
+ }
+
+ @Test
+ public void shouldThrowOnSubmitAfterRollback() throws Exception {
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+
+ tx.submit("g.addV()");
+ tx.rollback();
+
+ try {
+ tx.submit("g.V().count()");
+ fail("Expected exception on submit after rollback");
+ } catch (IllegalStateException ex) {
+ assertThat(ex.getMessage(), containsString("Transaction is not open"));
+ }
+ }
+
+ @Test
+ public void shouldThrowOnDoubleBegin() throws Exception {
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+
+ try {
+ tx.begin();
+ fail("Expected IllegalStateException on second begin()");
+ } catch (IllegalStateException ex) {
+ assertThat(ex.getMessage(), containsString("Transaction already started"));
+ }
+ }
+
+ @Test
+ public void shouldThrowOnCommitWhenNotOpen() throws Exception {
+ final RemoteTransaction tx = cluster.transact(GTX);
+ assertFalse(tx.isOpen());
+
+ try {
+ tx.commit();
+ fail("Expected IllegalStateException on commit when not open");
+ } catch (IllegalStateException ex) {
+ assertThat(ex.getMessage(), containsString("Transaction is not open"));
+ }
+ }
+
+ @Test
+ public void shouldThrowOnRollbackWhenNotOpen() throws Exception {
+ final RemoteTransaction tx = cluster.transact(GTX);
+ assertFalse(tx.isOpen());
+
+ try {
+ tx.rollback();
+ fail("Expected IllegalStateException on rollback when not open");
+ } catch (IllegalStateException ex) {
+ assertThat(ex.getMessage(), containsString("Transaction is not open"));
+ }
+ }
+
+ @Test
+ public void shouldReturnNullTransactionIdBeforeBegin() throws Exception {
+ final RemoteTransaction tx = cluster.transact(GTX);
+
+ // before begin, transactionId should be null
+ assertNull(tx.getTransactionId());
+
+ tx.begin();
+ // after begin, transactionId should be non-null and non-blank
+ assertNotNull(tx.getTransactionId());
+ assertFalse(tx.getTransactionId().isBlank());
+ }
+
+ @Test
+ public void shouldCommitOnCloseByDefault() throws Exception {
+ final RemoteTransaction tx1 = cluster.transact(GTX);
+ tx1.begin();
+ tx1.submit("g.addV('person').property('name','close_test')");
+ // close() should trigger default COMMIT behavior
+ tx1.close();
+ assertFalse(tx1.isOpen());
+
+ final RemoteTransaction tx2 = cluster.transact(GTX);
+ tx2.begin();
+ assertEquals(1L, tx2.submit("g.V().hasLabel('person').count()").one().getLong());
+ }
+
+ @Test
+ public void shouldRollbackOnCloseWhenConfigured() throws Exception {
+ final RemoteTransaction tx1 = cluster.transact(GTX);
+ tx1.onClose(Transaction.CLOSE_BEHAVIOR.ROLLBACK);
+ tx1.begin();
+ tx1.submit("g.addV('person').property('name','rollback_close_test')");
+ tx1.close();
+ assertFalse(tx1.isOpen());
+
+ final RemoteTransaction tx2 = cluster.transact(GTX);
+ tx2.begin();
+ assertEquals(0L, tx2.submit("g.V().hasLabel('person').count()").one().getLong());
+ }
+
+ @Test
+ public void shouldRollbackOpenTransactionsOnClusterClose() throws Exception {
+ final RemoteTransaction tx1 = cluster.transact(GTX);
+ tx1.begin();
+ tx1.submit("g.addV('cluster-close')");
+ tx1.submit("g.addV('cluster-close')");
+
+ final RemoteTransaction tx2 = cluster.transact(GTX);
+ tx2.begin();
+ tx2.submit("g.addV('cluster-close')");
+
+ // close cluster without committing — should rollback open transactions
+ cluster.close();
+ cluster = null;
+
+ // reconnect and verify data was not persisted
+ final Cluster cluster2 = TestClientFactory.open();
+ try {
+ final RemoteTransaction cluster2tx1 = cluster2.transact(GTX);
+ cluster2tx1.begin();
+ assertEquals(0L, cluster2tx1.submit("g.V().hasLabel('person').count()").all().get().get(0).getLong());
+ } finally {
+ cluster2.close();
+ }
+ }
+
+ @Test
+ public void shouldEnforceMaxConcurrentTransactions() throws Exception {
+ // first transaction fills the single slot
+ final RemoteTransaction tx1 = cluster.transact(GTX);
+ tx1.begin();
+
+ // second transaction should fail
+ try {
+ final RemoteTransaction tx2 = cluster.transact(GTX);
+ tx2.begin();
+ fail("Expected exception when max concurrent transactions exceeded");
+ } catch (Exception ex) {
+ final Throwable root = ExceptionHelper.getRootCause(ex);
+ assertThat(root.getMessage(), containsString("Maximum concurrent transactions exceeded"));
+ }
+ }
+
+ @Test
+ public void shouldTimeoutIdleTransaction() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+ tx.submit("g.addV('timeout_test')");
+
+ // wait for the transaction to timeout
+ Thread.sleep(2000);
+
+ // the transaction was rolled back server-side; attempting to commit should fail
+ try {
+ tx.commit();
+ fail("Expected exception on commit after server-side timeout");
+ } catch (Exception ex) {
+ final Throwable root = ExceptionHelper.getRootCause(ex);
+ assertThat(root.getMessage(), containsString("Transaction not found"));
+ }
+
+ // verify data was not persisted
+ assertEquals(0L, client.submit("g.V().hasLabel('timeout_test').count()").all().get().get(0).getLong());
+ }
+
+ @Test
+ public void shouldTimeoutIdleTransactionWithNoOperations() throws Exception {
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+
+ // wait for the transaction to timeout
+ Thread.sleep(2000);
+
+ // attempting to commit should fail because the server rolled back
+ try {
+ tx.commit();
+ fail("Expected exception on commit after server-side timeout");
+ } catch (Exception ex) {
+ final Throwable root = ExceptionHelper.getRootCause(ex);
+ assertThat(root.getMessage(), containsString("Transaction not found"));
+ }
+ }
+
+ @Test
+ public void shouldTimeoutOnlyIdleTransactionNotActiveOne() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+
+ final RemoteTransaction txActive = cluster.transact(GTX);
+ txActive.begin();
+ txActive.submit("g.addV('active')");
+
+ final RemoteTransaction txIdle = cluster.transact(GTX);
+ txIdle.begin();
+ txIdle.submit("g.addV('idle')");
+
+ // keep the active transaction alive by sending requests at intervals shorter than timeout
+ for (int i = 0; i < 3; i++) {
+ Thread.sleep(800);
+ txActive.submit("g.V().count()");
+ }
+
+ // by now the idle transaction should have timed out (2000ms elapsed)
+ // the active transaction should still be alive
+ txActive.commit();
+
+ // verify active transaction's data persisted
+ assertEquals(1L, client.submit("g.V().hasLabel('active').count()").all().get().get(0).getLong());
+
+ // idle transaction should have been rolled back by timeout
+ try {
+ txIdle.commit();
+ fail("Expected exception on commit of timed-out idle transaction");
+ } catch (Exception ex) {
+ final Throwable root = ExceptionHelper.getRootCause(ex);
+ assertThat(root.getMessage(), containsString("Transaction not found"));
+ }
+
+ // verify idle transaction's data was not persisted
+ assertEquals(0L, client.submit("g.V().hasLabel('idle').count()").all().get().get(0).getLong());
+ }
+
+ @Test
+ public void shouldRejectLateCommitAfterTimeout() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+ tx.submit("g.addV('person').property('name','late_commit')");
+
+ // wait for timeout
+ Thread.sleep(2000);
+
+ // attempt commit — should fail because server already rolled back
+ try {
+ tx.commit();
+ fail("Expected exception on late commit after timeout");
+ } catch (Exception ex) {
+ final Throwable root = ExceptionHelper.getRootCause(ex);
+ assertThat(root.getMessage(), containsString("Transaction not found"));
+ }
+
+ // verify data was not persisted
+ assertEquals(0L, client.submit("g.V().hasLabel('person').count()").all().get().get(0).getLong());
+ }
+
+ @Test
+ public void shouldIsolateConcurrentTransactions() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+
+ final RemoteTransaction tx1 = cluster.transact(GTX);
+ tx1.begin();
+ final RemoteTransaction tx2 = cluster.transact(GTX);
+ tx2.begin();
+
+ tx1.submit("g.addV('tx1')");
+ tx2.submit("g.addV('tx2')");
+
+ // tx1 should not see tx2's data and vice versa
+ assertEquals(0L, tx1.submit("g.V().hasLabel('tx2').count()").all().get().get(0).getLong());
+ assertEquals(0L, tx2.submit("g.V().hasLabel('tx1').count()").all().get().get(0).getLong());
+
+ tx1.commit();
+ tx2.commit();
+
+ // both should be visible after commit
+ assertEquals(1L, client.submit("g.V().hasLabel('tx1').count()").all().get().get(0).getLong());
+ assertEquals(1L, client.submit("g.V().hasLabel('tx2').count()").all().get().get(0).getLong());
+ }
+
+ @Test
+ public void shouldOpenAndCloseManyTransactionsSequentially() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+ final long numberOfTransactions = 50;
+
+ for (int i = 0; i < numberOfTransactions; i++) {
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+ tx.submit("g.addV('stress')");
+ tx.commit();
+ }
+
+ final long count = client.submit("g.V().hasLabel('stress').count()").all().get().get(0).getLong();
+ assertEquals(numberOfTransactions, count);
+
+ Thread.sleep(100);
+ // this should be 0, but to prevent flakiness, make it a reasonable number less than numberOfTransactions
+ assertTrue(server.getServerGremlinExecutor().getTransactionManager().getActiveTransactionCount() < 35);
+ }
+
+ @Test
+ public void shouldIsolateTransactionalAndNonTransactionalRequests() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+ tx.submit("g.addV('tx_data')");
+
+ // non-transactional read should not see uncommitted tx data
+ assertEquals(0L, client.submit("g.V().hasLabel('tx_data').count()").all().get().get(0).getLong());
+
+ tx.commit();
+
+ // now the data should be visible
+ assertEquals(1L, client.submit("g.V().hasLabel('tx_data').count()").all().get().get(0).getLong());
+ }
+
+ @Test
+ public void shouldRejectBeginOnNonTransactionalGraph() throws Exception {
+ final RemoteTransaction tx = cluster.transact("gclassic");
+ try {
+ tx.begin();
+ fail("Expected exception when beginning transaction on non-transactional graph");
+ } catch (Exception ex) {
+ final Throwable root = ExceptionHelper.getRootCause(ex);
+ assertThat(root.getMessage(), containsString("Graph does not support transactions"));
+ }
+ }
+
+ @Test
+ public void shouldTargetCorrectGraph() throws Exception {
+ final Client client = cluster.connect();
+
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+
+ tx.submit("g.addV('routed')");
+ tx.commit();
+
+ // vertex should exist in the transactional graph (gtx)
+ assertEquals(1L, client.submit("g.V().hasLabel('routed').count()",
+ RequestOptions.build().addG(GTX).create()).all().get().get(0).getLong());
+
+ // vertex should NOT exist in the classic graph (gclassic)
+ assertEquals(0L, client.submit("g.V().hasLabel('routed').count()",
+ RequestOptions.build().addG("gclassic").create()).all().get().get(0).getLong());
+ }
+
+ @Test
+ public void shouldAutoCommitNonTransactionalWrite() throws Exception {
+ final Client client = cluster.connect();
+ client.submit("g.addV('auto')").all().get();
+ assertEquals(1L, client.submit("g.V().hasLabel('auto').count()").one().getLong());
+ }
+
+ @Test
+ public void shouldUseProvidedRequestOptions() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+ final RequestOptions ro = RequestOptions.build().
+ addParameter("x", "vowels").
+ materializeProperties(Tokens.MATERIALIZE_PROPERTIES_TOKENS).
+ create();
+ final Vertex v = tx.submit("g.addV(x).property('a', 'b')", ro).
+ all().
+ get(MAX_GET_WAIT, TimeUnit.MILLISECONDS).
+ get(0).
+ getVertex();
+ assertFalse(v.properties().hasNext());
+ tx.commit();
+
+ final List<Result> results =
+ client.submit("g.V().hasLabel('vowels')").all().get(MAX_GET_WAIT, TimeUnit.MILLISECONDS);
+ assertEquals(1L, results.size());
+ }
+
+ @Test
+ public void shouldUseProvidedParameters() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+ final Map params = new HashMap();
+ params.put("x", "consonants");
+ tx.submit("g.addV(x)", params);
+ tx.commit();
+
+ final List<Result> results =
+ client.submit("g.V().hasLabel('consonants')").all().get(MAX_GET_WAIT, TimeUnit.MILLISECONDS);
+ assertEquals(1L, results.size());
+ }
+
+ @Test
+ public void shouldCleanUpOnBeginFailure() throws Exception {
+ final RemoteTransaction tx = cluster.transact("gclassic");
+
+ try {
+ tx.begin();
+ fail("Expected exception on begin for non-transactional graph");
+ } catch (RuntimeException ex) {
+ assertThat(ex.getMessage(), containsString("Failed to begin transaction"));
+ }
+
+ // verify cleanup: transaction is closed and has no ID
+ assertFalse(tx.isOpen());
+ assertNull(tx.getTransactionId());
+
+ // second begin should fail — state moved to CLOSED, not back to NOT_STARTED
+ try {
+ tx.begin();
+ fail("Expected IllegalStateException on begin after failed begin");
+ } catch (IllegalStateException ex) {
+ assertThat(ex.getMessage(), containsString("Transaction already started"));
+ }
+ }
+
+ @Test
+ public void shouldKeepTransactionOpenAfterTraversalError() throws Exception {
+ final Client client = cluster.connect().alias(GTX);
+
+ final RemoteTransaction tx = cluster.transact(GTX);
+ tx.begin();
+ tx.submit("g.addV('good_vertex')");
+
+ // submit a bad traversal that should fail
+ try {
+ tx.submit("g.V().fail()");
+ } catch (Exception ex) {
+ // expected error from bad traversal
+ }
+
+ // transaction should still be open — rollback should work
+ assertTrue(tx.isOpen());
+ tx.rollback();
+
+ assertFalse(tx.isOpen());
+ assertEquals(0L, client.submit("g.V().hasLabel('good_vertex').count()").all().get().get(0).getLong());
+ }
+
+ @Test
+ public void shouldWorkWithDriverRemoteConnection() throws Exception {
+ final GraphTraversalSource g = traversal().with(DriverRemoteConnection.using(cluster, GTX));
+ final GraphTraversalSource gtx = g.tx().begin();
+ gtx.addV("val").iterate();
+ gtx.tx().commit();
+
+ assertEquals(1L, g.V().hasLabel("val").count().next().longValue());
+ }
+}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/HttpDriverIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/HttpDriverIntegrateTest.java
index 17563cdd8b..f6452abd09 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/HttpDriverIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/HttpDriverIntegrateTest.java
@@ -209,22 +209,6 @@ public class HttpDriverIntegrateTest extends AbstractGremlinServerIntegrationTes
}
}
- @Test
- public void shouldFailToUseSession() {
- final Cluster cluster = TestClientFactory.build().create();
- try {
- final Client client = cluster.connect("shouldFailToUseSession");
- client.submit("g.inject(2)").all().get();
- fail("Can't use session with HTTP");
- } catch (Exception ex) {
- final Throwable t = ExceptionUtils.getRootCause(ex);
- // assertEquals("Cannot use sessions or tx() with HttpChannelizer", t.getMessage());
- assertEquals("not implemented", t.getMessage());
- } finally {
- cluster.close();
- }
- }
-
@Test
public void shouldDeserializeErrorWithGraphBinary() {
final Cluster cluster = TestClientFactory.build().create();