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/10/31 17:25:47 UTC
[31/31] tinkerpop git commit: TINKERPOP-932 Added "force" option on
session close.
TINKERPOP-932 Added "force" option on session close.
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/5802f164
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/5802f164
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/5802f164
Branch: refs/heads/TINKERPOP-932
Commit: 5802f16484516151e8e3dd68ad16e7b2079c9fee
Parents: 1eac35b
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Mon Oct 24 16:06:38 2016 -0400
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Mon Oct 31 13:24:56 2016 -0400
----------------------------------------------------------------------
CHANGELOG.asciidoc | 2 +
docs/src/dev/provider/index.asciidoc | 5 ++
.../upgrade/release-3.2.x-incubating.asciidoc | 20 +++++
.../apache/tinkerpop/gremlin/driver/Client.java | 24 ++++++
.../tinkerpop/gremlin/driver/Connection.java | 7 +-
.../apache/tinkerpop/gremlin/driver/Tokens.java | 1 +
.../tinkerpop/gremlin/server/Settings.java | 1 -
.../gremlin/server/op/session/Session.java | 83 ++++++++++++++------
.../server/op/session/SessionOpProcessor.java | 21 ++++-
.../GremlinServerSessionIntegrateTest.java | 59 ++++++++++++--
10 files changed, 189 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/5802f164/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index cf8c007..ba445b4 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,8 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
TinkerPop 3.2.4 (Release Date: NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+* Added a `force` option for killing sessions without waiting for transaction close or timeout of a currently running job or multiple jobs.
+* Deprecated `Session.kill()` and `Session.manualKill()`.
* `SparkGraphComputer` no longer starts a worker iteration if the worker's partition is empty.
* Added `ProjectStep.getProjectKeys()` for strategies that rely on such information.
* Added `VertexFeatures.supportsDuplicateMultiProperties()` for graphs that only support unique values in multi-properties.
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/5802f164/docs/src/dev/provider/index.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/dev/provider/index.asciidoc b/docs/src/dev/provider/index.asciidoc
index 00ac4a9..c2f9f20 100644
--- a/docs/src/dev/provider/index.asciidoc
+++ b/docs/src/dev/provider/index.asciidoc
@@ -901,6 +901,11 @@ to send an alias pair with key of "g" and value of "g2" and thus allow the scrip
|=========================================================
|Key |Type |Description
|session |String | *Required* The session identifier for the session to close.
+|force |Boolean | Determines if the session should be force closed when the client is closed. Force closing will not
+attempt to close open transactions from existing running jobs and leave it to the underlying graph to decided how to
+proceed with those orphaned transactions. Setting this to `true` tends to lead to faster close operation and release
+of resources which can be desirable if Gremlin Server has a long session timeout and a long script evaluation timeout
+as attempts to close long run jobs can occur more rapidly. If not provided, this value is `false`.
|=========================================================
Traversal OpProcessor
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/5802f164/docs/src/upgrade/release-3.2.x-incubating.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/upgrade/release-3.2.x-incubating.asciidoc b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
index 30daae1..aac616b 100644
--- a/docs/src/upgrade/release-3.2.x-incubating.asciidoc
+++ b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
@@ -54,6 +54,26 @@ following fields were deprecated:
* `OptIn.SUITE_PROCESS_PERFORMANCE`
* `OptIn.SUITE_STRUCTURE_PERFORMANCE`
+Drivers Providers
+^^^^^^^^^^^^^^^^^
+
+Force Close
++++++++++++
+
+Closing a session will first attempt a proper close of any open transactions. A problem can occur, however, if there is
+a long run job (e.g. an OLAP-based traversal) executing, as that job will block the calls to close the transactions.
+By exercising the option to a do a "forced close" the session will skip trying to close the transactions and just
+attempt to interrupt the long run job. By not closing transactions, the session leaves it up to the underlying graph
+database to sort out how it will deal with those orphaned transactions. On the positive side though (for those graphs
+which do that well) , long run jobs have the opporunity to be cancelled without waiting for a timeout of the job itself
+which will allow resources to be released earlier.
+
+The "force" argument is passed on the "close" message and is a boolean value. This is an optional argument to "close"
+and defaults to `false`.
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-932[TINKERPOP-932],
+link:http://tinkerpop.apache.org/docs/current/dev/provider/#_session_opprocessor[Provider Documentation - Session OpProcessor]
+
TinkerPop 3.2.3
---------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/5802f164/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 bd397a1..2c448dc 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
@@ -804,10 +804,12 @@ public abstract class Client {
public static class SessionSettings {
private final boolean manageTransactions;
private final String sessionId;
+ private final boolean forceClosed;
private SessionSettings(final Builder builder) {
manageTransactions = builder.manageTransactions;
sessionId = builder.sessionId;
+ forceClosed = builder.forceClosed;
}
/**
@@ -824,6 +826,14 @@ public abstract class Client {
return sessionId;
}
+ /**
+ * Determines if the session will be force closed. See {@link Builder#forceClosed(boolean)} for more details
+ * on what that means.
+ */
+ public boolean isForceClosed() {
+ return forceClosed;
+ }
+
public static SessionSettings.Builder build() {
return new SessionSettings.Builder();
}
@@ -831,6 +841,7 @@ public abstract class Client {
public static class Builder {
private boolean manageTransactions = false;
private String sessionId = UUID.randomUUID().toString();
+ private boolean forceClosed = false;
private Builder() {}
@@ -854,6 +865,19 @@ public abstract class Client {
return this;
}
+ /**
+ * Determines if the session should be force closed when the client is closed. Force closing will not
+ * attempt to close open transactions from existing running jobs and leave it to the underlying graph to
+ * decided how to proceed with those orphaned transactions. Setting this to {@code true} tends to lead to
+ * faster close operation which can be desirable if Gremlin Server has a long session timeout and a long
+ * script evaluation timeout as attempts to close long run jobs can occur more rapidly. By default, this
+ * value is {@link false}.
+ */
+ public Builder forceClosed(final boolean forced) {
+ this.forceClosed = forced;
+ return this;
+ }
+
public SessionSettings create() {
return new SessionSettings(this);
}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/5802f164/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
----------------------------------------------------------------------
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 2a68032..571d906 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
@@ -306,9 +306,12 @@ final class Connection {
// be called once. once shutdown is initiated, it shouldn't be executed a second time or else it sends more
// messages at the server and leads to ugly log messages over there.
if (shutdownInitiated.compareAndSet(false, true)) {
+ // maybe this should be delegated back to the Client implementation??? kinda weird to instanceof here.....
if (client instanceof Client.SessionedClient) {
- // maybe this should be delegated back to the Client implementation???
- final RequestMessage closeMessage = client.buildMessage(RequestMessage.build(Tokens.OPS_CLOSE)).create();
+ final boolean forceClose = client.getSettings().getSession().get().isForceClosed();
+ final RequestMessage closeMessage = client.buildMessage(
+ RequestMessage.build(Tokens.OPS_CLOSE).addArg(Tokens.ARGS_FORCE, forceClose)).create();
+
final CompletableFuture<ResultSet> closed = new CompletableFuture<>();
write(closeMessage, closed);
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/5802f164/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 25c79af..fb577d7 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
@@ -70,6 +70,7 @@ public final class Tokens {
public static final String ARGS_BATCH_SIZE = "batchSize";
public static final String ARGS_BINDINGS = "bindings";
public static final String ARGS_ALIASES = "aliases";
+ public static final String ARGS_FORCE = "force";
public static final String ARGS_GREMLIN = "gremlin";
public static final String ARGS_LANGUAGE = "language";
public static final String ARGS_SCRIPT_EVAL_TIMEOUT = "scriptEvaluationTimeout";
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/5802f164/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java
index 8b9fe89..4bb2089 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java
@@ -20,7 +20,6 @@ package org.apache.tinkerpop.gremlin.server;
import io.netty.handler.ssl.SslContext;
import org.apache.tinkerpop.gremlin.driver.MessageSerializer;
-import org.apache.tinkerpop.gremlin.driver.ser.GryoMessageSerializerV1d0;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
import org.apache.tinkerpop.gremlin.server.auth.AllowAllAuthenticator;
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/5802f164/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
index c9bc7c1..8f4da66 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
@@ -52,6 +52,7 @@ public class Session {
private final String session;
private final ScheduledExecutorService scheduledExecutorService;
private final long configuredSessionTimeout;
+ private final long configuredPerGraphCloseTimeout;
private AtomicBoolean killing = new AtomicBoolean(false);
private AtomicReference<ScheduledFuture> kill = new AtomicReference<>();
@@ -85,7 +86,10 @@ public class Session {
final Settings.ProcessorSettings processorSettings = this.settings.processors.stream()
.filter(p -> p.className.equals(SessionOpProcessor.class.getCanonicalName()))
.findAny().orElse(SessionOpProcessor.DEFAULT_SETTINGS);
- this.configuredSessionTimeout = Long.parseLong(processorSettings.config.get(SessionOpProcessor.CONFIG_SESSION_TIMEOUT).toString());
+ this.configuredSessionTimeout = Long.parseLong(processorSettings.config.getOrDefault(
+ SessionOpProcessor.CONFIG_SESSION_TIMEOUT, SessionOpProcessor.DEFAULT_SESSION_TIMEOUT).toString());
+ this.configuredPerGraphCloseTimeout = Long.parseLong(processorSettings.config.getOrDefault(
+ SessionOpProcessor.CONFIG_PER_GRAPH_CLOSE_TIMEOUT, SessionOpProcessor.DEFAULT_PER_GRAPH_CLOSE_TIMEOUT).toString());
this.gremlinExecutor = initializeGremlinExecutor().create();
}
@@ -129,17 +133,43 @@ public class Session {
/**
* Stops the session with call to {@link #kill()} but also stops the session expiration call which ensures that
- * the session is only killed once.
+ * the session is only killed once. Calls {@link #manualKill(boolean)} with {@code false}.
+ *
+ * @deprecated As of release 3.2.4, replaced by {@link #manualKill(boolean)}.
*/
+ @Deprecated
public void manualKill() {
+ manualKill(false);
+ }
+
+ /**
+ * Stops the session with call to {@link #kill()} but also stops the session expiration call which ensures that
+ * the session is only killed once. See {@link #kill(boolean)} for information on how what "forcing" the session
+ * kill will mean.
+ */
+ public void manualKill(final boolean force) {
kill.get().cancel(true);
- kill();
+ kill(force);
}
/**
- * Kills the session and rollback any uncommitted changes on transactional graphs.
+ * Kills the session and rollback any uncommitted changes on transactional graphs. Same as calling
+ * {@link #kill(boolean)} with {@code false}.
+ *
+ * @deprecated As of release 3.2.4, replaced by {@link #kill(boolean)}.
*/
+ @Deprecated
public synchronized void kill() {
+ kill(false);
+ }
+
+ /**
+ * Kills the session and rollback any uncommitted changes on transactional graphs. When "force" closed, the
+ * session won't bother to try to submit transaction close commands. It will be up to the underlying graph
+ * implementation to determine how it will clean up orphaned transactions. The force will try to cancel scheduled
+ * jobs and interrupt any currently running ones. Interruption is not guaranteed, but an attempt will be made.
+ */
+ public synchronized void kill(final boolean force) {
killing.set(true);
// if the session has already been removed then there's no need to do this process again. it's possible that
@@ -147,26 +177,33 @@ public class Session {
// kill() from being called more than once
if (!sessions.containsKey(session)) return;
- // when the session is killed open transaction should be rolled back
- graphManager.getGraphs().entrySet().forEach(kv -> {
- final Graph g = kv.getValue();
- if (g.features().graph().supportsTransactions()) {
- // have to execute the rollback in the executor because the transaction is associated with
- // that thread of execution from this session
- try {
- executor.submit(() -> {
- if (g.tx().isOpen()) {
- logger.info("Rolling back open transactions on {} before killing session: {}", kv.getKey(), session);
- g.tx().rollback();
- }
- }).get(30000, TimeUnit.MILLISECONDS);
- } catch (Exception ex) {
- logger.warn("An error occurred while attempting rollback when closing session: " + session, ex);
+ if (!force) {
+ // when the session is killed open transaction should be rolled back
+ graphManager.getGraphs().entrySet().forEach(kv -> {
+ final Graph g = kv.getValue();
+ if (g.features().graph().supportsTransactions()) {
+ // have to execute the rollback in the executor because the transaction is associated with
+ // that thread of execution from this session
+ try {
+ executor.submit(() -> {
+ if (g.tx().isOpen()) {
+ logger.info("Rolling back open transactions on {} before killing session: {}", kv.getKey(), session);
+ g.tx().rollback();
+ }
+ }).get(configuredPerGraphCloseTimeout, TimeUnit.MILLISECONDS);
+ } catch (Exception ex) {
+ logger.warn(String.format("An error occurred while attempting rollback on %s when closing session: %s", kv.getKey(), session), ex);
+ }
}
- }
- });
-
- // prevent any additional requests from processing now that the mass rollback has been completed
+ });
+ } else {
+ logger.info("Skipped attempt to close open graph transactions on {} - close was forced", session);
+ }
+
+ // prevent any additional requests from processing. if the kill was not "forced" then jobs were scheduled to
+ // try to rollback open transactions. those jobs either timed-out or completed successfully. either way, no
+ // additional jobs will be allowed, running jobs will be cancelled (if possible) and any scheduled jobs will
+ // be cancelled
executor.shutdownNow();
sessions.remove(session);
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/5802f164/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
index bec0c55..651bdb0 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
@@ -66,14 +66,26 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
}
/**
- * Configuration setting for how long a session will be available before it timesout.
+ * Configuration setting for how long a session will be available before it times out.
*/
public static final String CONFIG_SESSION_TIMEOUT = "sessionTimeout";
/**
+ * Configuration setting for how long to wait in milliseconds for each configured graph to close any open
+ * transactions when the session is killed.
+ */
+ public static final String CONFIG_PER_GRAPH_CLOSE_TIMEOUT = "perGraphCloseTimeout";
+
+ /**
* Default timeout for a session is eight hours.
*/
- public static final long DEFAULT_SESSION_TIMEOUT = 28800000l;
+ public static final long DEFAULT_SESSION_TIMEOUT = 28800000;
+
+ /**
+ * Default amount of time to wait in milliseconds for each configured graph to close any open transactions when
+ * the session is killed.
+ */
+ public static final long DEFAULT_PER_GRAPH_CLOSE_TIMEOUT = 10000;
static final Settings.ProcessorSettings DEFAULT_SETTINGS = new Settings.ProcessorSettings();
@@ -81,6 +93,7 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
DEFAULT_SETTINGS.className = SessionOpProcessor.class.getCanonicalName();
DEFAULT_SETTINGS.config = new HashMap<String, Object>() {{
put(CONFIG_SESSION_TIMEOUT, DEFAULT_SESSION_TIMEOUT);
+ put(CONFIG_PER_GRAPH_CLOSE_TIMEOUT, DEFAULT_PER_GRAPH_CLOSE_TIMEOUT);
}};
}
@@ -106,6 +119,8 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
throw new OpProcessorException(msg, ResponseMessage.build(requestMessage).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
}
+ final boolean force = requestMessage.<Boolean>optionalArgs(Tokens.ARGS_FORCE).orElse(false);
+
return Optional.of(ctx -> {
// validate the session is present and then remove it if it is.
final Session sessionToClose = sessions.get(requestMessage.getArgs().get(Tokens.ARGS_SESSION).toString());
@@ -114,7 +129,7 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
throw new OpProcessorException(msg, ResponseMessage.build(requestMessage).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
}
- sessionToClose.manualKill();
+ sessionToClose.manualKill(force);
});
} else {
return Optional.empty();
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/5802f164/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSessionIntegrateTest.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSessionIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSessionIntegrateTest.java
index 3c1fef9..c384742 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSessionIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSessionIntegrateTest.java
@@ -49,6 +49,7 @@ import java.util.stream.IntStream;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.core.IsCollectionContaining.hasItem;
import static org.hamcrest.core.StringStartsWith.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
@@ -84,47 +85,95 @@ public class GremlinServerSessionIntegrateTest extends AbstractGremlinServerInt
final String nameOfTest = name.getMethodName();
switch (nameOfTest) {
case "shouldHaveTheSessionTimeout":
- case "shouldCloseSessionOnceOnRequest":
settings.processors.clear();
final Settings.ProcessorSettings processorSettings = new Settings.ProcessorSettings();
processorSettings.className = SessionOpProcessor.class.getCanonicalName();
processorSettings.config = new HashMap<>();
processorSettings.config.put(SessionOpProcessor.CONFIG_SESSION_TIMEOUT, 3000L);
settings.processors.add(processorSettings);
-
+ break;
+ case "shouldBlockAdditionalRequestsDuringForceClose":
+ case "shouldCloseSessionOnceOnRequest":
+ clearNeo4j(settings);
Logger.getRootLogger().setLevel(Level.INFO);
break;
case "shouldEnsureSessionBindingsAreThreadSafe":
settings.threadPoolWorker = 2;
break;
+ case "shouldBlockAdditionalRequestsDuringClose":
case "shouldExecuteInSessionAndSessionlessWithoutOpeningTransactionWithSingleClient":
case "shouldExecuteInSessionWithTransactionManagement":
case "shouldRollbackOnEvalExceptionForManagedTransaction":
- deleteDirectory(new File("/tmp/neo4j"));
- settings.graphs.put("graph", "conf/neo4j-empty.properties");
+ clearNeo4j(settings);
break;
}
return settings;
}
+ private static void clearNeo4j(Settings settings) {
+ deleteDirectory(new File("/tmp/neo4j"));
+ settings.graphs.put("graph", "conf/neo4j-empty.properties");
+ }
+
@Test
public void shouldBlockAdditionalRequestsDuringClose() throws Exception {
+ assumeNeo4jIsPresent();
+
// this is sorta cobbled together a bit given limits/rules about how you can use Cluster/Client instances.
// basically, we need one to submit the long run job and one to do the close operation that will cancel the
// long run job. it is probably possible to do this with some low-level message manipulation but that's
// probably not necessary
final Cluster cluster1 = Cluster.build().create();
final Client client1 = cluster1.connect(name.getMethodName());
- client1.submit("1+1").all().join();
+ client1.submit("graph.addVertex()").all().join();
final Cluster cluster2 = Cluster.build().create();
final Client client2 = cluster2.connect(name.getMethodName());
client2.submit("1+1").all().join();
+ final ResultSet rs = client1.submit("Thread.sleep(3000);1+1");
+
+ // close while the previous request is still executing
+ client2.close();
+
+ assertEquals(2, rs.all().join().get(0).getInt());
+
+ client1.close();
+
+ cluster1.close();
+ cluster2.close();
+
+ // triggered an error during close and since we didn't force close, the attempt to close the transaction
+ // is made
+ assertThat(recordingAppender.getMessages(), hasItem("INFO - Rolling back open transactions on graph before killing session: " + name.getMethodName() + "\n"));
+
+ }
+
+ @Test
+ public void shouldBlockAdditionalRequestsDuringForceClose() throws Exception {
+ assumeNeo4jIsPresent();
+
+ // this is sorta cobbled together a bit given limits/rules about how you can use Cluster/Client instances.
+ // basically, we need one to submit the long run job and one to do the close operation that will cancel the
+ // long run job. it is probably possible to do this with some low-level message manipulation but that's
+ // probably not necessary
+ final Cluster cluster1 = Cluster.build().create();
+ final Client client1 = cluster1.connect(name.getMethodName());
+ client1.submit("graph.addVertex()").all().join();
+ final Cluster cluster2 = Cluster.build().create();
+ final Client.SessionSettings sessionSettings = Client.SessionSettings.build()
+ .sessionId(name.getMethodName())
+ .forceClosed(true).create();
+ final Client client2 = cluster2.connect(Client.Settings.build().useSession(sessionSettings).create());
+ client2.submit("1+1").all().join();
+
final ResultSet rs = client1.submit("Thread.sleep(10000);1+1");
client2.close();
+ // because the close was forced, the message should appear immediately
+ assertThat(recordingAppender.getMessages(), hasItem("INFO - Skipped attempt to close open graph transactions on " + name.getMethodName() + " - close was forced\n"));
+
try {
rs.all().join();
fail("The close of the session on client2 should have interrupted the script sent on client1");