You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2016/02/11 18:13:00 UTC
incubator-tinkerpop git commit: Allowed for transaction management in
sessions.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1039 [created] e9966b3bd
Allowed for transaction management in sessions.
Added a per request argument called manageTransaction to allow an in-session client to signal that it would like Gremlin Server to commit/rollback at the end of the request. This basically allows a session to behave much like the sessionless request does by default.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/e9966b3b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/e9966b3b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/e9966b3b
Branch: refs/heads/TINKERPOP-1039
Commit: e9966b3bdbbf8ca34a45e6f9f7a48e9b8b2d5720
Parents: 4a6f9a2
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Thu Feb 11 12:10:37 2016 -0500
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Thu Feb 11 12:10:37 2016 -0500
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../src/reference/gremlin-applications.asciidoc | 103 +++++++++++++------
.../upgrade/release-3.1.x-incubating.asciidoc | 36 +++++++
.../apache/tinkerpop/gremlin/driver/Client.java | 5 +-
.../tinkerpop/gremlin/driver/Cluster.java | 24 ++++-
.../apache/tinkerpop/gremlin/driver/Tokens.java | 11 +-
.../server/op/AbstractEvalOpProcessor.java | 30 +++---
.../server/GremlinDriverIntegrateTest.java | 57 ++++++++--
.../server/GremlinServerIntegrateTest.java | 62 +++++++++++
9 files changed, 273 insertions(+), 56 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e9966b3b/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index caccb0f..de5bc67 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -30,6 +30,7 @@ TinkerPop 3.1.2 (NOT OFFICIALLY RELEASED YET)
* Deprecated `ScriptElementFactory` and made the local `StarGraph` globally available for `ScriptInputFormat`'s `parse()` method.
* Optimized memory-usage in `TraversalVertexProgram`.
* `Graph` instances are not merely "closed" at the end of tests, they are "cleared" via `GraphProvider.clear()`, which should in turn cleans up old data for an implementation.
+* Expanded the Gremlin Server protocol to allow for transaction management on in-session requests and updated the `gremlin-driver` to take advantage of that.
* Greatly reduced the amount of objects required in OLAP for the `ReducingBarrierStep` steps.
[[release-3.1.1-incubating]]
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e9966b3b/docs/src/reference/gremlin-applications.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/reference/gremlin-applications.asciidoc b/docs/src/reference/gremlin-applications.asciidoc
index 097312e..43608b1 100644
--- a/docs/src/reference/gremlin-applications.asciidoc
+++ b/docs/src/reference/gremlin-applications.asciidoc
@@ -1178,7 +1178,7 @@ The parameter is called `#jsr223.groovy.engine.keep.globals` and has four option
By specifying an option other than `hard`, an `OutOfMemoryError` in Gremlin Server should be avoided. Of course,
this approach will come with the downside that compiled scripts could be garbage collected and thus removed from the
-cache, forcing Gremlin Server to recompile later.
+cache, forcing Gremlin Server to recompile later if that script is later encountered.
[[sessions]]
Considering Sessions
@@ -1206,6 +1206,57 @@ boundaries are managed properly from one request to the next.
server that the session was initialized in. Gremlin Server does not share session state as the transactional context
of a `Graph` is bound to the thread it was initialized in.
+To connect to a session with Java via the `gremlin-driver`, it is necessary to create a `SessionedClient` from the
+`Cluster` object:
+
+[source,java]
+----
+Cluster cluster = Cluster.open(); <1>
+Client client = cluster.connect("sessionName"); <2>
+----
+
+<1> Opens a reference to `localhost` as <<connecting-via-java,previously shown>>.
+<2> Creates a `SessionedClient` given the configuration options of the Cluster. The `connect()` method is given a
+`String` value that becomes the unique name of the session. It is often best to simply use a `UUID` to represent
+the session.
+
+It is also possible to have Gremlin Server manage the transactions as is done with sessionless requests. The user is
+in control of enabling this feature when creating the `SessionedClient`:
+
+[source,java]
+----
+Cluster cluster = Cluster.open();
+Client client = cluster.connect("sessionName", true);
+----
+
+Specifying `true` to the `connect()` method signifies that the `client` should make each request as one encapsulated
+in a transaction. With this configuration of `client` there is no need to close a transaction manually.
+
+When using this mode of the `SessionedClient` it is important to recognize that global variable state for the session
+is not rolled-back on failure depending on where the failure occurs. For example, sending the following script would
+create a variable "x" in global session scope that would be acccessible on the next request:
+
+[source,groovy]
+x = 1
+
+However, sending this script which explicitly throws an exception:
+
+[source,groovy]
+y = 2
+throw new RuntimeException()
+
+will result in an obvious failure during script evaluation and "y" will not be available to the next request. The
+complication arises where the script evaluates successfully, but fails during result iteration or serialization. For
+example, this script:
+
+[source,groovy]
+a = 1
+g.addV()
+
+would sucessfully evaluate and return a `Traversal`. The variable "a" would be available on the next request. However,
+if there was a failure in transaction management on the call to `commit()`, "a" would still be available to the next
+request.
+
A session is a "heavier" approach to the simple "request/response" approach of sessionless requests, but is sometimes
necessary for a given use case.
@@ -1213,8 +1264,9 @@ necessary for a given use case.
Considering Transactions
^^^^^^^^^^^^^^^^^^^^^^^^
-Gremlin Server performs automated transaction handling for "sessionless" requests (i.e. no state between requests). It
-will automatically commit or rollback transactions depending on the success or failure of the request.
+Gremlin Server performs automated transaction handling for "sessionless" requests (i.e. no state between requests) and
+for "in-session" requests with that feature enabled. It will automatically commit or rollback transactions depending
+on the success or failure of the request.
Another aspect of Transaction Management that should be considered is the usage of the `strictTransactionManagement`
setting. It is `false` by default, but when set to `true`, it forces the user to pass `aliases` for all requests.
@@ -1336,7 +1388,7 @@ Gremlin Server will send:
|=========================================================
|Code |Name |Description
|200 |SUCCESS |The server successfully processed a request to completion - there are no messages remaining in this stream.
-|204 |NO CONTENT |The server processed the request but there is no result to return (e.g. an `Iterator` with no elements).
+|204 |NO CONTENT |The server processed the request but there is no result to return (e.g. an `Iterator` with no elements) - there are no messages remaining in this stream.
|206 |PARTIAL CONTENT |The server successfully returned some content, but there is more in the stream to arrive - wait for a `SUCCESS` to signify the end of the stream.
|401 |UNAUTHORIZED |The request attempted to access resources that the requesting user did not have access to.
|407 |AUTHENTICATE |A challenge from the server for the client to authenticate its request.
@@ -1348,16 +1400,6 @@ Gremlin Server will send:
|599 |SERVER SERIALIZATION ERROR |The server was not capable of serializing an object that was returned from the script supplied on the request. Either transform the object into something Gremlin Server can process within the script or install mapper serialization classes to Gremlin Server.
|=========================================================
-`SUCCESS` and `NO CONTENT` messages are terminating messages that indicate that a request was properly handled on the
-server and that there are no additional messages streaming in for that request. When developing a driver, it is
-important to note the slight differences in semantics for these result codes when it comes to sessionless versus
-in-session requests. For a sessionless request, which operates under automatic transaction management, Gremlin Server
-will only send one of these message types after result iteration and transaction `commit()`. In other words, the
-driver could potentially expect to receive a number of "successful" `PARTIAL CONTENT` messages before ultimately
-ending in failure on `commit()`. For in-session requests, the client is responsible for managing the transaction
-and therefore, a first request could receive multiple "success" related messages, only to fail on a future request
-that finally issues the `commit()`.
-
OpProcessors Arguments
^^^^^^^^^^^^^^^^^^^^^^
@@ -1386,12 +1428,12 @@ evaluated and is committed when the script completes (or rolled back if an error
[width="100%",cols="3,10a",options="header"]
|=========================================================
|Key |Description
-|processor |As this is the default `OpProcessor` this value can be set to an empty string
+|processor |As this is the default `OpProcessor` this value can be set to an empty string.
|op |[width="100%",cols="3,10",options="header"]
!=========================================================
!Key !Description
!`authentication` !A request that contains the response to a server challenge for authentication.
-!`eval` !Evaluate a Gremlin script provided as a `String`
+!`eval` !Evaluate a Gremlin script provided as a `String`.
!=========================================================
|=========================================================
@@ -1406,9 +1448,9 @@ evaluated and is committed when the script completes (or rolled back if an error
[width="100%",cols="2,2,9",options="header"]
|=========================================================
|Key |Type |Description
-|gremlin |String | *Required* The Gremlin script to evaluate
-|bindings |Map |A map of key/value pairs to apply as variables in the context of the Gremlin script
-|language |String |The flavor used (e.g. `gremlin-groovy`)
+|gremlin |String | *Required* The Gremlin script to evaluate.
+|bindings |Map |A map of key/value pairs to apply as variables in the context of the Gremlin script.
+|language |String |The flavor of Gremlin used (e.g. `gremlin-groovy`).
|aliases |Map |A map of key/value pairs that allow globally bound `Graph` and `TraversalSource` objects to
be aliased to different variable names for purposes of the current request. The value represents the name the
global variable and its key represents the new binding name as it will be referenced in the Gremlin query. For
@@ -1420,11 +1462,11 @@ Session OpProcessor
+++++++++++++++++++
The "session" `OpProcessor` handles requests for the primary function of Gremlin Server - executing Gremlin. It is
-like the "standard" `OpProcessor`, but instead maintains state between sessions and leaves all transaction management
-up to the calling client. It is important that clients that open sessions, commit or roll them back, however Gremlin
-Server will try to clean up such things when a session is killed that has been abandoned. It is important to consider
-that a session can only be maintained with a single machine. In the event that multiple Gremlin Server are deployed,
-session state is not shared among them.
+like the "standard" `OpProcessor`, but instead maintains state between sessions and allows the option to leave all
+transaction management up to the calling client. It is important that clients that open sessions, commit or roll
+them back, however Gremlin Server will try to clean up such things when a session is killed that has been abandoned.
+It is important to consider that a session can only be maintained with a single machine. In the event that multiple
+Gremlin Server are deployed, session state is not shared among them.
[width="100%",cols="3,10a",options="header"]
|=========================================================
@@ -1434,8 +1476,8 @@ session state is not shared among them.
[cols="3,10",options="header"]
!=========================================================
!Key !Description
-!`authentication` !A request that contains the response to a server challenge for authentication
-!`eval` !Evaluate a Gremlin script provided as a `String`
+!`authentication` !A request that contains the response to a server challenge for authentication.
+!`eval` !Evaluate a Gremlin script provided as a `String`.
!`close` !Close the specified session and rollback any open transactions.
|=========================================================
@@ -1450,10 +1492,11 @@ session state is not shared among them.
[width="100%",options="header"]
|=========================================================
|Key |Type |Description
-|gremlin |String | *Required* The Gremlin script to evaluate
-|session |String | *Required* The session identifier for the current session - typically this value should be a UUID (the session will be created if it doesn't exist)
-|bindings |Map |A map of key/value pairs to apply as variables in the context of the Gremlin script
-|language |String |The flavor used (e.g. `gremlin-groovy`)
+|gremlin |String | *Required* The Gremlin script to evaluate.
+|session |String | *Required* The session identifier for the current session - typically this value should be a UUID (the session will be created if it doesn't exist).
+|manageTransaction |Boolean |When set to `true` the transaction for the current request is auto-committed or rolled-back as are done with sessionless requests - defaulted to `false`.
+|bindings |Map |A map of key/value pairs to apply as variables in the context of the Gremlin script.
+|language |String |The flavor of Gremlin used (e.g. `gremlin-groovy`)
|=========================================================
'`close` operation arguments'
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e9966b3b/docs/src/upgrade/release-3.1.x-incubating.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/upgrade/release-3.1.x-incubating.asciidoc b/docs/src/upgrade/release-3.1.x-incubating.asciidoc
index 840d20c..1b7425c 100644
--- a/docs/src/upgrade/release-3.1.x-incubating.asciidoc
+++ b/docs/src/upgrade/release-3.1.x-incubating.asciidoc
@@ -29,6 +29,27 @@ TinkerPop 3.1.2
Please see the link:https://github.com/apache/incubator-tinkerpop/blob/3.1.1-incubating/CHANGELOG.asciidoc#tinkerpop-312-release-date-XXXXXXXXXXXXXXXXXXXXXXXXXX[changelog] for a complete list of all the modifications that are part of this release.
+Upgrading for Users
+~~~~~~~~~~~~~~~~~~~
+
+Session Transaction Management
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+When connecting to a session with `gremlin-driver`, it is now possible to configure the `Client` instance so as to
+request that the server manage the transaction for each requests.
+
+[source,java]
+----
+Cluster cluster = Cluster.open();
+Client client = cluster.connect("sessionName", true);
+----
+
+Specifying `true` to the `connect()` method signifies that the `client` should make each request as one encapsulated
+in a transaction. With this configuration of `client` there is no need to close a transaction manually.
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-1039[TINKERPOP-1039],
+link:http://tinkerpop.apache.org/docs/3.1.2-incubating/reference/#sessions[Reference Documentation - Considering Sessions]
+
Upgrading for Providers
~~~~~~~~~~~~~~~~~~~~~~~
@@ -48,6 +69,21 @@ automated method for clearing graphs at the end of a test and some tests call `c
See: link:https://issues.apache.org/jira/browse/TINKERPOP-1146[TINKERPOP-1146]
+Driver Providers
+^^^^^^^^^^^^^^^^
+
+Session Transaction Management
+++++++++++++++++++++++++++++++
+
+Up until now transaction management has been a feature of sessionless requests only, but the new `manageTransaction`
+request argument for the link:http://tinkerpop.apache.org/docs/3.1.2-incubating/reference/#_session_opprocessor[Session OpProcessor]
+changes that. Session-based requests can now pass this boolean value on each request to signal to
+Gremlin Server that it should attempt to commit (or rollback) the transaction at the end of the request. By default,
+this value as `false`, so there is no change to the protocol for this feature.
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-1039[TINKERPOP-1039],
+link:http://tinkerpop.apache.org/docs/3.1.2-incubating/reference/#sessions[Reference Documentation - Considering Sessions]
+
TinkerPop 3.1.1
---------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e9966b3b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
----------------------------------------------------------------------
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 8c80b8a..6a91f0e 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
@@ -490,12 +490,14 @@ public abstract class Client {
*/
public final static class SessionedClient extends Client {
private final String sessionId;
+ private final boolean manageTransactions;
private ConnectionPool connectionPool;
- SessionedClient(final Cluster cluster, final String sessionId) {
+ SessionedClient(final Cluster cluster, final String sessionId, final boolean manageTransactions) {
super(cluster);
this.sessionId = sessionId;
+ this.manageTransactions = manageTransactions;
}
String getSessionId() {
@@ -531,6 +533,7 @@ public abstract class Client {
public RequestMessage buildMessage(final RequestMessage.Builder builder) {
builder.processor("session");
builder.addArg(Tokens.ARGS_SESSION, sessionId);
+ builder.addArg(Tokens.ARGS_MANAGE_TRANSACTION, manageTransactions);
return builder.create();
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e9966b3b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
----------------------------------------------------------------------
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 c8d3bd6..b29cc95 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
@@ -86,7 +86,7 @@ public final class Cluster {
* Creates a {@link Client.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 rolledback manually.
+ * 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
@@ -96,9 +96,29 @@ public final class Cluster {
* @param sessionId user supplied id for the session which should be unique (a UUID is ideal).
*/
public <T extends Client> T connect(final String sessionId) {
+ return connect(sessionId, false);
+ }
+
+ /**
+ * Creates a {@link Client.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
+ */
+ public <T extends Client> T connect(final String sessionId, final boolean manageTransactions) {
if (null == sessionId || sessionId.isEmpty())
throw new IllegalArgumentException("sessionId cannot be null or empty");
- return (T) new Client.SessionedClient(this, sessionId);
+ return (T) new Client.SessionedClient(this, sessionId, manageTransactions);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e9966b3b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Tokens.java
----------------------------------------------------------------------
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Tokens.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Tokens.java
index 64d2fe1..dbb37c6 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Tokens.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Tokens.java
@@ -49,20 +49,27 @@ public final class Tokens {
public static final String ARGS_LANGUAGE = "language";
/**
- * @deprecated As of release 3.1.0, replaced by {@link #ARGS_ALIASES}.
+ * @deprecated As of release 3.1.0-incubating, replaced by {@link #ARGS_ALIASES}.
*/
@Deprecated
public static final String ARGS_REBINDINGS = "rebindings";
public static final String ARGS_SESSION = "session";
+ public static final String ARGS_MANAGE_TRANSACTION = "manageTransaction";
public static final String ARGS_SASL = "sasl";
public static final String ARGS_COORDINATES_GROUP = "group";
public static final String ARGS_COORDINATES_ARTIFACT = "artifact";
public static final String ARGS_COORDINATES_VERSION = "version";
+ public static final String ARGS_INFO_TYPE_DEPENDENCIES = "dependencies";
+
+ /**
+ * @deprecated As of release 3.1.1-incubating, replaced by {@link #ARGS_INFO_TYPE_DEPENDENCIES}
+ */
+ @Deprecated
public static final String ARGS_INFO_TYPE_DEPDENENCIES = "dependencies";
public static final String ARGS_INFO_TYPE_IMPORTS = "imports";
- public static final List<String> INFO_TYPES = Arrays.asList(ARGS_INFO_TYPE_DEPDENENCIES,
+ public static final List<String> INFO_TYPES = Arrays.asList(ARGS_INFO_TYPE_DEPENDENCIES,
ARGS_INFO_TYPE_IMPORTS);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e9966b3b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
index aab9950..e6cf2e6 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
@@ -57,7 +57,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
import java.util.regex.Pattern;
-import java.util.stream.Collectors;
import static com.codahale.metrics.MetricRegistry.name;
@@ -190,6 +189,10 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
final String language = args.containsKey(Tokens.ARGS_LANGUAGE) ? (String) args.get(Tokens.ARGS_LANGUAGE) : null;
final Bindings bindings = bindingsSupplier.get();
+ // sessionless requests are always transaction managed, but in-session requests are configurable.
+ final boolean managedTransactionsForRequest = manageTransactions ?
+ true : (Boolean) args.getOrDefault(Tokens.ARGS_MANAGE_TRANSACTION, false);
+
final CompletableFuture<Object> evalFuture = gremlinExecutor.eval(script, language, bindings, null, o -> {
final Iterator itty = IteratorUtils.asIterator(o);
@@ -201,11 +204,11 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
final String errorMessage = String.format("Response iteration exceeded the configured threshold for request [%s] - %s", msg, ex.getMessage());
logger.warn(errorMessage);
ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT).statusMessage(errorMessage).create());
- if (manageTransactions) attemptRollback(msg, context.getGraphManager(), settings.strictTransactionManagement);
+ if (managedTransactionsForRequest) attemptRollback(msg, context.getGraphManager(), settings.strictTransactionManagement);
} catch (Exception ex) {
logger.warn(String.format("Exception processing a script on request [%s].", msg), ex);
ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR).statusMessage(ex.getMessage()).create());
- if (manageTransactions) attemptRollback(msg, context.getGraphManager(), settings.strictTransactionManagement);
+ if (managedTransactionsForRequest) attemptRollback(msg, context.getGraphManager(), settings.strictTransactionManagement);
}
});
@@ -245,11 +248,15 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
final boolean useBinary = ctx.channel().attr(StateKey.USE_BINARY).get();
boolean warnOnce = false;
+ // sessionless requests are always transaction managed, but in-session requests are configurable.
+ final boolean managedTransactionsForRequest = manageTransactions ?
+ true : (Boolean) msg.getArgs().getOrDefault(Tokens.ARGS_MANAGE_TRANSACTION, false);
+
// we have an empty iterator - happens on stuff like: g.V().iterate()
if (!itty.hasNext()) {
// as there is nothing left to iterate if we are transaction managed then we should execute a
// commit here before we send back a NO_CONTENT which implies success
- if (manageTransactions) attemptCommit(msg, context.getGraphManager(), settings.strictTransactionManagement);
+ if (managedTransactionsForRequest) attemptCommit(msg, context.getGraphManager(), settings.strictTransactionManagement);
ctx.writeAndFlush(ResponseMessage.build(msg)
.code(ResponseStatusCode.NO_CONTENT)
.create());
@@ -260,9 +267,6 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
final StopWatch stopWatch = new StopWatch();
stopWatch.start();
- // if we manage the transactions then we need to commit stuff now that eval is complete.
- Iterator toIterate = itty;
-
// the batch size can be overridden by the request
final int resultIterationBatchSize = (Integer) msg.optionalArgs(Tokens.ARGS_BATCH_SIZE)
.orElse(settings.resultIterationBatchSize);
@@ -271,7 +275,7 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
// use an external control to manage the loop as opposed to just checking hasNext() in the while. this
// prevent situations where auto transactions create a new transaction after calls to commit() withing
// the loop on calls to hasNext().
- boolean hasMore = toIterate.hasNext();
+ boolean hasMore = itty.hasNext();
while (hasMore) {
if (Thread.interrupted()) throw new InterruptedException();
@@ -280,13 +284,13 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
// so iterating next() if the message is not written and flushed would bump the aggregate size beyond
// the expected resultIterationBatchSize. Total serialization time for the response remains in
// effect so if the client is "slow" it may simply timeout.
- if (aggregate.size() < resultIterationBatchSize) aggregate.add(toIterate.next());
+ if (aggregate.size() < resultIterationBatchSize) aggregate.add(itty.next());
// send back a page of results if batch size is met or if it's the end of the results being iterated.
// also check writeability of the channel to prevent OOME for slow clients.
if (ctx.channel().isWritable()) {
- if (aggregate.size() == resultIterationBatchSize || !toIterate.hasNext()) {
- final ResponseStatusCode code = toIterate.hasNext() ? ResponseStatusCode.PARTIAL_CONTENT : ResponseStatusCode.SUCCESS;
+ if (aggregate.size() == resultIterationBatchSize || !itty.hasNext()) {
+ final ResponseStatusCode code = itty.hasNext() ? ResponseStatusCode.PARTIAL_CONTENT : ResponseStatusCode.SUCCESS;
// serialize here because in sessionless requests the serialization must occur in the same
// thread as the eval. as eval occurs in the GremlinExecutor there's no way to get back to the
@@ -294,7 +298,7 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
serializeResponseMessage(ctx, msg, serializer, useBinary, aggregate, code);
// only need to reset the aggregation list if there's more stuff to write
- if (toIterate.hasNext())
+ if (itty.hasNext())
aggregate = new ArrayList<>(resultIterationBatchSize);
else {
// iteration and serialization are both complete which means this finished successfully. note that
@@ -302,7 +306,7 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
// local errors will get rolledback below because the exceptions aren't thrown in those cases to be
// caught by the GremlinExecutor for global rollback logic. this only needs to be committed if
// there are no more items to iterate and serialization is complete
- if (manageTransactions) attemptCommit(msg, context.getGraphManager(), settings.strictTransactionManagement);
+ if (managedTransactionsForRequest) attemptCommit(msg, context.getGraphManager(), settings.strictTransactionManagement);
// exit the result iteration loop as there are no more results left. using this external control
// because of the above commit. some graphs may open a new transaction on the call to
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e9966b3b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
----------------------------------------------------------------------
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 2dc0935..125afa9 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
@@ -109,6 +109,7 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
case "shouldExecuteSessionlessScriptOnTransactionalGraph":
case "shouldExecuteScriptInSessionOnTransactionalWithManualTransactionsGraph":
case "shouldExecuteInSessionAndSessionlessWithoutOpeningTransaction":
+ case "shouldManageTransactionsInSession":
deleteDirectory(new File("/tmp/neo4j"));
settings.graphs.put("graph", "conf/neo4j-empty.properties");
break;
@@ -729,11 +730,11 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
client.submit("v.property(\"color\",\"blue\")").all().get();
client.submit("graph.tx().commit()").all().get();
-
+
// Run a sessionless request to change transaction.readWriteConsumer back to AUTO
// The will make the next in session request fail if consumers aren't ThreadLocal
sessionlessClient.submit("graph.vertices().next()").all().get();
-
+
client.submit("graph.tx().open()").all().get();
final Vertex vertexAfterTx = client.submit("graph.vertices().next()").all().get().get(0).getVertex();
@@ -748,28 +749,28 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
@Test
public void shouldExecuteInSessionAndSessionlessWithoutOpeningTransaction() throws Exception {
assumeNeo4jIsPresent();
-
+
final Cluster cluster = Cluster.build().create();
final Client sessionClient = cluster.connect(name.getMethodName());
final Client sessionlessClient = cluster.connect();
-
+
//open transaction in session, then add vertex and commit
sessionClient.submit("graph.tx().open()").all().get();
final Vertex vertexBeforeTx = sessionClient.submit("v=graph.addVertex(\"name\",\"stephen\")").all().get().get(0).getVertex();
assertEquals("stephen", vertexBeforeTx.values("name").next());
sessionClient.submit("graph.tx().commit()").all().get();
-
+
// check that session transaction is closed
final boolean isOpen = sessionClient.submit("graph.tx().isOpen()").all().get().get(0).getBoolean();
assertTrue("Transaction should be closed", !isOpen);
-
+
//run a sessionless read
sessionlessClient.submit("graph.traversal().V()").all().get();
-
+
// check that session transaction is still closed
final boolean isOpenAfterSessionless = sessionClient.submit("graph.tx().isOpen()").all().get().get(0).getBoolean();
assertTrue("Transaction should stil be closed", !isOpenAfterSessionless);
-
+
}
@Test
@@ -1019,4 +1020,44 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
cluster.close();
}
+
+ @Test
+ public void shouldManageTransactionsInSession() throws Exception {
+ assumeNeo4jIsPresent();
+
+ final Cluster cluster = Cluster.build().create();
+ final Client client = cluster.connect();
+ final Client sessionWithManagedTx = cluster.connect(name.getMethodName() + "-managed", true);
+ final Client sessionWithoutManagedTx = cluster.connect(name.getMethodName() + "-not-managed");
+
+ // this should auto-commit
+ final Vertex vStephen = sessionWithManagedTx.submit("v = g.addV('name','stephen').next()").all().get().get(0).getVertex();
+ assertEquals("stephen", vStephen.value("name"));
+
+ // the other clients should see that change because of auto-commit
+ assertThat(client.submit("g.V().has('name','stephen').hasNext()").all().get().get(0).getBoolean(), is(true));
+ assertThat(sessionWithoutManagedTx.submit("g.V().has('name','stephen').hasNext()").all().get().get(0).getBoolean(), is(true));
+
+ // this should NOT auto-commit
+ final Vertex vDaniel = sessionWithoutManagedTx.submit("v = g.addV('name','daniel').next()").all().get().get(0).getVertex();
+ assertEquals("daniel", vDaniel.value("name"));
+
+ // the other clients should NOT see that change because of auto-commit
+ assertThat(client.submit("g.V().has('name','daniel').hasNext()").all().get().get(0).getBoolean(), is(false));
+ assertThat(sessionWithManagedTx.submit("g.V().has('name','daniel').hasNext()").all().get().get(0).getBoolean(), is(false));
+
+ // but "v" should still be there
+ final Vertex vDanielAgain = sessionWithoutManagedTx.submit("v").all().get().get(0).getVertex();
+ assertEquals("daniel", vDanielAgain.value("name"));
+
+ // now commit manually
+ sessionWithoutManagedTx.submit("g.tx().commit()").all().get();
+
+ // should be there for all now
+ assertThat(client.submit("g.V().has('name','daniel').hasNext()").all().get().get(0).getBoolean(), is(true));
+ assertThat(sessionWithManagedTx.submit("g.V().has('name','daniel').hasNext()").all().get().get(0).getBoolean(), is(true));
+ assertThat(sessionWithoutManagedTx.submit("g.V().has('name','daniel').hasNext()").all().get().get(0).getBoolean(), is(true));
+
+ cluster.close();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e9966b3b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
index 1d3d650..96c2727 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
@@ -131,6 +131,7 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
settings.processors.add(processorSettings);
break;
case "shouldExecuteInSessionAndSessionlessWithoutOpeningTransactionWithSingleClient":
+ case "shouldExecuteInSessionWithTransactionManagement":
deleteDirectory(new File("/tmp/neo4j"));
settings.graphs.put("graph", "conf/neo4j-empty.properties");
break;
@@ -661,6 +662,67 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
@Test
@SuppressWarnings("unchecked")
+ public void shouldExecuteInSessionWithTransactionManagement() throws Exception {
+ assumeNeo4jIsPresent();
+
+ try (final SimpleClient client = new WebSocketClient()) {
+ final RequestMessage addRequest = RequestMessage.build(Tokens.OPS_EVAL)
+ .processor("session")
+ .addArg(Tokens.ARGS_SESSION, name.getMethodName())
+ .addArg(Tokens.ARGS_GREMLIN, "v=graph.addVertex(\"name\",\"stephen\")")
+ .addArg(Tokens.ARGS_MANAGE_TRANSACTION, true)
+ .create();
+ final List<ResponseMessage> addResponses = client.submit(addRequest);
+ assertEquals(1, addResponses.size());
+ assertEquals(ResponseStatusCode.SUCCESS, addResponses.get(0).getStatus().getCode());
+
+ // Check to see if the transaction is closed.
+ final RequestMessage checkRequest = RequestMessage.build(Tokens.OPS_EVAL)
+ .processor("session")
+ .addArg(Tokens.ARGS_SESSION, name.getMethodName())
+ .addArg(Tokens.ARGS_GREMLIN, "graph.tx().isOpen()")
+ .create();
+ final List<ResponseMessage> checkResponses = client.submit(checkRequest);
+ assertEquals(1, checkResponses.size());
+ assertEquals(ResponseStatusCode.SUCCESS, checkResponses.get(0).getStatus().getCode());
+ assertThat(((List<Boolean>) checkResponses.get(0).getResult().getData()).get(0), is(false));
+
+ // lets run a sessionless read and validate that the transaction was managed
+ final RequestMessage sessionlessRequest = RequestMessage.build(Tokens.OPS_EVAL)
+ .addArg(Tokens.ARGS_GREMLIN, "graph.traversal().V().values('name')")
+ .create();
+ final List<ResponseMessage> sessionlessResponses = client.submit(sessionlessRequest);
+ assertEquals(1, sessionlessResponses.size());
+ assertEquals(ResponseStatusCode.SUCCESS, sessionlessResponses.get(0).getStatus().getCode());
+ assertEquals("stephen", ((List<String>) sessionlessResponses.get(0).getResult().getData()).get(0));
+
+ // make sure the session is intact
+ final RequestMessage getRequest = RequestMessage.build(Tokens.OPS_EVAL)
+ .processor("session")
+ .addArg(Tokens.ARGS_SESSION, name.getMethodName())
+ .addArg(Tokens.ARGS_GREMLIN, "v.values(\"name\")")
+ .addArg(Tokens.ARGS_MANAGE_TRANSACTION, true)
+ .create();
+ final List<ResponseMessage> getResponses = client.submit(getRequest);
+ assertEquals(1, getResponses.size());
+ assertEquals(ResponseStatusCode.SUCCESS, getResponses.get(0).getStatus().getCode());
+ assertEquals("stephen", ((List<String>) getResponses.get(0).getResult().getData()).get(0));
+
+ // Check to see if the transaction is still closed.
+ final RequestMessage checkAgainRequest = RequestMessage.build(Tokens.OPS_EVAL)
+ .processor("session")
+ .addArg(Tokens.ARGS_SESSION, name.getMethodName())
+ .addArg(Tokens.ARGS_GREMLIN, "graph.tx().isOpen()")
+ .create();
+ final List<ResponseMessage> checkAgainstResponses = client.submit(checkAgainRequest);
+ assertEquals(1, checkAgainstResponses.size());
+ assertEquals(ResponseStatusCode.SUCCESS, checkAgainstResponses.get(0).getStatus().getCode());
+ assertThat(((List<Boolean>) checkAgainstResponses.get(0).getResult().getData()).get(0), is(false));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
public void shouldStillSupportDeprecatedRebindingsParameterOnServer() throws Exception {
// this test can be removed when the rebindings arg is removed
try (SimpleClient client = new WebSocketClient()) {