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 2024/03/21 21:44:52 UTC
(tinkerpop) branch remove-sessions-unifiedchannelizer created (now 91a2ded428)
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a change to branch remove-sessions-unifiedchannelizer
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
at 91a2ded428 Remove the UnifiedChannelizer/UnifiedHandler and associated classes.
This branch includes the following new commits:
new 91a2ded428 Remove the UnifiedChannelizer/UnifiedHandler and associated classes.
The 1 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.
(tinkerpop) 01/01: Remove the UnifiedChannelizer/UnifiedHandler and associated classes.
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch remove-sessions-unifiedchannelizer
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 91a2ded428e9beddc559fe8eb5bdaac5d3347c8c
Author: Ken Hu <10...@users.noreply.github.com>
AuthorDate: Wed Mar 20 22:10:57 2024 -0700
Remove the UnifiedChannelizer/UnifiedHandler and associated classes.
---
docs/src/reference/gremlin-applications.asciidoc | 34 +-
.../apache/tinkerpop/gremlin/driver/Client.java | 21 -
.../apache/tinkerpop/gremlin/server/Settings.java | 44 +-
.../gremlin/server/channel/UnifiedChannelizer.java | 86 --
.../gremlin/server/handler/AbstractSession.java | 863 ---------------------
.../gremlin/server/handler/MultiTaskSession.java | 316 --------
.../tinkerpop/gremlin/server/handler/Session.java | 84 --
.../gremlin/server/handler/SessionException.java | 43 -
.../gremlin/server/handler/SessionTask.java | 42 -
.../gremlin/server/handler/SingleTaskSession.java | 99 ---
.../gremlin/server/handler/UnifiedHandler.java | 356 ---------
.../server/op/traversal/TraversalOpProcessor.java | 1 -
.../AbstractGremlinServerIntegrationTest.java | 22 -
.../gremlin/server/GremlinDriverIntegrateTest.java | 2 +-
.../server/GremlinServerHttpIntegrateTest.java | 2 +-
.../gremlin/server/GremlinServerIntegrateTest.java | 97 +--
.../server/GremlinServerSessionIntegrateTest.java | 203 +----
...SessionIntegrateWithUnifiedChannelizerTest.java | 61 --
.../server/GremlinSessionTxIntegrateTest.java | 5 +-
.../channel/UnifiedChannelizerIntegrateTest.java | 57 --
.../server/channel/UnifiedTestChannelizer.java | 48 --
.../server/channel/WsAndHttpTestChannelizer.java | 2 +-
gremlin-server/src/test/resources/logback-test.xml | 2 -
.../org/apache/tinkerpop/gremlin/util/Tokens.java | 7 -
24 files changed, 19 insertions(+), 2478 deletions(-)
diff --git a/docs/src/reference/gremlin-applications.asciidoc b/docs/src/reference/gremlin-applications.asciidoc
index 3674308924..283320182a 100644
--- a/docs/src/reference/gremlin-applications.asciidoc
+++ b/docs/src/reference/gremlin-applications.asciidoc
@@ -847,10 +847,6 @@ channelizer: org.apache.tinkerpop.gremlin.server.channel.HttpChannelizer
[source,yaml]
channelizer: org.apache.tinkerpop.gremlin.server.channel.WsAndHttpChannelizer
-NOTE: The `UnifiedChannelizer` introduced in 3.5.0 can also be used to support HTTP requests as its functionality
-is similar to `WsAndHttpChannelizer`. Please see the Gremlin Server UnifiedChannelizer Section of the Upgrade
-Documentation for 3.5.0 for more link:https://tinkerpop.apache.org/docs/x.y.z/upgrade/#_tinkerpop_3_5_0[details].
-
The `HttpChannelizer` is already configured in the `gremlin-server-rest-modern.yaml` file that is packaged with the Gremlin
Server distribution. To utilize it, start Gremlin Server as follows:
@@ -1010,8 +1006,7 @@ The following table describes the various YAML configuration options that Gremli
|maxContentLength |The maximum length of the aggregated content for a message. Works in concert with `maxChunkSize` where chunked requests are accumulated back into a single message. A request exceeding this size will return a `413 - Request Entity Too Large` status code. A response exceeding this size will raise an internal exception. |65536
|maxHeaderSize |The maximum length of all headers. |8192
|maxInitialLineLength |The maximum length of the initial line (e.g. "GET / HTTP/1.0") processed in a request, which essentially controls the maximum length of the submitted URI. |4096
-|maxParameters |The maximum number of parameters that can be passed on a request. Larger numbers may impact performance for scripts. This configuration only applies to the `UnifiedChannelizer`. |16
-|maxSessionTaskQueueSize |The maximum size that an individual session can queue requests before starting to reject them. This configuration only applies to the `UnifiedChannelizer`. |4096
+|maxParameters |The maximum number of parameters that can be passed on a request. Larger numbers may impact performance for scripts. This configuration only applies to the `HttpChannelizer`. |16
|maxWorkQueueSize |The maximum size the general processing queue can grow before the `gremlinPool` starts to reject requests. |8192
|metrics.consoleReporter.enabled |Turns on console reporting of metrics. |false
|metrics.consoleReporter.interval |Time in milliseconds between reports of metrics to console. |180000
@@ -1045,7 +1040,6 @@ The following table describes the various YAML configuration options that Gremli
|serializers |A `List` of `Map` settings, where each `Map` represents a `MessageSerializer` implementation to use along with its configuration. If this value is not set, then Gremlin Server will configure with GraphSON and GraphBinary but will not register any `ioRegistries` for configured graphs. |_empty_
|serializers[X].className |The full class name of the `MessageSerializer` implementation. |_none_
|serializers[X].config |A `Map` containing `MessageSerializer` specific configurations. |_none_
-|sessionLifetimeTimeout |The maximum time in milliseconds that a session can exist. This value cannot be extended beyond this value irrespective of the number of requests and their individual timeouts. The session life cannot be extended once started. This configuration only applies to the `UnifiedChannelizer`. |600000 (10 minutes)
|ssl.enabled |Determines if SSL is turned on or not. |false
|ssl.keyStore |The private key in JKS or PKCS#12 format. |_none_
|ssl.keyStorePassword |The password of the `keyStore` if it is password-protected. |_none_
@@ -1058,9 +1052,7 @@ The following table describes the various YAML configuration options that Gremli
|strictTransactionManagement |Set to `true` to require `aliases` to be submitted on every requests, where the `aliases` become the scope of transaction management. |false
|threadPoolBoss |The number of threads available to Gremlin Server for accepting connections. Should always be set to `1`. |1
|threadPoolWorker |The number of threads available to Gremlin Server for processing non-blocking reads and writes. |1
-|useCommonEngineForSessions |Ensures that the same `ScriptEngine` is used to support sessions and sessionless requests which will lead to better performance. Do not change this setting from the default without a specific use case in mind. This configuration only applies to the `UnifiedChannelizer`. |true
|useEpollEventLoop |Try to use epoll event loops (works only on Linux os) instead of netty NIO. |false
-|useGlobalFunctionCacheForSessions |Enable the global function cache for sessions when using the `UnifiedChannelizer`. When `true` it means that functions created in one request to a session remain available on the next request to that session. This setting is only relevant when `useGlobalFunctionCacheForSessions` is `false`. |true
|writeBufferHighWaterMark | If the number of bytes in the network send buffer exceeds this value then the channel is no longer writeable, accepting no additional writes until buffer is drained and the `writeBufferLowWaterMark` is met. |65536
|writeBufferLowWaterMark | Once the number of bytes queued in the network send buffer exceeds the `writeBufferHighWaterMark`, the channel will not become writeable again until the buffer is drained and it drops below this value. |65536
|=========================================================
@@ -1070,9 +1062,6 @@ See the <<metrics,Metrics>> section for more information on how to configure Gan
[[opprocessor-configurations]]
==== OpProcessor Configurations
-IMPORTANT: The `UnifiedChannelizer` does not rely on `OpProcessor` infrastructure. If using that channelizer, these
-configuration options can be ignored.
-
An `OpProcessor` provides a way to plug-in handlers to Gremlin Server's processing flow. Gremlin Server uses this
plug-in system itself to expose the packaged functionality that it exposes. Configurations can be supplied to an
`OpProcessor` through the `processors` key in the Gremlin Server configuration file. Each `OpProcessor` can take a
@@ -1273,9 +1262,7 @@ NOTE: Installing Ganglia will include `org.acplt:oncrpc`, which is an LGPL licen
Regardless of the output, the metrics gathered are the same. Each metric is prefixed with
`org.apache.tinkerpop.gremlin.server.GremlinServer` and the following metrics are reported:
-* `sessions` - The number of sessions open at the time the metric was last measured. For the `UnifiedChannelizer`, each
-request creates a "session", even a so-called "sessionless request", which is basically a session that will only
-execute within the context of that single request.
+* `sessions` - The number of sessions open at the time the metric was last measured.
* `errors` - The number of total errors, mean rate, as well as the 1, 5, and 15-minute error rates.
* `op.eval` - The number of script evaluations, mean rate, 1, 5, and 15 minute rates, minimum, maximum, median, mean,
and standard deviation evaluation times, as well as the 75th, 95th, 98th, 99th and 99.9th percentile evaluation times
@@ -1285,10 +1272,9 @@ maximum, median, mean, and standard deviation evaluation times, as well as the 7
percentile evaluation times.
* `engine-name.session.session-id.*` - Metrics related to different `GremlinScriptEngine` instances configured for
session-based requests where "engine-name" will be the actual name of the engine, such as "gremlin-groovy" and
-"session-id" will be the identifier for the session itself. This metric is not measured under the `UnifiedChannelizer`.
+"session-id" will be the identifier for the session itself.
* `engine-name.sessionless.*` - Metrics related to different `GremlinScriptEngine` instances configured for sessionless
-requests where "engine-name" will be the actual name of the engine, such as "gremlin-groovy". This metric is not
-measured under the `UnifiedChannelizer`.
+requests where "engine-name" will be the actual name of the engine, such as "gremlin-groovy".
* `user-agent.*` - Counts the number of connection requests from clients providing a given user agent.
NOTE: Gremlin Server has a limit of 10000 unique user agents to be tracked by metrics. If this cap is exceeded
@@ -2141,12 +2127,6 @@ sessionless requests and construct a single threaded pool for each session creat
tend to optimize sessions to be long-lived. For short-lived sessions, which may be typical when using bytecode based
remote transactions, quickly creating and destroying these sessions can be expensive. It is likely that there will be
increased garbage collection times and frequency as well as a general increase in overall server processing.
-** `UnifiedChannelizer` - The threads of the `gremlinPool` are used to service both sessions and sessionless requests.
-With a common thread pool, this channelizer is a better choice when using lots of short-lived sessions as compared to
-`WebSocketChannelizer` and `WsAndHttpChannelizer`, because there is less cost in starting and stopping sessions. It is
-important though to understand the expected workload for the server and plan the size accordingly to ensure that the
-server does not need to wait for an extended period of time for a thread to be available to process the queue of
-incoming requests.
* Graph element serialization for `Vertex` and `Edge` can be expensive, as their data structures are complex given the
possible existence of multi-properties and meta-properties. When returning data from Gremlin Server only return the
data that is required. For example, if only two properties of a `Vertex` are needed then simply return the two rather
@@ -2480,12 +2460,6 @@ simply executed serially as they arrive to the session. A failed asynchronous re
in the session which may not allow later requests to succeed. Either use synchronous requests only or carefully
consider error conditions with asynchronous requests.
-If using the `UnifiedChannelizer`, failures in evaluation will result in the session being closed and state being
-lost. Asynchronous requests that are queued on the server will be cancelled and additional requests, in-flight or
-otherwise will be rejected. Users should create a new session from the `Cluster` object in this case. The alternative,
-to match the old `OpProcessor` GremlinServer behavior, is to add the `maintainStateAfterException` session setting to
-`true` which will instead have similar behavior to that described in this section.
-
[source,java]
----
Client.SessionSettings settings =
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 d522e2c6cd..303269c5b4 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
@@ -764,7 +764,6 @@ public abstract class Client {
public final static class SessionedClient extends Client {
private final String sessionId;
private final boolean manageTransactions;
- private final boolean maintainStateAfterException;
private ConnectionPool connectionPool;
@@ -774,7 +773,6 @@ public abstract class Client {
super(cluster, settings);
this.sessionId = settings.getSession().get().sessionId;
this.manageTransactions = settings.getSession().get().manageTransactions;
- this.maintainStateAfterException = settings.getSession().get().maintainStateAfterException;
}
/**
@@ -792,7 +790,6 @@ public abstract class Client {
builder.processor("session");
builder.addArg(Tokens.ARGS_SESSION, sessionId);
builder.addArg(Tokens.ARGS_MANAGE_TRANSACTION, manageTransactions);
- builder.addArg(Tokens.ARGS_MAINTAIN_STATE_AFTER_EXCEPTION, maintainStateAfterException);
return builder;
}
@@ -924,13 +921,11 @@ public abstract class Client {
private final boolean manageTransactions;
private final String sessionId;
private final boolean forceClosed;
- private final boolean maintainStateAfterException;
private SessionSettings(final Builder builder) {
manageTransactions = builder.manageTransactions;
sessionId = builder.sessionId;
forceClosed = builder.forceClosed;
- maintainStateAfterException = builder.maintainStateAfterException;
}
/**
@@ -955,10 +950,6 @@ public abstract class Client {
return forceClosed;
}
- public boolean maintainStateAfterException() {
- return maintainStateAfterException;
- }
-
public static SessionSettings.Builder build() {
return new SessionSettings.Builder();
}
@@ -967,22 +958,10 @@ public abstract class Client {
private boolean manageTransactions = false;
private String sessionId = UUID.randomUUID().toString();
private boolean forceClosed = false;
- private boolean maintainStateAfterException = false;
private Builder() {
}
- /**
- * When {@code true} an exception within a session will not close the session and remove the state bound to
- * that session. This setting is for the {@code UnifiedChannelizer} and when set to {@code true} will allow
- * sessions to behave similar to how they did under the {@code OpProcessor} approach original to Gremlin
- * Server. By default this value is {@code false}.
- */
- public Builder maintainStateAfterException(final boolean maintainStateAfterException) {
- this.maintainStateAfterException = maintainStateAfterException;
- return this;
- }
-
/**
* If enabled, transactions will be "managed" such that each request will represent a complete transaction.
* By default this value is {@code false}.
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 7a21afcf0d..1c752535c2 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
@@ -28,10 +28,8 @@ import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
import org.apache.tinkerpop.gremlin.server.auth.AllowAllAuthenticator;
import org.apache.tinkerpop.gremlin.server.auth.Authenticator;
import org.apache.tinkerpop.gremlin.server.authz.Authorizer;
-import org.apache.tinkerpop.gremlin.server.channel.UnifiedChannelizer;
import org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer;
import org.apache.tinkerpop.gremlin.server.handler.AbstractAuthenticationHandler;
-import org.apache.tinkerpop.gremlin.server.handler.Session;
import org.apache.tinkerpop.gremlin.server.util.DefaultGraphManager;
import org.apache.tinkerpop.gremlin.server.util.LifeCycleHook;
import org.apache.tinkerpop.gremlin.structure.Graph;
@@ -205,56 +203,16 @@ public class Settings {
* will be processed by this thread pool. If the threads are exhausted, the requests will queue to the size
* specified by this value after which they will begin to reject the requests.
* <p/>
- * This value should be taken in account with the {@link #maxSessionTaskQueueSize} which is related in some
- * respects. A request that starts a new {@link Session} is handled by this queue, but additional requests to a
- * created {@link Session} will queue separately given that setting per session.
- * <p/>
* By default this value is set to 8192.
*/
public int maxWorkQueueSize = 8192;
- /**
- * Maximum size that an individual {@link Session} can queue requests before starting to reject them. Note that this
- * configuration only applies to the {@link UnifiedChannelizer}. By default this value is set to 4096.
- *
- * @see #maxWorkQueueSize
- */
- public int maxSessionTaskQueueSize = 4096;
-
/**
* Maximum number of parameters that can be passed on a request. Larger numbers may impact performance for scripts.
- * The default is 16 and this setting only applies to the {@link UnifiedChannelizer}.
+ * The default is 16 and this setting only applies to the {@link org.apache.tinkerpop.gremlin.server.channel.HttpChannelizer}.
*/
public int maxParameters = 16;
- /**
- * The time in milliseconds that a {@link UnifiedChannelizer} session can exist. This value cannot be extended
- * beyond this value irrespective of the number of requests and their individual timeouts. Requests must complete
- * within this time frame. If this timeout is reached while there is a running evaluation, there will be an attempt
- * to interrupt it which will result in a timeout error to the client. If there are existing requests enqueued for
- * the session when this timeout is reached, those requests will not be executed and will be closed with server
- * errors. Open transactions will be issued a rollback. The default is 10 minutes.
- */
- public long sessionLifetimeTimeout = 600000;
-
- /**
- * Enable the global function cache for sessions when using the {@link UnifiedChannelizer}. This setting is only
- * relevant when {@link #useGlobalFunctionCacheForSessions} is {@code false}. When {@link true} it means that
- * functions created in one request to a session remain available on the next request to that session.
- */
- public boolean useGlobalFunctionCacheForSessions = true;
-
- /**
- * When {@code true} and using the {@link UnifiedChannelizer} the same engine that will be used to server
- * sessionless requests will also be use to serve sessions. The default value of {@code true} is recommended as
- * it reduces the amount of object creation required for each session and should generally lead to better
- * performance especially if the expectation is that there will be many sessions being created and destroyed
- * rapidly. Setting this value to {@code false} is mostly present to support specific use cases that may require
- * each session having its own engine or to match previous functionality provided by the older channelizers
- * produced prior to 3.5.0.
- */
- public boolean useCommonEngineForSessions = true;
-
/**
* Configured metrics for Gremlin Server.
*/
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/UnifiedChannelizer.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/UnifiedChannelizer.java
deleted file mode 100644
index f3e910e4ab..0000000000
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/UnifiedChannelizer.java
+++ /dev/null
@@ -1,86 +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.server.channel;
-
-import io.netty.channel.ChannelPipeline;
-import org.apache.tinkerpop.gremlin.server.AbstractChannelizer;
-import org.apache.tinkerpop.gremlin.server.Channelizer;
-import org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler;
-import org.apache.tinkerpop.gremlin.server.handler.UnifiedHandler;
-import org.apache.tinkerpop.gremlin.server.handler.WsAndHttpChannelizerHandler;
-import org.apache.tinkerpop.gremlin.server.util.ServerGremlinExecutor;
-
-/**
- * A {@link Channelizer} that supports websocket and HTTP requests and does so with the most streamlined processing
- * model for Gremlin Server introduced with 3.5.0.
- */
-public class UnifiedChannelizer extends AbstractChannelizer {
-
- private WsAndHttpChannelizerHandler wsAndHttpChannelizerHandler;
- private UnifiedHandler unifiedHandler;
- protected static final String PIPELINE_UNIFIED = "unified";
-
- @Override
- public void init(final ServerGremlinExecutor serverGremlinExecutor) {
- super.init(serverGremlinExecutor);
-
- wsAndHttpChannelizerHandler = new WsAndHttpChannelizerHandler();
- wsAndHttpChannelizerHandler.init(serverGremlinExecutor, new HttpGremlinEndpointHandler(serializers, gremlinExecutor, graphManager, settings));
-
- // these handlers don't share any state and can thus be initialized once per pipeline
- unifiedHandler = new UnifiedHandler(settings, graphManager, gremlinExecutor, scheduledExecutorService, this);
- }
-
- @Override
- public void configure(final ChannelPipeline pipeline) {
- wsAndHttpChannelizerHandler.configure(pipeline);
- pipeline.addAfter(PIPELINE_HTTP_REQUEST_DECODER, "WsAndHttpChannelizerHandler", wsAndHttpChannelizerHandler);
- }
-
- @Override
- public void finalize(final ChannelPipeline pipeline) {
- super.finalize(pipeline);
-
- // currently the AbstractChannelizer adds the following handlers which prior to 3.5.0 were essentially
- // required by any Gremlin-processing pipeline you could think of because they provided the functionality
- // of the OpProcessor infrastructure. the OpProcessor infrastructure is on its way to deprecation after
- // TINKERPOP-2245 which introduced this channelizer implementation. since AbstractChannelizer does a nice
- // job of rigging up the rest of the pipeline it seemed to make sense to leave it unchanged to ensure it
- // does not break anyone's channelizers that may depend on it (including TinkerPop's and simply remove
- // those bits here. in the future, when we remove OpProcessor stuff completely we can clean this up.
- pipeline.remove(PIPELINE_OP_SELECTOR);
- pipeline.remove(PIPELINE_OP_EXECUTOR);
-
- pipeline.addLast(PIPELINE_UNIFIED, unifiedHandler);
- }
-
- public UnifiedHandler getUnifiedHandler() {
- return unifiedHandler;
- }
-
- @Override
- public boolean supportsIdleMonitor() {
- return true;
- }
-
- @Override
- public Object createIdleDetectionMessage() {
- return wsAndHttpChannelizerHandler.getWsChannelizer().createIdleDetectionMessage();
- }
-}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/AbstractSession.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/AbstractSession.java
deleted file mode 100644
index 7c04e9eb02..0000000000
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/AbstractSession.java
+++ /dev/null
@@ -1,863 +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.server.handler;
-
-import com.codahale.metrics.Timer;
-import groovy.lang.GroovyRuntimeException;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelHandlerContext;
-import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.apache.tinkerpop.gremlin.util.MessageSerializer;
-import org.apache.tinkerpop.gremlin.util.Tokens;
-import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
-import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
-import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
-import org.apache.tinkerpop.gremlin.util.ser.MessageTextSerializer;
-import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor;
-import org.apache.tinkerpop.gremlin.groovy.jsr223.TimedInterruptTimeoutException;
-import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptEngine;
-import org.apache.tinkerpop.gremlin.jsr223.JavaTranslator;
-import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
-import org.apache.tinkerpop.gremlin.process.traversal.Failure;
-import org.apache.tinkerpop.gremlin.process.traversal.GraphOp;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.VerificationException;
-import org.apache.tinkerpop.gremlin.process.traversal.util.BytecodeHelper;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException;
-import org.apache.tinkerpop.gremlin.server.GraphManager;
-import org.apache.tinkerpop.gremlin.server.GremlinServer;
-import org.apache.tinkerpop.gremlin.server.Settings;
-import org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser;
-import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
-import org.apache.tinkerpop.gremlin.server.util.TraverserIterator;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Transaction;
-import org.apache.tinkerpop.gremlin.structure.util.TemporaryException;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
-import org.codehaus.groovy.control.MultipleCompilationErrorsException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.script.Bindings;
-import javax.script.ScriptException;
-import javax.script.SimpleBindings;
-import java.io.InterruptedIOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Future;
-import java.util.concurrent.FutureTask;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Stream;
-
-import static org.apache.tinkerpop.gremlin.process.traversal.GraphOp.TX_COMMIT;
-import static org.apache.tinkerpop.gremlin.process.traversal.GraphOp.TX_ROLLBACK;
-
-/**
- * A base implementation of {@link Session} which offers some common functionality that matches typical Gremlin Server
- * request response expectations for script, bytecode and graph operations. The class is designed to be extended but
- * take care in understanding the way that different methods are called as they do depend on one another a bit. It
- * maybe best to examine the source code to determine how best to use this class or to extend from the higher order
- * classes of {@link SingleTaskSession} or {@link MultiTaskSession}.
- */
-public abstract class AbstractSession implements Session, AutoCloseable {
- private static final Logger logger = LoggerFactory.getLogger(AbstractSession.class);
- private static final Logger auditLogger = LoggerFactory.getLogger(GremlinServer.AUDIT_LOGGER_NAME);
-
- private final boolean sessionIdOnRequest;
- private final Channel initialChannel;
- private final boolean transactionManaged;
- private final String sessionId;
- private final AtomicReference<ScheduledFuture<?>> sessionCancelFuture = new AtomicReference<>();
- private final AtomicReference<Future<?>> sessionFuture = new AtomicReference<>();
- private long actualTimeoutLengthWhenClosed = 0;
-
- /**
- * The session thread is a reference to the thread that is running the session and should be set by an
- * implementation as the first line of the {@link #run()} method.
- */
- protected Thread sessionThread;
- protected final boolean maintainStateAfterException;
- protected final AtomicReference<CloseReason> closeReason = new AtomicReference<>();
- protected final GraphManager graphManager;
- protected final ConcurrentMap<String, Session> sessions;
- protected final Set<String> aliasesUsedBySession = new HashSet<>();
- protected final AtomicBoolean sessionTaskStarted = new AtomicBoolean(false);
-
- /**
- * The reason that a particular session closed. The reason for the close is generally not important as a
- * final disposition for the {@link Session} instance and is more useful in aiding flow control during the
- * close process.
- */
- protected enum CloseReason {
-
- /**
- * The session exits in a fashion that did not precipitate from some form of interruption, timeout or
- * exception, i.e. it is simply allowed to process to an exit through its normal execution flow. This status
- * may or may not be possible given the context of the implementation. For example, a {@link MultiTaskSession}
- * needs to be interrupted to stop processing.
- */
- EXIT_PROCESSING,
-
- /**
- * The session was interrupted by the channel closing, which can be something initiated by closing the
- * {@link Client} or might be triggered by the server. This may not be considered an error situation and
- * depending on context, might be similar to a {@link #EXIT_PROCESSING} termination.
- */
- CHANNEL_CLOSED,
-
- /**
- * The session encountered an exception related to execution like a script error, traversal iteration problem,
- * serialization issue, etc.
- */
- PROCESSING_EXCEPTION,
-
- /**
- * The session was interrupted by the session lifetime timeout.
- */
- SESSION_TIMEOUT,
-
- /**
- * The session was interrupted by the request timeout.
- */
- REQUEST_TIMEOUT
- }
-
- AbstractSession(final SessionTask sessionTask, final String sessionId,
- final boolean transactionManaged,
- final ConcurrentMap<String, Session> sessions) {
- // this only applies to sessions
- this.maintainStateAfterException = (boolean) sessionTask.getRequestMessage().
- optionalArgs(Tokens.ARGS_MAINTAIN_STATE_AFTER_EXCEPTION).orElse(false);
- this.sessionIdOnRequest = sessionTask.getRequestMessage().optionalArgs(Tokens.ARGS_SESSION).isPresent();
- this.transactionManaged = transactionManaged;
- this.sessionId = sessionId;
- this.initialChannel = sessionTask.getChannelHandlerContext().channel();
-
- // close session if the channel closes to cleanup and close transactions
- this.initialChannel.closeFuture().addListener(f -> {
- if (closeReason.compareAndSet(null, CloseReason.CHANNEL_CLOSED)) {
- close();
- }
- });
- this.sessions = sessions;
- this.graphManager = sessionTask.getGraphManager();
- }
-
- protected synchronized void cancel(final boolean mayInterruptIfRunning) {
- final FutureTask<?> sf = (FutureTask) sessionFuture.get();
- if (sf != null && !sf.isDone()) {
- sf.cancel(mayInterruptIfRunning);
-
- if (!sessionTaskStarted.get()) {
- sendTimeoutResponseForUncommencedTask();
- }
- }
- }
-
- public boolean isTransactionManaged() {
- return transactionManaged;
- }
-
- @Override
- public String getSessionId() {
- return sessionId;
- }
-
- public boolean isBoundTo(final Channel channel) {
- return channel == initialChannel;
- }
-
- public long getActualTimeoutLengthWhenClosed() {
- return actualTimeoutLengthWhenClosed;
- }
-
- public Optional<CloseReason> getCloseReason() {
- return Optional.ofNullable(closeReason.get());
- }
-
- /**
- * Gets the script engine from the cached one in the {@link GremlinExecutor}.
- */
- public GremlinScriptEngine getScriptEngine(final SessionTask sessionTask, final String language) {
- return sessionTask.getGremlinExecutor().getScriptEngineManager().getEngineByName(language);
- }
-
- /**
- * Respond to the client with the specific timeout response for this Session implementation.
- * This is for situations where the Session hasn't started running.
- */
- protected abstract void sendTimeoutResponseForUncommencedTask();
-
- @Override
- public void setSessionCancelFuture(final ScheduledFuture<?> f) {
- if (!sessionCancelFuture.compareAndSet(null, f))
- throw new IllegalStateException("Session cancellation future is already set");
- }
-
- @Override
- public void setSessionFuture(final Future<?> f) {
- if (!sessionFuture.compareAndSet(null, f))
- throw new IllegalStateException("Session future is already set");
- }
-
- @Override
- public synchronized void triggerTimeout(final long timeout, final boolean causedBySession) {
- // triggering timeout triggers the stop of the session which will end in close()
- // for final cleanup
- final Future<?> f = sessionFuture.get();
- if (f != null && !f.isDone()) {
- if (closeReason.compareAndSet(null, causedBySession ? CloseReason.SESSION_TIMEOUT : CloseReason.REQUEST_TIMEOUT)) {
- actualTimeoutLengthWhenClosed = timeout;
-
- // if caused by a session timeout for a MultiTaskSession OR if it is a request timeout for a
- // SingleTaskSession request then we can just straight cancel() the session instance
- if (causedBySession || !sessionIdOnRequest)
- cancel(true);
- else {
- // in both MultiTaskSession and SingleTaskSession the thread gets set immediately at the start
- // of run() as it should (though "single" has little need for it). As triggerTimeout() for a
- // request MultiTaskSession can only be called AFTER we are deep in the run() there should be
- // no chance of race conditions or situations where the sessionThread is null at this point.
- if (sessionThread != null) {
- sessionThread.interrupt();
- } else {
- logger.debug("{} is a {} which cannot be interrupted as the thread running the session has not " +
- "been set - please check the implementation if this is not desirable",
- sessionId, this.getClass().getSimpleName());
- }
- }
- }
- }
- }
-
- protected void process(final SessionTask sessionTask) throws SessionException {
- final RequestMessage msg = sessionTask.getRequestMessage();
- final Map<String, Object> args = msg.getArgs();
- final Object gremlinToExecute = args.get(Tokens.ARGS_GREMLIN);
-
- // for strict transactions track the aliases used so that we can commit them and only them on close()
- if (sessionTask.getSettings().strictTransactionManagement)
- msg.optionalArgs(Tokens.ARGS_ALIASES).ifPresent(m -> aliasesUsedBySession.addAll(((Map<String,String>) m).values()));
-
- final Timer.Context timer = getMetricsTimer(sessionTask);
- try {
- // itty is optional as Bytecode could be a "graph operation" rather than a Traversal. graph operations
- // don't need to be iterated and handle their own lifecycle
- final Optional<Iterator<?>> itty = gremlinToExecute instanceof Bytecode ?
- fromBytecode(sessionTask, (Bytecode) gremlinToExecute) :
- Optional.of(fromScript(sessionTask, (String) gremlinToExecute));
-
- processAuditLog(sessionTask.getSettings(), sessionTask.getChannelHandlerContext(), gremlinToExecute);
-
- if (itty.isPresent())
- handleIterator(sessionTask, itty.get());
- } catch (Throwable t) {
- handleException(sessionTask, t);
- } finally {
- timer.stop();
- }
- }
-
- protected void handleException(final SessionTask sessionTask, final Throwable t) throws SessionException {
- if (t instanceof SessionException) throw (SessionException) t;
-
- final Optional<Throwable> possibleSpecialException = determineIfSpecialException(t);
- if (possibleSpecialException.isPresent()) {
- final Throwable special = possibleSpecialException.get();
- final ResponseMessage.Builder specialResponseMsg = ResponseMessage.build(sessionTask.getRequestMessage()).
- statusMessage(special.getMessage()).
- statusAttributeException(special);
- if (special instanceof TemporaryException) {
- specialResponseMsg.code(ResponseStatusCode.SERVER_ERROR_TEMPORARY);
- } else if (special instanceof Failure) {
- final Failure failure = (Failure) special;
- specialResponseMsg.code(ResponseStatusCode.SERVER_ERROR_FAIL_STEP).
- statusAttribute(Tokens.STATUS_ATTRIBUTE_FAIL_STEP_MESSAGE, failure.format());
- }
- throw new SessionException(special.getMessage(), specialResponseMsg.create());
- }
-
- final Throwable root = ExceptionHelper.getRootCause(t);
-
- if (root instanceof TimedInterruptTimeoutException) {
- // occurs when the TimedInterruptCustomizerProvider is in play
- final String msg = String.format("A timeout occurred within the script during evaluation of [%s] - consider increasing the limit given to TimedInterruptCustomizerProvider",
- sessionTask.getRequestMessage().getRequestId());
- throw new SessionException(msg, root, ResponseMessage.build(sessionTask.getRequestMessage())
- .code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
- .statusMessage("Timeout during script evaluation triggered by TimedInterruptCustomizerProvider")
- .create());
- }
-
- if (root instanceof TimeoutException) {
- final String errorMessage = String.format("Script evaluation exceeded the configured threshold for request [%s]",
- sessionTask.getRequestMessage().getRequestId());
- throw new SessionException(errorMessage, root, ResponseMessage.build(sessionTask.getRequestMessage())
- .code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
- .statusMessage(t.getMessage())
- .create());
- }
-
- if (root instanceof InterruptedException ||
- root instanceof TraversalInterruptedException ||
- root instanceof InterruptedIOException) {
- String msg = "Processing interrupted but the reason why was not known";
- switch (closeReason.get()) {
- case CHANNEL_CLOSED:
- msg = "Processing interrupted because the channel was closed";
- break;
- case SESSION_TIMEOUT:
- msg = String.format("Session closed - %s - sessionLifetimeTimeout of %s ms exceeded", sessionId, actualTimeoutLengthWhenClosed);
- break;
- case REQUEST_TIMEOUT:
- msg = String.format("Evaluation exceeded timeout threshold of %s ms", actualTimeoutLengthWhenClosed);
- break;
- }
- final ResponseStatusCode code = closeReason.get() == CloseReason.SESSION_TIMEOUT || closeReason.get() == CloseReason.REQUEST_TIMEOUT ?
- ResponseStatusCode.SERVER_ERROR_TIMEOUT : ResponseStatusCode.SERVER_ERROR;
- throw new SessionException(msg, root, ResponseMessage.build(sessionTask.getRequestMessage())
- .code(code)
- .statusMessage(msg).create());
- }
-
- if (root instanceof MultipleCompilationErrorsException && root.getMessage().contains("Method too large") &&
- ((MultipleCompilationErrorsException) root).getErrorCollector().getErrorCount() == 1) {
- final String errorMessage = String.format("The Gremlin statement that was submitted exceeds the maximum compilation size allowed by the JVM, please split it into multiple smaller statements - %s", trimMessage(sessionTask.getRequestMessage()));
- logger.warn(errorMessage);
- throw new SessionException(errorMessage, root, ResponseMessage.build(sessionTask.getRequestMessage())
- .code(ResponseStatusCode.SERVER_ERROR_EVALUATION)
- .statusMessage(errorMessage)
- .statusAttributeException(root).create());
- }
-
- // GroovyRuntimeException will hit a pretty wide range of eval type errors, like MissingPropertyException,
- // CompilationFailedException, MissingMethodException, etc. If more specific handling is required then
- // try to catch it earlier above.
- if (root instanceof GroovyRuntimeException ||
- root instanceof VerificationException ||
- root instanceof ScriptException) {
- throw new SessionException(root.getMessage(), root, ResponseMessage.build(sessionTask.getRequestMessage())
- .code(ResponseStatusCode.SERVER_ERROR_EVALUATION)
- .statusMessage(root.getMessage())
- .statusAttributeException(root).create());
- }
-
- throw new SessionException(root.getClass().getSimpleName() + ": " + root.getMessage(), root,
- ResponseMessage.build(sessionTask.getRequestMessage())
- .code(ResponseStatusCode.SERVER_ERROR)
- .statusAttributeException(root)
- .statusMessage(root.getMessage()).create());
- }
-
- /**
- * Used to decrease the size of a Gremlin script that triggered a "method too large" exception so that it
- * doesn't log a massive text string nor return a large error message.
- */
- private RequestMessage trimMessage(final RequestMessage msg) {
- final RequestMessage trimmedMsg = RequestMessage.from(msg).create();
- if (trimmedMsg.getArgs().containsKey(Tokens.ARGS_GREMLIN))
- trimmedMsg.getArgs().put(Tokens.ARGS_GREMLIN, trimmedMsg.getArgs().get(Tokens.ARGS_GREMLIN).toString().substring(0, 1021) + "...");
-
- return trimmedMsg;
- }
-
- /**
- * Check if any exception in the chain is {@link TemporaryException} or {@link Failure} then respond with the
- * right error code so that the client knows to retry.
- */
- protected static Optional<Throwable> determineIfSpecialException(final Throwable ex) {
- return Stream.of(ExceptionUtils.getThrowables(ex)).
- filter(i -> i instanceof TemporaryException || i instanceof Failure).findFirst();
- }
-
- /**
- * Removes the session from the session list and cancels the future that manages the lifetime of the session.
- */
- @Override
- public synchronized void close() {
- // already closing/closed
- if (!sessions.containsKey(sessionId)) return;
-
- sessions.remove(sessionId);
-
- if (sessionCancelFuture.get() != null) {
- final ScheduledFuture<?> f = sessionCancelFuture.get();
- if (!f.isDone()) f.cancel(true);
- }
- }
-
- /**
- * Constructs an {@code Iterator} from the results of a script evaluation provided in the {@link SessionTask}.
- *
- * @param sessionTask The session task which can be used as a context in constructing the {@code Iterator}
- * @param script The script extracted by the calling method from the {@code sessionTask}
- */
- protected Iterator<?> fromScript(final SessionTask sessionTask, final String script) throws Exception {
- final RequestMessage msg = sessionTask.getRequestMessage();
- final Map<String, Object> args = msg.getArgs();
- final String language = args.containsKey(Tokens.ARGS_LANGUAGE) ? (String) args.get(Tokens.ARGS_LANGUAGE) : "gremlin-groovy";
- return IteratorUtils.asIterator(getScriptEngine(sessionTask, language).eval(
- script, mergeBindingsFromRequest(sessionTask, getWorkerBindings())));
- }
-
- /**
- * Constructs an {@code Iterator} from {@link Bytecode} provided in the {@link SessionTask}. If the {@link Bytecode}
- * is found to evalute to a {@link GraphOp} then it is processed and an empty {@code Optional} is returned.
- *
- * @param sessionTask The session task which can be used as a context in constructing the {@code Iterator}
- * @param bytecode The {@link Bytecode} extracted by the calling method from the {@code sessionTask}
- */
- protected Optional<Iterator<?>> fromBytecode(final SessionTask sessionTask, final Bytecode bytecode) throws Exception {
- final RequestMessage msg = sessionTask.getRequestMessage();
-
- final Traversal.Admin<?, ?> traversal;
- final Map<String, String> aliases = (Map<String, String>) msg.optionalArgs(Tokens.ARGS_ALIASES).get();
- final GraphManager graphManager = sessionTask.getGraphManager();
- final String traversalSourceName = aliases.entrySet().iterator().next().getValue();
- final TraversalSource g = graphManager.getTraversalSource(traversalSourceName);
-
- // handle bytecode based graph operations like commit/rollback commands
- if (BytecodeHelper.isGraphOperation(bytecode)) {
- handleGraphOperation(sessionTask, bytecode, g.getGraph());
- return Optional.empty();
- } else {
-
- final Optional<String> lambdaLanguage = BytecodeHelper.getLambdaLanguage(bytecode);
- if (!lambdaLanguage.isPresent())
- traversal = JavaTranslator.of(g).translate(bytecode);
- else {
- final SimpleBindings bindings = new SimpleBindings();
- bindings.put(traversalSourceName, g);
- traversal = sessionTask.getGremlinExecutor().getScriptEngineManager().
- getEngineByName(lambdaLanguage.get()).eval(bytecode, bindings, traversalSourceName);
- }
-
- // compile the traversal - without it getEndStep() has nothing in it
- traversal.applyStrategies();
-
- return Optional.of(new TraverserIterator(traversal));
- }
- }
-
- protected Bindings getWorkerBindings() throws SessionException {
- return new SimpleBindings(graphManager.getAsBindings());
- }
-
- protected Bindings mergeBindingsFromRequest(final SessionTask sessionTask, final Bindings bindings) throws SessionException {
- // alias any global bindings to a different variable.
- final RequestMessage msg = sessionTask.getRequestMessage();
- if (msg.getArgs().containsKey(Tokens.ARGS_ALIASES)) {
- final Map<String, String> aliases = (Map<String, String>) msg.getArgs().get(Tokens.ARGS_ALIASES);
- for (Map.Entry<String,String> aliasKv : aliases.entrySet()) {
- boolean found = false;
-
- // first check if the alias refers to a Graph instance
- final Graph graph = sessionTask.getGraphManager().getGraph(aliasKv.getValue());
- if (null != graph) {
- bindings.put(aliasKv.getKey(), graph);
- found = true;
- }
-
- // if the alias wasn't found as a Graph then perhaps it is a TraversalSource - it needs to be
- // something
- if (!found) {
- final TraversalSource ts = sessionTask.getGraphManager().getTraversalSource(aliasKv.getValue());
- if (null != ts) {
- bindings.put(aliasKv.getKey(), ts);
- found = true;
- }
- }
-
- // this validation is important to calls to GraphManager.commit() and rollback() as they both
- // expect that the aliases supplied are valid
- if (!found) {
- final String error = String.format("Could not alias [%s] to [%s] as [%s] not in the Graph or TraversalSource global bindings",
- aliasKv.getKey(), aliasKv.getValue(), aliasKv.getValue());
- throw new SessionException(error, ResponseMessage.build(msg)
- .code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(error).create());
- }
- }
- } else {
- // there's no bindings so determine if that's ok with Gremlin Server
- if (sessionTask.getSettings().strictTransactionManagement) {
- final String error = "Gremlin Server is configured with strictTransactionManagement as 'true' - the 'aliases' arguments must be provided";
- throw new SessionException(error, ResponseMessage.build(msg)
- .code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(error).create());
- }
- }
-
- // add any bindings to override any other supplied
- Optional.ofNullable((Map<String, Object>) msg.getArgs().get(Tokens.ARGS_BINDINGS)).ifPresent(bindings::putAll);
- return bindings;
- }
-
- /**
- * Provides a generic way of iterating a result set back to the client.
- *
- * @param sessionTask The Gremlin Server {@link SessionTask} object containing settings, request message, etc.
- * @param itty The result to iterator
- */
- protected void handleIterator(final SessionTask sessionTask, final Iterator<?> itty) throws InterruptedException {
- final ChannelHandlerContext nettyContext = sessionTask.getChannelHandlerContext();
- final RequestMessage msg = sessionTask.getRequestMessage();
- final Settings settings = sessionTask.getSettings();
- boolean warnOnce = false;
-
- // sessionless requests are always transaction managed, but in-session requests are configurable.
- final boolean managedTransactionsForRequest = transactionManaged ?
- 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()) {
- final Map<String, Object> attributes = generateStatusAttributes(sessionTask,ResponseStatusCode.NO_CONTENT, itty);
- // 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 (managedTransactionsForRequest)
- closeTransaction(sessionTask, Transaction.Status.COMMIT);
-
- sessionTask.writeAndFlush(ResponseMessage.build(msg)
- .code(ResponseStatusCode.NO_CONTENT)
- .statusAttributes(attributes)
- .create());
- return;
- }
-
- // the batch size can be overridden by the request
- final int resultIterationBatchSize = (Integer) msg.optionalArgs(Tokens.ARGS_BATCH_SIZE)
- .orElse(settings.resultIterationBatchSize);
- List<Object> aggregate = new ArrayList<>(resultIterationBatchSize);
-
- // 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 = itty.hasNext();
-
- while (hasMore) {
- if (Thread.interrupted()) throw new InterruptedException();
-
- // check if an implementation needs to force flush the aggregated results before the iteration batch
- // size is reached.
- // todo: what implementation does this?! can we kill it going forward - seems always false
- // final boolean forceFlush = isForceFlushed(nettyContext, msg, itty);
- final boolean forceFlush = false;
-
- // have to check the aggregate size because it is possible that the channel is not writeable (below)
- // 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.
- //
- // there is a need to check hasNext() on the iterator because if the channel is not writeable the
- // previous pass through the while loop will have next()'d the iterator and if it is "done" then a
- // NoSuchElementException will raise its head. also need a check to ensure that this iteration doesn't
- // require a forced flush which can be forced by sub-classes.
- //
- // this could be placed inside the isWriteable() portion of the if-then below but it seems better to
- // allow iteration to continue into a batch if that is possible rather than just doing nothing at all
- // while waiting for the client to catch up
- if (aggregate.size() < resultIterationBatchSize && itty.hasNext() && !forceFlush) 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.
- //
- // clients might decide to close the Netty channel to the server with a CloseWebsocketFrame after errors
- // like CorruptedFrameException. On the server, although the channel gets closed, there might be some
- // executor threads waiting for watermark to clear which will not clear in these cases since client has
- // already given up on these requests. This leads to these executors waiting for the client to consume
- // results till the timeout. checking for isActive() should help prevent that.
- if (nettyContext.channel().isActive() && nettyContext.channel().isWritable()) {
- if (forceFlush || aggregate.size() == resultIterationBatchSize || !itty.hasNext()) {
- final ResponseStatusCode code = itty.hasNext() ? ResponseStatusCode.PARTIAL_CONTENT : ResponseStatusCode.SUCCESS;
- Frame frame = null;
- try {
- frame = makeFrame(sessionTask, aggregate, code, itty);
- } catch (Exception ex) {
- // a frame may use a Bytebuf which is a countable release - if it does not get written
- // downstream it needs to be released here
- if (frame != null) frame.tryRelease();
-
- // exception is handled in makeFrame() - serialization error gets written back to driver
- // at that point
- if (managedTransactionsForRequest)
- closeTransaction(sessionTask, Transaction.Status.ROLLBACK);
- break;
- }
-
- // track whether there is anything left in the iterator because it needs to be accessed after
- // the transaction could be closed - in that case a call to hasNext() could open a new transaction
- // unintentionally
- final boolean moreInIterator = itty.hasNext();
-
- try {
- // only need to reset the aggregation list if there's more stuff to write
- if (moreInIterator)
- aggregate = new ArrayList<>(resultIterationBatchSize);
- else {
- // iteration and serialization are both complete which means this finished successfully. note that
- // errors internal to script eval or timeout will rollback given GremlinServer's global configurations.
- // 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 (managedTransactionsForRequest)
- closeTransaction(sessionTask, Transaction.Status.COMMIT);
-
- // 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
- // hasNext()
- hasMore = false;
- }
- } catch (Exception ex) {
- // a frame may use a Bytebuf which is a countable release - if it does not get written
- // downstream it needs to be released here
- if (frame != null) frame.tryRelease();
- throw ex;
- }
-
- if (!moreInIterator) iterateComplete(sessionTask, itty);
-
- // the flush is called after the commit has potentially occurred. in this way, if a commit was
- // required then it will be 100% complete before the client receives it. the "frame" at this point
- // should have completely detached objects from the transaction (i.e. serialization has occurred)
- // so a new one should not be opened on the flush down the netty pipeline
- sessionTask.writeAndFlush(code, frame);
- }
- } else {
- // don't keep triggering this warning over and over again for the same request
- if (!warnOnce) {
- logger.warn("Pausing response writing as writeBufferHighWaterMark exceeded on {} - writing will continue once client has caught up", msg);
- warnOnce = true;
- }
-
- // since the client is lagging we can hold here for a period of time for the client to catch up.
- // this isn't blocking the IO thread - just a worker.
- TimeUnit.MILLISECONDS.sleep(10);
- }
- }
- }
-
- /**
- * If {@link Bytecode} is detected to contain a {@link GraphOp} then it gets processed by this method.
- */
- protected void handleGraphOperation(final SessionTask sessionTask, final Bytecode bytecode, final Graph graph) throws Exception {
- final RequestMessage msg = sessionTask.getRequestMessage();
- if (graph.features().graph().supportsTransactions()) {
- if (TX_COMMIT.equals(bytecode) || TX_ROLLBACK.equals(bytecode)) {
- final boolean commit = TX_COMMIT.equals(bytecode);
- closeTransaction(sessionTask, commit ? Transaction.Status.COMMIT : Transaction.Status.ROLLBACK);
-
- // write back a no-op for success
- final Map<String, Object> attributes = generateStatusAttributes(sessionTask,
- ResponseStatusCode.NO_CONTENT, Collections.emptyIterator());
- sessionTask.writeAndFlush(ResponseMessage.build(msg)
- .code(ResponseStatusCode.NO_CONTENT)
- .statusAttributes(attributes)
- .create());
- } else {
- throw new IllegalStateException(String.format(
- "Bytecode in request is not a recognized graph operation: %s", bytecode.toString()));
- }
- } else {
- throw Graph.Exceptions.transactionsNotSupported();
- }
- }
-
- /**
- * Called when iteration within {@link #handleIterator(SessionTask, Iterator)} is on its final pass and the final
- * frame is about to be sent back to the client. This method only gets called on successful iteration of the
- * entire result.
- */
- protected void iterateComplete(final SessionTask sessionTask, final Iterator<?> itty) {
- // do nothing by default
- }
-
- /**
- * Generates response status meta-data to put on a {@link ResponseMessage}.
- *
- * @param itty a reference to the current {@link Iterator} of results - it is not meant to be forwarded in
- * this method
- */
- protected Map<String, Object> generateStatusAttributes(final SessionTask sessionTask,
- final ResponseStatusCode code, final Iterator<?> itty) {
- // only return server metadata on the last message
- if (itty.hasNext()) return Collections.emptyMap();
-
- final Map<String, Object> metaData = new HashMap<>();
- metaData.put(Tokens.ARGS_HOST, sessionTask.getChannelHandlerContext().channel().remoteAddress().toString());
-
- return metaData;
- }
-
- /**
- * Generates response result meta-data to put on a {@link ResponseMessage}.
- *
- * @param itty a reference to the current {@link Iterator} of results - it is not meant to be forwarded in
- * this method
- */
- protected Map<String, Object> generateResponseMetaData(final SessionTask sessionTask,
- final ResponseStatusCode code, final Iterator<?> itty) {
- return Collections.emptyMap();
- }
-
- protected Frame makeFrame(final SessionTask sessionTask, final List<Object> aggregate,
- final ResponseStatusCode code, final Iterator<?> itty) throws Exception {
- final RequestMessage msg = sessionTask.getRequestMessage();
- final ChannelHandlerContext nettyContext = sessionTask.getChannelHandlerContext();
- final MessageSerializer serializer = nettyContext.channel().attr(StateKey.SERIALIZER).get();
- final boolean useBinary = nettyContext.channel().attr(StateKey.USE_BINARY).get();
-
- final Map<String, Object> responseMetaData = generateResponseMetaData(sessionTask, code, itty);
- final Map<String, Object> statusAttributes = generateStatusAttributes(sessionTask, code, itty);
- try {
- sessionTask.handleDetachment(aggregate);
-
- if (useBinary) {
- return new Frame(serializer.serializeResponseAsBinary(ResponseMessage.build(msg)
- .code(code)
- .statusAttributes(statusAttributes)
- .responseMetaData(responseMetaData)
- .result(aggregate).create(), nettyContext.alloc()));
- } else {
- // the expectation is that the GremlinTextRequestDecoder will have placed a MessageTextSerializer
- // instance on the channel.
- final MessageTextSerializer textSerializer = (MessageTextSerializer) serializer;
- return new Frame(textSerializer.serializeResponseAsString(ResponseMessage.build(msg)
- .code(code)
- .statusAttributes(statusAttributes)
- .responseMetaData(responseMetaData)
- .result(aggregate).create(), nettyContext.alloc()));
- }
- } catch (Exception ex) {
- logger.warn("The result [{}] in the request {} could not be serialized and returned.", aggregate, msg.getRequestId(), ex);
- final String errorMessage = String.format("Error during serialization: %s", ExceptionHelper.getMessageFromExceptionOrCause(ex));
- final ResponseMessage error = ResponseMessage.build(msg.getRequestId())
- .statusMessage(errorMessage)
- .statusAttributeException(ex)
- .code(ResponseStatusCode.SERVER_ERROR_SERIALIZATION).create();
- sessionTask.writeAndFlush(error);
- throw ex;
- }
- }
-
- /**
- * Called right before a transaction starts within {@link #run()}. The default implementation checks for open
- * transactions and throws an exception if it finds any and generally assumes auto-transactions are enabled on
- * graphs (i.e. transaction automatically start on read/write).
- * <p/>
- * Providers who do not follow these sorts of transaction semantics should provide an override to this method.
- */
- protected void startTransaction(final SessionTask sessionTask) {
- if (graphManager.hasAnyOpenTransactions()) {
- // logger and handling in calling method
- throw new IllegalStateException(String.format(
- "Attempted to start transaction for %s but the transaction was already open",
- sessionTask.getRequestMessage().getRequestId()));
- }
- }
-
- /**
- * Close the transaction without a {@link SessionTask} which supplies {@code null} to that argument for
- * {@link #closeTransaction(SessionTask, Transaction.Status)}. This method is idempotent.
- */
- protected void closeTransaction(final Transaction.Status status) {
- closeTransaction(null, status);
- }
-
- /**
- * Tries to close the transaction but will catch exceptions and log them. This method is idempotent.
- */
- protected void closeTransactionSafely(final Transaction.Status status) {
- closeTransactionSafely(null, status);
- }
-
- /**
- * Tries to close the transaction but will catch exceptions and log them. This method is idempotent.
- */
- protected void closeTransactionSafely(final SessionTask sessionTask, final Transaction.Status status) {
- try {
- closeTransaction(sessionTask, status);
- } catch (Exception ex) {
- logger.error("Failed to close transaction", ex);
- }
- }
-
- private void processAuditLog(final Settings settings, final ChannelHandlerContext ctx, final Object gremlinToExecute) {
- if (settings.enableAuditLog) {
- AuthenticatedUser user = ctx.channel().attr(StateKey.AUTHENTICATED_USER).get();
- if (null == user) { // This is expected when using the AllowAllAuthenticator
- user = AuthenticatedUser.ANONYMOUS_USER;
- }
- String address = ctx.channel().remoteAddress().toString();
- if (address.startsWith("/") && address.length() > 1) address = address.substring(1);
- auditLogger.info("User {} with address {} requested: {}", user.getName(), address, gremlinToExecute);
- }
- }
-
- /**
- * Closes a transaction with commit or rollback. Strict transaction management settings are observed when
- * configured as such in {@link Settings#strictTransactionManagement} and when aliases are present on the
- * request in the current {@link SessionTask}. If the supplied {@link SessionTask} is {@code null} then "strict" is
- * bypassed so this form must be called with care. Bypassing is often useful to ensure that all transactions
- * are cleaned up when multiple graphs are referenced. Prefer calling {@link #closeTransaction(Transaction.Status)}
- * in this case instead. This method is idempotent.
- */
- protected void closeTransaction(final SessionTask sessionTask, final Transaction.Status status) {
- if (status != Transaction.Status.COMMIT && status != Transaction.Status.ROLLBACK)
- throw new IllegalStateException(String.format("Transaction.Status not supported: %s", status));
-
- final boolean commit = status == Transaction.Status.COMMIT;
- final boolean strict = sessionTask != null && sessionTask.getSettings().strictTransactionManagement;
-
- if (strict) {
- if (commit)
- graphManager.commit(new HashSet<>(aliasesUsedBySession));
- else
- graphManager.rollback(new HashSet<>(aliasesUsedBySession));
- } else {
- if (commit)
- graphManager.commitAll();
- else
- graphManager.rollbackAll();
- }
- }
-
- private Timer.Context getMetricsTimer(final SessionTask sessionTask) {
- // getting something other than bytecode or script at this point is unlikely as earlier validations
- // should have picked this up.
- switch (sessionTask.getRequestContentType()) {
- case BYTECODE:
- return Session.traversalOpTimer.time();
- case SCRIPT:
- return Session.evalOpTimer.time();
- default:
- throw new IllegalStateException("Unrecognized content of the 'gremlin' argument in the request");
- }
- }
-}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/MultiTaskSession.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/MultiTaskSession.java
deleted file mode 100644
index fb623a9368..0000000000
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/MultiTaskSession.java
+++ /dev/null
@@ -1,316 +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.server.handler;
-
-import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
-import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
-import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor;
-import org.apache.tinkerpop.gremlin.groovy.jsr223.GroovyCompilerGremlinPlugin;
-import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptEngine;
-import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptEngineManager;
-import org.apache.tinkerpop.gremlin.server.Settings;
-import org.apache.tinkerpop.gremlin.structure.Transaction;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.script.Bindings;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static org.apache.tinkerpop.gremlin.server.op.session.SessionOpProcessor.CONFIG_GLOBAL_FUNCTION_CACHE_ENABLED;
-
-/**
- * A {@link Session} implementation that queues tasks given to it and executes them in a serial fashion within the
- * same thread which thus allows multiple tasks to be executed in the same transaction. The first {@link SessionTask}
- * to execute is supplied on the constructor and additional ones may be added as they arrive with
- * {@link #submitTask(SessionTask)} where they will be added to a queue where they will await execution in the thread
- * bound to this session.
- */
-public class MultiTaskSession extends AbstractSession {
- private static final Logger logger = LoggerFactory.getLogger(MultiTaskSession.class);
- protected final BlockingQueue<SessionTask> queue;
- private final AtomicBoolean ending = new AtomicBoolean(false);
- private final ScheduledExecutorService scheduledExecutorService;
- private final GremlinScriptEngineManager scriptEngineManager;
- private ScheduledFuture<?> requestCancelFuture;
- private Bindings bindings;
- private final SessionTask initialSessionTask;
-
- /**
- * Creates a new {@code MultiTaskSession} object providing the initial starting {@link SessionTask} that gets
- * executed by the session when it starts.
- *
- * @param initialSessionTask The tasks that starts the session.
- * @param sessionId The id of the session
- * @param sessions The session id to {@link Session} instances mapping
- */
- public MultiTaskSession(final SessionTask initialSessionTask, final String sessionId,
- final ConcurrentMap<String, Session> sessions) {
- super(initialSessionTask, sessionId, false, sessions);
-
- this.initialSessionTask = initialSessionTask;
- queue = new LinkedBlockingQueue<>(initialSessionTask.getSettings().maxSessionTaskQueueSize);
-
- // using a global function cache is cheaper than creating a new on per session especially if you have to
- // create a lot of sessions. it will generate a ton of throw-away objects. mostly keeping the option open
- // to not use it to preserve the ability to use the old functionality if wanted or if there is some specific
- // use case with sessions that needs it. if we wanted this could eventually become a per-request option
- // so that the client could control it as necessary and get scriptengine isolation if they need it.
- if (initialSessionTask.getSettings().useCommonEngineForSessions)
- scriptEngineManager = initialSessionTask.getGremlinExecutor().getScriptEngineManager();
- else
- scriptEngineManager = initializeGremlinExecutor(initialSessionTask).getScriptEngineManager();
-
- scheduledExecutorService = initialSessionTask.getScheduledExecutorService();
-
- // would be odd if this tossed an exception because on construction the queue should be empty,
- // but let's handle it in the same fashion as if it had been rejected by the queue itself but put up
- // an additional log because something would be rather off here if we hit this condition
- if (!submitTask(initialSessionTask)) {
- logger.error("Task {} rejected on creation of the {} for session {}",
- initialSessionTask.getRequestMessage().getRequestId(), this.getClass().getSimpleName(),
- getSessionId());
- final String msg = String.format("Task %s rejected from session %s",
- initialSessionTask.getRequestMessage().getRequestId(), getSessionId());
- throw new RejectedExecutionException(msg);
- }
- }
-
- /**
- * Gets the script engine specific to this session which is dependent on the
- * {@link Settings#useCommonEngineForSessions} configuration.
- */
- @Override
- public GremlinScriptEngine getScriptEngine(final SessionTask sessionTask, final String language) {
- return scriptEngineManager.getEngineByName(language);
- }
-
- @Override
- public boolean isAcceptingTasks() {
- return !ending.get();
- }
-
- @Override
- public void sendTimeoutResponseForUncommencedTask() {
- this.close(); // Need to close as this task won't ever start to run, so it should just end.
- initialSessionTask.sendTimeoutResponse(
- String.format("A session timeout occurred during traversal evaluation of [%s] - consider increasing the limit given to sessionLifetimeTimeout", initialSessionTask.getRequestMessage()));
- }
-
- @Override
- public boolean submitTask(final SessionTask sessionTask) throws RejectedExecutionException {
- try {
- return isAcceptingTasks() && queue.add(sessionTask);
- } catch (IllegalStateException ise) {
- final String msg = String.format("Task %s rejected from session %s",
- sessionTask.getRequestMessage().getRequestId(), getSessionId());
- throw new RejectedExecutionException(msg);
- }
- }
-
- @Override
- public void run() {
- sessionTaskStarted.set(true);
-
- // allow the Session to know about the thread that is running it which will be a thread from the gremlinPool
- // the thread really only has relevance once the session has started. this thread is then held below in the
- // while() loop awaiting items to arrive on the request queue. it will hold until the session is properly
- // closed by the client or through interruption/error
- this.sessionThread = Thread.currentThread();
-
- // there must be one item in the queue at least since addTask() gets called before the worker
- // is ever started
- SessionTask sessionTask = queue.poll();
- if (null == sessionTask)
- throw new IllegalStateException(String.format("Worker has no initial context for session: %s", getSessionId()));
-
- try {
- startTransaction(sessionTask);
- try {
- while (true) {
- // schedule timeout for the current request from the queue
- final long seto = sessionTask.getRequestTimeout();
- requestCancelFuture = scheduledExecutorService.schedule(
- () -> this.triggerTimeout(seto, false),
- seto, TimeUnit.MILLISECONDS);
-
- // only stop processing stuff in the queue if this session isn't configured to hold state between
- // exceptions (i.e. the old OpProcessor way) or if this session is closing down by certain death
- // (i.e. channel close or lifetime session timeout)
- try {
- process(sessionTask);
- } catch (SessionException ex) {
- if (!maintainStateAfterException || closeReason.get() == CloseReason.CHANNEL_CLOSED ||
- closeReason.get() == CloseReason.SESSION_TIMEOUT) {
- throw ex;
- }
-
- // reset the close reason as we are maintaining state
- closeReason.set(null);
-
- logger.warn(ex.getMessage(), ex);
- sessionTask.writeAndFlush(ex.getResponseMessage());
- }
-
- // work is done within the timeout period so cancel it
- cancelRequestTimeout();
-
- sessionTask = queue.take();
- }
- } catch (Exception ex) {
- stopAcceptingRequests();
-
- // the current context gets its exception handled...
- handleException(sessionTask, ex);
- }
- } catch (SessionException rexex) {
- // if the close reason isn't already set then things stopped during gremlin execution somewhere and not
- // more external issues like channel close or timeouts.
- closeReason.compareAndSet(null, CloseReason.PROCESSING_EXCEPTION);
-
- // remaining work items in the queue are ignored since this worker is closing. must send
- // back some sort of response to satisfy the client. writeAndFlush code is different than
- // the ResponseMessage as we don't want the message to be "final" for the Context. that
- // status must be reserved for the message that caused the error
- for (SessionTask st : queue) {
- st.write(ResponseStatusCode.PARTIAL_CONTENT, ResponseMessage.build(st.getRequestMessage())
- .code(ResponseStatusCode.SERVER_ERROR)
- .statusMessage(String.format(
- "An earlier request [%s] failed prior to this one having a chance to execute",
- sessionTask.getRequestMessage().getRequestId())).create());
- }
-
- // exception should trigger a rollback in the session. a more focused rollback may have occurred
- // during process() and the related result iteration IF transaction management was enabled on
- // the request
- closeTransactionSafely(Transaction.Status.ROLLBACK);
-
- // the current context could already be completed with SUCCESS and we're just waiting for another
- // one to show up while a timeout occurs or the channel closes. in these cases, this would be a valid
- // close in all likelihood so there's no reason to log or alert the client as the client already has
- // the best answer
- if (!sessionTask.isFinalResponseWritten()) {
- logger.warn(rexex.getMessage(), rexex);
- sessionTask.write(rexex.getResponseMessage());
- }
- } finally {
- // if this is a normal end to the session or if there is some general processing exception which tanks
- // the entire session or if the session life timeout is exceeded then the session needs to be removed
- // and everything cleaned up
- if (closeReason.compareAndSet(null, CloseReason.EXIT_PROCESSING) ||
- closeReason.get() == CloseReason.PROCESSING_EXCEPTION ||
- closeReason.get() == CloseReason.SESSION_TIMEOUT) {
- close();
-
- // the session is now in a state where it is no longer in the set of current sessions so flush
- // remaining messages to the transport, if any. the remaining message should be failures from
- // the SessionException in the catch. this prevents an unlikely case where a fast client can
- // get ahead of the server and start to send messages to a technically errored out session. that
- // in itself is not necessarily bad but it makes tests fail sometimes because the tests are
- // designed to assert in the same fashion for OpProcessor and UnifiedChannelizer infrastructure
- // and they are slightly at odds with each other in certain situations.
- sessionTask.flush();
- }
- }
- }
-
- /**
- * This method stops incoming requests from being added to the session queue. It then cancels the session lifetime
- * timeout and then calls {@link super#close()} which will then interrupt this {@link #run()} which will clear out
- * remaining requests in the queue. This latter part of the close, where the interruption is concerned, is an
- * asynchronous operation and will not block as it finishes its processing in the "gremlinPool" thread that was
- * originally handling the work.
- */
- @Override
- public void close() {
- stopAcceptingRequests();
- cancelRequestTimeout();
- super.close();
- logger.debug("Session {} closed", getSessionId());
- }
-
- private void cancelRequestTimeout() {
- if (requestCancelFuture != null && !requestCancelFuture.isDone())
- requestCancelFuture.cancel(true);
- else
- logger.debug("Could not cancel request timeout for {} - {}", getSessionId(), requestCancelFuture);
- }
-
- private void stopAcceptingRequests() {
- if (ending.compareAndSet(false, true))
- cancel(true);
- }
-
- @Override
- protected Bindings getWorkerBindings() throws SessionException {
- if (null == bindings)
- bindings = super.getWorkerBindings();
- return this.bindings;
- }
-
- protected GremlinExecutor initializeGremlinExecutor(final SessionTask sessionTask) {
- final Settings settings = sessionTask.getSettings();
- final ExecutorService executor = sessionTask.getGremlinExecutor().getExecutorService();
- final boolean useGlobalFunctionCache = settings.useGlobalFunctionCacheForSessions;
-
- // these initial settings don't matter so much as we don't really execute things through the
- // GremlinExecutor directly. Just doing all this setup to make GremlinExecutor do the work of
- // rigging up the GremlinScriptEngineManager.
- final GremlinExecutor.Builder gremlinExecutorBuilder = GremlinExecutor.build()
- .evaluationTimeout(settings.getEvaluationTimeout())
- .executorService(executor)
- .globalBindings(graphManager.getAsBindings())
- .scheduledExecutorService(scheduledExecutorService);
-
- settings.scriptEngines.forEach((k, v) -> {
- // use plugins if they are present
- if (!v.plugins.isEmpty()) {
- // make sure that server related classes are available at init. the LifeCycleHook stuff will be
- // added explicitly via configuration using GremlinServerGremlinModule in the yaml. need to override
- // scriptengine settings with SessionOpProcessor specific ones as the processing for sessions is
- // different and a global setting may not make sense for a session
- if (v.plugins.containsKey(GroovyCompilerGremlinPlugin.class.getName())) {
- v.plugins.get(GroovyCompilerGremlinPlugin.class.getName()).put(CONFIG_GLOBAL_FUNCTION_CACHE_ENABLED, useGlobalFunctionCache);
- } else {
- final Map<String,Object> pluginConf = new HashMap<>();
- pluginConf.put(CONFIG_GLOBAL_FUNCTION_CACHE_ENABLED, useGlobalFunctionCache);
- v.plugins.put(GroovyCompilerGremlinPlugin.class.getName(), pluginConf);
- }
-
- gremlinExecutorBuilder.addPlugins(k, v.plugins);
- }
- });
-
- return gremlinExecutorBuilder.create();
- }
-
- @Override
- public String toString() {
- return String.format("%s - session: %s", MultiTaskSession.class.getSimpleName(), getSessionId());
- }
-}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/Session.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/Session.java
deleted file mode 100644
index 3849b84cd9..0000000000
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/Session.java
+++ /dev/null
@@ -1,84 +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.server.handler;
-
-import com.codahale.metrics.Timer;
-import io.netty.channel.Channel;
-import org.apache.tinkerpop.gremlin.server.GremlinServer;
-import org.apache.tinkerpop.gremlin.server.util.MetricManager;
-
-import java.util.concurrent.Future;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ScheduledFuture;
-
-import static com.codahale.metrics.MetricRegistry.name;
-
-/**
- * Requests that arrive through the {@link UnifiedHandler} are all processed within a {@code Session} implementation.
- * A session accepts a {@link SessionTask} as the work it is meant to perform and those tasks are constructed from
- * incoming requests. A session may be as short-lived as processing a single task before exiting, as with the
- * {@link SingleTaskSession} or may be longer-lived by handling multiple tasks, as with the {@link MultiTaskSession}.
- */
-public interface Session extends Runnable {
-
- public static final Timer traversalOpTimer = MetricManager.INSTANCE.getTimer(name(GremlinServer.class, "op", "traversal"));
- public static final Timer evalOpTimer = MetricManager.INSTANCE.getTimer(name(GremlinServer.class, "op", "eval"));
-
- /**
- * Gets the identifier for the session.
- */
- String getSessionId();
-
- /**
- * Adds a task for session to complete.
- *
- * @throws RejectedExecutionException if the task cannot be queued
- */
- boolean submitTask(final SessionTask sessionTask) throws RejectedExecutionException;
-
- /**
- * Sets a reference to the job that will cancel this session if it exceeds its timeout period.
- */
- void setSessionCancelFuture(final ScheduledFuture<?> f);
-
- /**
- * Sets a reference to the job itself that is running this session.
- */
- void setSessionFuture(final Future<?> f);
-
- /**
- * Provides a general way to interrupt the session by way of a timeout.
- *
- * @param timeout the length of time that passed for the timeout to have triggered
- * @param causedBySession determines if the timeout triggered due to a particular request (i.e. {@code false} or
- * because of the session lifetime (i.e. {@code true}
- */
- void triggerTimeout(final long timeout, boolean causedBySession);
-
- /**
- * Determines if the supplied {@code Channel} object is the same as the one bound to the {@code Session}.
- */
- boolean isBoundTo(final Channel channel);
-
- /**
- * Determines if this session can accept additional tasks or not.
- */
- boolean isAcceptingTasks();
-
-}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SessionException.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SessionException.java
deleted file mode 100644
index 5705de2bc8..0000000000
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SessionException.java
+++ /dev/null
@@ -1,43 +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.server.handler;
-
-import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
-
-/**
- * An exception that holds the error-related {@link ResponseMessage} which is meant to be returned to the calling
- * client.
- */
-public class SessionException extends Exception {
- private final ResponseMessage responseMessage;
-
- public SessionException(final String message, final ResponseMessage responseMessage) {
- super(message);
- this.responseMessage = responseMessage;
- }
-
- public SessionException(final String message, final Throwable cause, final ResponseMessage responseMessage) {
- super(message, cause);
- this.responseMessage = responseMessage;
- }
-
- public ResponseMessage getResponseMessage() {
- return this.responseMessage;
- }
-}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SessionTask.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SessionTask.java
deleted file mode 100644
index 5dbe5e0bca..0000000000
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SessionTask.java
+++ /dev/null
@@ -1,42 +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.server.handler;
-
-import io.netty.channel.ChannelHandlerContext;
-import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
-import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor;
-import org.apache.tinkerpop.gremlin.server.Context;
-import org.apache.tinkerpop.gremlin.server.GraphManager;
-import org.apache.tinkerpop.gremlin.server.OpProcessor;
-import org.apache.tinkerpop.gremlin.server.Settings;
-
-import java.util.concurrent.ScheduledExecutorService;
-
-/**
- * A {@code SessionTask} equates to a particular incoming request to the {@link UnifiedHandler} and is analogous to
- * a {@link Context} in the {@link OpProcessor} approach to handling requests to the server.
- */
-public class SessionTask extends Context {
- public SessionTask(final RequestMessage requestMessage, final ChannelHandlerContext ctx,
- final Settings settings, final GraphManager graphManager,
- final GremlinExecutor gremlinExecutor,
- final ScheduledExecutorService scheduledExecutorService) {
- super(requestMessage, ctx, settings, graphManager, gremlinExecutor, scheduledExecutorService);
- }
-}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SingleTaskSession.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SingleTaskSession.java
deleted file mode 100644
index 8a786d08dc..0000000000
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/SingleTaskSession.java
+++ /dev/null
@@ -1,99 +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.server.handler;
-
-import org.apache.tinkerpop.gremlin.structure.Transaction;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.RejectedExecutionException;
-
-/**
- * A simple {@link Session} implementation that accepts one request, processes it and exits.
- */
-public class SingleTaskSession extends AbstractSession {
- private static final Logger logger = LoggerFactory.getLogger(SingleTaskSession.class);
- protected final SessionTask onlySessionTask;
-
- /**
- * Constructs a {@code SingleTaskSession} that has the task to execute supplied on construction.
- */
- public SingleTaskSession(final SessionTask onlySessionTask, final String sessionId,
- final ConcurrentMap<String, Session> sessions) {
- super(onlySessionTask, sessionId,true, sessions);
- this.onlySessionTask = onlySessionTask;
- }
-
- /**
- * The {@code SingleWorker} can only process one request so the initial construction of it already has the
- * request in it and no more can be added, therefore this method always return {@code false}.
- */
- @Override
- public boolean isAcceptingTasks() {
- return false;
- }
-
- @Override
- public void sendTimeoutResponseForUncommencedTask() {
- onlySessionTask.sendTimeoutResponse();
- }
-
- /**
- * This implementation only take a single {@link SessionTask} on construction - no additional tasks can be
- * submitted.
- */
- @Override
- public boolean submitTask(final SessionTask sessionTask) throws RejectedExecutionException {
- throw new UnsupportedOperationException("SingleWorker doesn't accept tasks beyond the one provided to the constructor");
- }
-
- @Override
- public void run() {
- sessionTaskStarted.set(true);
-
- // allow the Session to know about the thread that is running it - the thread really only has relevance
- // once the session has started.
- this.sessionThread = Thread.currentThread();
-
- try {
- startTransaction(onlySessionTask);
- process(onlySessionTask);
- } catch (SessionException we) {
- // if the close reason isn't already set then things stopped during gremlin execution somewhere and not
- // more external issues like channel close or timeouts.
- closeReason.compareAndSet(null, CloseReason.PROCESSING_EXCEPTION);
-
- logger.warn(we.getMessage(), we);
-
- // should have already rolledback - this is a safety valve
- closeTransactionSafely(onlySessionTask, Transaction.Status.ROLLBACK);
-
- onlySessionTask.writeAndFlush(we.getResponseMessage());
- } finally {
- closeReason.compareAndSet(null, CloseReason.EXIT_PROCESSING);
- close();
- }
- }
-
- @Override
- public String toString() {
- return String.format("%s - session: %s", SingleTaskSession.class.getSimpleName(), getSessionId());
- }
-}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/UnifiedHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/UnifiedHandler.java
deleted file mode 100644
index 318a0a0792..0000000000
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/UnifiedHandler.java
+++ /dev/null
@@ -1,356 +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.server.handler;
-
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.SimpleChannelInboundHandler;
-import io.netty.handler.timeout.IdleState;
-import io.netty.handler.timeout.IdleStateEvent;
-import io.netty.util.ReferenceCountUtil;
-import org.apache.tinkerpop.gremlin.util.Tokens;
-import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
-import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
-import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
-import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor;
-import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
-import org.apache.tinkerpop.gremlin.process.traversal.Operator;
-import org.apache.tinkerpop.gremlin.process.traversal.Order;
-import org.apache.tinkerpop.gremlin.process.traversal.Pop;
-import org.apache.tinkerpop.gremlin.process.traversal.Scope;
-import org.apache.tinkerpop.gremlin.server.Channelizer;
-import org.apache.tinkerpop.gremlin.server.GraphManager;
-import org.apache.tinkerpop.gremlin.server.Settings;
-import org.apache.tinkerpop.gremlin.server.channel.UnifiedChannelizer;
-import org.apache.tinkerpop.gremlin.structure.Column;
-import org.apache.tinkerpop.gremlin.structure.T;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-
-/**
- * Handler for websockets to be used with the {@link UnifiedChannelizer}.
- */
-@ChannelHandler.Sharable
-public class UnifiedHandler extends SimpleChannelInboundHandler<RequestMessage> {
- private static final Logger logger = LoggerFactory.getLogger(UnifiedHandler.class);
-
- protected final Settings settings;
- protected final GraphManager graphManager;
- protected final GremlinExecutor gremlinExecutor;
- protected final ScheduledExecutorService scheduledExecutorService;
- protected final ExecutorService sessionExecutor;
- protected final Channelizer channelizer;
-
- protected final ConcurrentMap<String, Session> sessions = new ConcurrentHashMap<>();
-
- /**
- * This may or may not be the full set of invalid binding keys. It is dependent on the static imports made to
- * Gremlin Server. This should get rid of the worst offenders though and provide a good message back to the
- * calling client.
- * <p/>
- * Use of {@code toUpperCase()} on the accessor values of {@link T} solves an issue where the {@code ScriptEngine}
- * ignores private scope on {@link T} and imports static fields.
- */
- protected static final Set<String> INVALID_BINDINGS_KEYS = new HashSet<>();
-
- static {
- INVALID_BINDINGS_KEYS.addAll(Arrays.asList(
- T.id.name(), T.key.name(),
- T.label.name(), T.value.name(),
- T.id.getAccessor(), T.key.getAccessor(),
- T.label.getAccessor(), T.value.getAccessor(),
- T.id.getAccessor().toUpperCase(), T.key.getAccessor().toUpperCase(),
- T.label.getAccessor().toUpperCase(), T.value.getAccessor().toUpperCase()));
-
- for (Column enumItem : Column.values()) {
- INVALID_BINDINGS_KEYS.add(enumItem.name());
- }
-
- for (Order enumItem : Order.values()) {
- INVALID_BINDINGS_KEYS.add(enumItem.name());
- }
-
- for (Operator enumItem : Operator.values()) {
- INVALID_BINDINGS_KEYS.add(enumItem.name());
- }
-
- for (Scope enumItem : Scope.values()) {
- INVALID_BINDINGS_KEYS.add(enumItem.name());
- }
-
- for (Pop enumItem : Pop.values()) {
- INVALID_BINDINGS_KEYS.add(enumItem.name());
- }
- }
-
- public UnifiedHandler(final Settings settings, final GraphManager graphManager,
- final GremlinExecutor gremlinExecutor,
- final ScheduledExecutorService scheduledExecutorService,
- final Channelizer channelizer) {
- this.settings = settings;
- this.graphManager = graphManager;
- this.gremlinExecutor = gremlinExecutor;
- this.scheduledExecutorService = scheduledExecutorService;
- this.channelizer = channelizer;
- this.sessionExecutor = gremlinExecutor.getExecutorService();
- }
-
- @Override
- protected void channelRead0(final ChannelHandlerContext ctx, final RequestMessage msg) throws Exception {
- try {
- try {
- validateRequest(msg, graphManager);
- } catch (SessionException we) {
- ctx.writeAndFlush(we.getResponseMessage());
- return;
- }
-
- // this is for backward compatibility for drivers still sending a close message. the close message was
- // removed in 3.5.0 but then added back for 3.5.2.
- if (msg.getOp().equals(Tokens.OPS_CLOSE)) {
- ctx.writeAndFlush(ResponseMessage.build(msg)
- .code(ResponseStatusCode.NO_CONTENT)
- .create());
- return;
- }
-
- final Optional<String> optMultiTaskSession = msg.optionalArgs(Tokens.ARGS_SESSION);
- final String sessionId = optMultiTaskSession.orElse(msg.getRequestId().toString());
-
- // the SessionTask is really a Context from OpProcessor. we still need the GremlinExecutor/ScriptEngine
- // config that is all rigged up into the server nicely right now so it seemed best to just keep the general
- // Context object but extend (essentially rename) it to SessionTask so that it better fits the nomenclature
- // we have here. when we drop OpProcessor stuff and rid ourselves of GremlinExecutor then we can probably
- // pare down the constructor for SessionTask further.
- final SessionTask sessionTask = new SessionTask(msg, ctx, settings, graphManager,
- gremlinExecutor, scheduledExecutorService);
-
- if (sessions.containsKey(sessionId)) {
- final Session session = sessions.get(sessionId);
-
- // check if the session is bound to this channel, thus one client per session
- if (!session.isBoundTo(ctx.channel())) {
- final String sessionClosedMessage = String.format("Session %s is not bound to the connecting client", sessionId);
- final ResponseMessage response = ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR)
- .statusMessage(sessionClosedMessage).create();
- ctx.writeAndFlush(response);
- return;
- }
-
- // if the session is done accepting tasks then error time
- if (session.isAcceptingTasks() && !session.submitTask(sessionTask)) {
- final String sessionClosedMessage = String.format(
- "Session %s is no longer accepting requests as it has been closed", sessionId);
- final ResponseMessage response = ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR)
- .statusMessage(sessionClosedMessage).create();
- ctx.writeAndFlush(response);
- }
- } else {
- // determine the type of session to start - one that processes the current request only and close OR
- // one that will process this current request and ones that may arrive in the future.
- final Session session = optMultiTaskSession.isPresent() ?
- createMultiTaskSession(sessionTask, sessionId) :
- createSingleTaskSession(sessionTask, sessionId);
-
- // queue the session to startup when a thread is ready to take it
- final Future<?> sessionFuture = sessionExecutor.submit(session);
- session.setSessionFuture(sessionFuture);
- sessions.put(sessionId, session);
-
- // determine the max session life. for multi that's going to be "session life" and for single that
- // will be the span of the request timeout
- final long seto = sessionTask.getRequestTimeout();
- final long sessionLife = optMultiTaskSession.isPresent() ? settings.sessionLifetimeTimeout : seto;
-
- // if timeout is enabled when greater than zero schedule up a timeout which is a session life timeout
- // for a multi or technically a request timeout for a single. this will be cancelled when the session
- // closes by way of other reasons (i.e. success or exception) - see AbstractSession#close()
- if (seto > 0) {
- final ScheduledFuture<?> sessionCancelFuture =
- scheduledExecutorService.schedule(
- () -> session.triggerTimeout(sessionLife, optMultiTaskSession.isPresent()),
- sessionLife, TimeUnit.MILLISECONDS);
- session.setSessionCancelFuture(sessionCancelFuture);
- }
- }
- } catch (RejectedExecutionException ree) {
- logger.warn(ree.getMessage());
-
- // generic message seems ok here? like, you would know what you were submitting on, i.e. session or
- // sessionless, when you got this error. probably don't need gory details.
- final ResponseMessage response = ResponseMessage.build(msg).code(ResponseStatusCode.TOO_MANY_REQUESTS)
- .statusMessage("Rate limiting").create();
- ctx.writeAndFlush(response);
- } finally {
- ReferenceCountUtil.release(msg);
- }
- }
-
- protected void validateRequest(final RequestMessage message, final GraphManager graphManager) throws SessionException {
- // close message just needs to be accounted for here as of 3.5.2. it will not contain a "gremlin" arg. unified
- // channelizer basically ignores the close message otherwise
- if (!message.getOp().equals(Tokens.OPS_CLOSE) && !message.optionalArgs(Tokens.ARGS_GREMLIN).isPresent()) {
- final String msg = String.format("A message with a [%s] op code requires a [%s] argument.", message.getOp(), Tokens.ARGS_GREMLIN);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
-
- if (message.optionalArgs(Tokens.ARGS_SESSION).isPresent()) {
- final Optional<Object> mtx = message.optionalArgs(Tokens.ARGS_MANAGE_TRANSACTION);
- if (mtx.isPresent() && !(mtx.get() instanceof Boolean)) {
- final String msg = String.format("%s argument must be of type boolean", Tokens.ARGS_MANAGE_TRANSACTION);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
-
- final Optional<Object> msae = message.optionalArgs(Tokens.ARGS_MAINTAIN_STATE_AFTER_EXCEPTION);
- if (msae.isPresent() && !(msae.get() instanceof Boolean)) {
- final String msg = String.format("%s argument must be of type boolean", Tokens.ARGS_MAINTAIN_STATE_AFTER_EXCEPTION);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
- } else {
- if (message.optionalArgs(Tokens.ARGS_MANAGE_TRANSACTION).isPresent()) {
- final String msg = String.format("%s argument only applies to requests made for sessions", Tokens.ARGS_MANAGE_TRANSACTION);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
-
- if (message.optionalArgs(Tokens.ARGS_MAINTAIN_STATE_AFTER_EXCEPTION).isPresent()) {
- final String msg = String.format("%s argument only applies to requests made for sessions", Tokens.ARGS_MAINTAIN_STATE_AFTER_EXCEPTION);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
- }
-
- if (message.optionalArgs(Tokens.ARGS_BINDINGS).isPresent()) {
- final Map bindings = (Map) message.getArgs().get(Tokens.ARGS_BINDINGS);
- if (IteratorUtils.anyMatch(bindings.keySet().iterator(), k -> null == k || !(k instanceof String))) {
- final String msg = String.format("The [%s] message is using one or more invalid binding keys - they must be of type String and cannot be null", Tokens.OPS_EVAL);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
-
- final Set<String> badBindings = IteratorUtils.set(IteratorUtils.<String>filter(bindings.keySet().iterator(), INVALID_BINDINGS_KEYS::contains));
- if (!badBindings.isEmpty()) {
- final String msg = String.format("The [%s] message supplies one or more invalid parameters key of [%s] - these are reserved names.", Tokens.OPS_EVAL, badBindings);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
-
- // ignore control bindings that get passed in with the "#jsr223" prefix - those aren't used in compilation
- if (IteratorUtils.count(IteratorUtils.filter(bindings.keySet().iterator(), k -> !k.toString().startsWith("#jsr223"))) > settings.maxParameters) {
- final String msg = String.format("The [%s] message contains %s bindings which is more than is allowed by the server %s configuration",
- Tokens.OPS_EVAL, bindings.size(), settings.maxParameters);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
- }
-
- // validations for specific op codes
- if (message.getOp().equals(Tokens.OPS_EVAL)) {
- // eval must have gremlin that is type of String
- // likely a problem with the driver and how it is sending requests
- if (!(message.optionalArgs(Tokens.ARGS_GREMLIN).get() instanceof String)) {
- final String msg = String.format("A message with [%s] op code requires a [%s] argument that is of type %s.",
- Tokens.OPS_EVAL, Tokens.ARGS_GREMLIN, String.class.getSimpleName());
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
- } else if (message.getOp().equals(Tokens.OPS_BYTECODE)) {
- // bytecode should have gremlin that is of type Bytecode
- // likely a problem with the driver and how it is sending requests
- if (!(message.optionalArgs(Tokens.ARGS_GREMLIN).get() instanceof Bytecode)) {
- final String msg = String.format("A message with [%s] op code requires a [%s] argument that is of type %s.",
- Tokens.OPS_BYTECODE, Tokens.ARGS_GREMLIN, Bytecode.class.getSimpleName());
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
-
- // bytecode should have an alias bound
- final Optional<Map<String, String>> aliases = message.optionalArgs(Tokens.ARGS_ALIASES);
- if (!aliases.isPresent()) {
- final String msg = String.format("A message with [%s] op code requires a [%s] argument.", Tokens.OPS_BYTECODE, Tokens.ARGS_ALIASES);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
-
- if (aliases.get().size() != 1 || !aliases.get().containsKey(Tokens.VAL_TRAVERSAL_SOURCE_ALIAS)) {
- final String msg = String.format("A message with [%s] op code requires the [%s] argument to be a Map containing one alias assignment named '%s'.",
- Tokens.OPS_BYTECODE, Tokens.ARGS_ALIASES, Tokens.VAL_TRAVERSAL_SOURCE_ALIAS);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
-
- final String traversalSourceBindingForAlias = aliases.get().values().iterator().next();
- if (!graphManager.getTraversalSourceNames().contains(traversalSourceBindingForAlias)) {
- final String msg = String.format("The traversal source [%s] for alias [%s] is not configured on the server.", traversalSourceBindingForAlias, Tokens.VAL_TRAVERSAL_SOURCE_ALIAS);
- throw new SessionException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
- }
- }
- }
-
- @Override
- public void userEventTriggered(final ChannelHandlerContext ctx, final Object evt) throws Exception {
- // only need to handle this event if the idle monitor is on
- if (!channelizer.supportsIdleMonitor()) return;
-
- if (evt instanceof IdleStateEvent) {
- final IdleStateEvent e = (IdleStateEvent) evt;
-
- // if no requests (reader) then close, if no writes from server to client then ping. clients should
- // periodically ping the server, but coming from this direction allows the server to kill channels that
- // have dead clients on the other end
- if (e.state() == IdleState.READER_IDLE) {
- logger.info("Closing channel - client is disconnected after idle period of " + settings.idleConnectionTimeout + " " + ctx.channel().id().asShortText());
- ctx.close();
- } else if (e.state() == IdleState.WRITER_IDLE && settings.keepAliveInterval > 0) {
- logger.info("Checking channel - sending ping to client after idle period of " + settings.keepAliveInterval + " " + ctx.channel().id().asShortText());
- ctx.writeAndFlush(channelizer.createIdleDetectionMessage());
- }
- }
- }
-
- /**
- * Called when creating a single task session where the provided {@link SessionTask} will be the only one to be
- * executed and can therefore take a more efficient execution path.
- */
- protected Session createSingleTaskSession(final SessionTask sessionTask, final String sessionId) {
- return new SingleTaskSession(sessionTask, sessionId, sessions);
- }
-
- /**
- * Called when creating a {@link Session} that will be long-lived to extend over multiple requests and therefore
- * process the provided {@link SessionTask} as well as ones that may arrive in the future.
- */
- protected Session createMultiTaskSession(final SessionTask sessionTask, final String sessionId) {
- return new MultiTaskSession(sessionTask, sessionId, sessions);
- }
-
- public boolean isActiveSession(final String sessionId) {
- return sessions.containsKey(sessionId);
- }
-
- public int getActiveSessionCount() {
- return sessions.size();
- }
-}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java
index b8038528d3..1c2b525b6b 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java
@@ -131,7 +131,6 @@ public class TraversalOpProcessor extends AbstractOpProcessor {
throw new OpProcessorException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
}
- // matches functionality in the UnifiedHandler
if (!(message.optionalArgs(Tokens.ARGS_GREMLIN).get() instanceof Bytecode)) {
final String msg = String.format("A message with [%s] op code requires a [%s] argument that is of type %s.",
Tokens.OPS_BYTECODE, Tokens.ARGS_GREMLIN, Bytecode.class.getSimpleName());
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/AbstractGremlinServerIntegrationTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/AbstractGremlinServerIntegrationTest.java
index 147014c138..a61c8efae0 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/AbstractGremlinServerIntegrationTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/AbstractGremlinServerIntegrationTest.java
@@ -18,8 +18,6 @@
*/
package org.apache.tinkerpop.gremlin.server;
-import org.apache.tinkerpop.gremlin.server.channel.UnifiedChannelizer;
-import org.apache.tinkerpop.gremlin.server.channel.UnifiedChannelizerIntegrateTest;
import org.apache.tinkerpop.gremlin.server.op.OpLoader;
import org.apache.tinkerpop.gremlin.server.util.ServerGremlinExecutor;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.AbstractTinkerGraph;
@@ -105,10 +103,6 @@ public abstract class AbstractGremlinServerIntegrationTest {
} else {
final Settings oSettings = overrideSettings(settings);
- if (shouldTestUnified()) {
- oSettings.channelizer = UnifiedChannelizer.class.getName();
- }
-
ServerTestHelper.rewritePathsInGremlinServerSettings(oSettings);
if (GREMLIN_SERVER_EPOLL) {
oSettings.useEpollEventLoop = true;
@@ -131,10 +125,6 @@ public abstract class AbstractGremlinServerIntegrationTest {
overriddenSettings.useEpollEventLoop = true;
}
- if (shouldTestUnified()) {
- overriddenSettings.channelizer = UnifiedChannelizer.class.getName();
- }
-
this.server = new GremlinServer(overriddenSettings);
return server.start();
@@ -161,11 +151,6 @@ public abstract class AbstractGremlinServerIntegrationTest {
OpLoader.reset();
}
- protected boolean isUsingUnifiedChannelizer() {
- return server.getServerGremlinExecutor().
- getSettings().channelizer.equals(UnifiedChannelizer.class.getName());
- }
-
public static boolean deleteDirectory(final File directory) {
if (directory.exists()) {
final File[] files = directory.listFiles();
@@ -187,11 +172,4 @@ public abstract class AbstractGremlinServerIntegrationTest {
logger.info("Running transactional tests using TinkerTransactionGraph");
settings.graphs.put("graph", "conf/tinkertransactiongraph-empty.properties");
}
-
- private boolean shouldTestUnified() {
- // ignore all tests in the UnifiedChannelizerIntegrateTest package as they are already rigged to test
- // over the various channelizer implementations
- return Boolean.parseBoolean(System.getProperty("testUnified", "false")) &&
- !this.getClass().getPackage().equals(UnifiedChannelizerIntegrateTest.class.getPackage());
- }
}
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 8b1b8da59f..1605732331 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
@@ -1586,7 +1586,7 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
try {
// this configures the client to behave like OpProcessor for UnifiedChannelizer
final Client.SessionSettings settings = Client.SessionSettings.build().
- sessionId(name.getMethodName()).maintainStateAfterException(true).create();
+ sessionId(name.getMethodName()).create();
final Client client = cluster.connect(Client.Settings.build().useSession(settings).create());
for (int index = 0; index < 50; index++) {
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpIntegrateTest.java
index e12859b82d..e7487475ca 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpIntegrateTest.java
@@ -1063,7 +1063,7 @@ public class GremlinServerHttpIntegrateTest extends AbstractGremlinServerIntegra
@Test(timeout = 10000) // Add test timeout to prevent incorrect timeout behavior from stopping test run.
public void should500OnGETWithEvaluationTimeout() throws Exception {
- // Related to TINKERPOP-2769. This is a similar test to those for WebSocketChannelizer and UnifiedChannelizer.
+ // Related to TINKERPOP-2769. This is a similar test to the one for the WebSocketChannelizer.
final CloseableHttpClient firstClient = HttpClients.createDefault();
final CloseableHttpClient secondClient = HttpClients.createDefault();
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 01121310e9..639d1b8055 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
@@ -27,8 +27,6 @@ import org.apache.commons.configuration2.BaseConfiguration;
import org.apache.commons.configuration2.Configuration;
import org.apache.tinkerpop.gremlin.server.channel.HttpTestChannelizer;
import org.apache.tinkerpop.gremlin.server.channel.TestChannelizer;
-import org.apache.tinkerpop.gremlin.server.channel.UnifiedChannelizer;
-import org.apache.tinkerpop.gremlin.server.channel.UnifiedTestChannelizer;
import org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer;
import org.apache.tinkerpop.gremlin.server.channel.WebSocketTestChannelizer;
import org.apache.tinkerpop.gremlin.server.channel.WsAndHttpTestChannelizer;
@@ -52,7 +50,6 @@ import org.apache.tinkerpop.gremlin.groovy.jsr223.GroovyCompilerGremlinPlugin;
import org.apache.tinkerpop.gremlin.groovy.jsr223.customizer.SimpleSandboxExtension;
import org.apache.tinkerpop.gremlin.jsr223.ScriptFileGremlinPlugin;
import org.apache.tinkerpop.gremlin.server.handler.OpSelectorHandler;
-import org.apache.tinkerpop.gremlin.server.handler.UnifiedHandler;
import org.apache.tinkerpop.gremlin.structure.RemoteGraph;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
@@ -150,11 +147,9 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
name.getMethodName().equals("shouldCloseChannelIfClientDoesntRespond") ||
name.getMethodName().equals("shouldCaptureUserAgentFromClient")) {
final Logger opSelectorHandlerLogger = (Logger) LoggerFactory.getLogger(OpSelectorHandler.class);
- final Logger unifiedHandlerLogger = (Logger) LoggerFactory.getLogger(UnifiedHandler.class);
final Logger wsUserAgentHandlerLogger = (Logger) LoggerFactory.getLogger(WsUserAgentHandler.class);
previousLogLevel = opSelectorHandlerLogger.getLevel();
opSelectorHandlerLogger.setLevel(Level.INFO);
- unifiedHandlerLogger.setLevel(Level.INFO);
wsUserAgentHandlerLogger.setLevel(Level.DEBUG);
}
@@ -168,8 +163,6 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
name.getMethodName().equals("shouldCaptureUserAgentFromClient")) {
final Logger opSelectorHandlerLogger = (Logger) LoggerFactory.getLogger(OpSelectorHandler.class);
opSelectorHandlerLogger.setLevel(previousLogLevel);
- final Logger unifiedHandlerLogger = (Logger) LoggerFactory.getLogger(UnifiedHandler.class);
- unifiedHandlerLogger.setLevel(previousLogLevel);
final Logger wsUserAgentHandlerLogger = (Logger) LoggerFactory.getLogger(WsUserAgentHandler.class);
wsUserAgentHandlerLogger.setLevel(previousLogLevel);
}
@@ -193,9 +186,6 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
}};
settings.processors.clear();
settings.processors.add(processorSettingsBig);
-
- // Unified setting
- settings.maxParameters = Integer.MAX_VALUE;
break;
case "shouldRespectHighWaterMarkSettingAndSucceed":
settings.writeBufferHighWaterMark = 64;
@@ -235,9 +225,6 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
}};
settings.processors.clear();
settings.processors.add(processorSettingsSmall);
-
- // Unified settings
- settings.maxParameters = 1;
break;
case "shouldTimeOutRemoteTraversal":
settings.evaluationTimeout = 500;
@@ -259,18 +246,6 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
settings.gremlinPool = POOL_SIZE_FOR_TIMEOUT_TESTS;
settings.channelizer = WebSocketChannelizer.class.getName();
break;
- case "shouldRespondToTimeoutCancelledSingleTaskUnifiedRequest":
- settings.evaluationTimeout = 5000;
- settings.gremlinPool = POOL_SIZE_FOR_TIMEOUT_TESTS;
- settings.channelizer = UnifiedChannelizer.class.getName();
- break;
- case "shouldRespondToTimeoutCancelledMultiTaskUnifiedRequest":
- useTinkerTransactionGraph(settings);
- settings.evaluationTimeout = 30000;
- settings.sessionLifetimeTimeout = 5000; // This needs to be shorter because of the delay in scheduling session task.
- settings.gremlinPool = POOL_SIZE_FOR_TIMEOUT_TESTS;
- settings.channelizer = UnifiedChannelizer.class.getName();
- break;
case "shouldStoreUserAgentInContextWebSocket":
settings.channelizer = WebSocketTestChannelizer.class.getName();
break;
@@ -280,9 +255,6 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
case "shouldStoreUserAgentInContextWsAndHttp":
settings.channelizer = WsAndHttpTestChannelizer.class.getName();
break;
- case "shouldStoreUserAgentInContextUnified":
- settings.channelizer = UnifiedTestChannelizer.class.getName();
- break;
default:
break;
}
@@ -345,12 +317,6 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
shouldStoreUserAgentInHttpContext();
}
- @Test
- public void shouldStoreUserAgentInContextUnified() throws InterruptedException {
- shouldStoreUserAgentInContext();
- shouldStoreUserAgentInHttpContext();
- }
-
private void shouldStoreUserAgentInContext() {
if(server.getChannelizer() instanceof TestChannelizer) {
TestChannelizer channelizer = (TestChannelizer) server.getChannelizer();
@@ -576,9 +542,6 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
@Test
public void shouldProduceProperExceptionOnTimeout() throws Exception {
- // this test will not work quite right on UnifiedChannelizer
- assumeThat("Must use OpProcessor", isUsingUnifiedChannelizer(), is(false));
-
final Cluster cluster = TestClientFactory.open();
final Client client = cluster.connect(name.getMethodName());
@@ -881,9 +844,6 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
@Test
@SuppressWarnings("unchecked")
public void shouldBatchResultsByTwos() throws Exception {
- // this test will not work quite right on UnifiedChannelizer, but seems to only be a problem in Travis
- assumeThat("Must use OpProcessor", isUsingUnifiedChannelizer(), is(false));
-
try (SimpleClient client = TestClientFactory.createWebSocketClient()) {
final RequestMessage request = RequestMessage.build(Tokens.OPS_EVAL)
.addArg(Tokens.ARGS_GREMLIN, "[0,1,2,3,4,5,6,7,8,9]").create();
@@ -1244,10 +1204,6 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
*/
@Test(timeout = 180000) // Add timeout in case the test hangs.
public void shouldRespondToTimeoutCancelledSessionRequest() throws Exception {
- // Don't test with UnifiedChannelizer since we only want to test the case where a task is cancelled before
- // running which is handled by shouldRespondToTimeoutCancelledMultiTaskUnifiedRequest.
- assumeThat("Must use OpProcessor", isUsingUnifiedChannelizer(), is(false));
-
final Cluster cluster = TestClientFactory.build().create();
final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
final GraphTraversalSource gtx = g.tx().begin();
@@ -1258,57 +1214,6 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
cluster.close();
}
- /**
- * Reproducer for TINKERPOP-2769 with request sent to UnifiedChannelizer.
- */
- @Test(timeout = 180000) // Add timeout in case the test hangs.
- public void shouldRespondToTimeoutCancelledSingleTaskUnifiedRequest() throws Exception {
- final GraphTraversalSource g = traversal().withRemote(conf);
- runTimeoutTest(g);
- g.close();
- }
-
- /**
- * Reproducer for TINKERPOP-2769 with request having a Session ID sent to UnifiedChannelizer.
- */
- @Test(timeout = 180000) // Add timeout in case the test hangs.
- public void shouldRespondToTimeoutCancelledMultiTaskUnifiedRequest() throws Exception {
- final Cluster cluster = TestClientFactory.build().create();
- final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
- final GraphTraversalSource gtx = g.tx().begin();
-
- g.addV("person").as("p").addE("self").to("p").iterate();
-
- // Number of threads/tasks must be larger than the size of the gremlinPool set in overrideSettings().
- final ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(2);
- // Use a request without session to fill the queue since we want to test cancel on first session request.
- final Future<?> firstResult = threadPool.submit(() -> g.V().repeat(__.out()).until(__.outE().count().is(0)).iterate());
- // Delay this request slightly as we want to test that the MultiTaskSession is properly returning an error when it is cancelled.
- final Future<?> secondResult = threadPool.schedule(() -> gtx.V().repeat(__.out()).until(__.outE().count().is(0)).iterate(), 5000, TimeUnit.MILLISECONDS);
-
- try {
- firstResult.get();
- fail("This traversal should have timed out");
- } catch (Exception ex) {
- final Throwable t = ex.getCause().getCause(); // Get the nested ResponseException.
- assertThat(t, instanceOf(ResponseException.class));
- assertEquals(ResponseStatusCode.SERVER_ERROR_TIMEOUT, ((ResponseException) t).getResponseStatusCode());
- }
-
- try {
- secondResult.get();
- fail("This traversal should have timed out");
- } catch (Exception ex) {
- final Throwable t = ex.getCause().getCause(); // Get the nested ResponseException.
- assertThat(t, instanceOf(ResponseException.class));
- assertEquals(ResponseStatusCode.SERVER_ERROR_TIMEOUT, ((ResponseException) t).getResponseStatusCode());
- }
-
- threadPool.shutdown();
- gtx.tx().rollback();
- cluster.close();
- }
-
private void runTimeoutTest(GraphTraversalSource g) throws Exception {
// make a graph with a cycle in it to force a long run traversal
g.addV("person").as("p").addE("self").to("p").iterate();
@@ -1337,7 +1242,7 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
}
/**
- * Reproducer for TINKERPOP-2765 when run using the UnifiedChannelizer.
+ * Reproducer for TINKERPOP-2765.
*/
@Test
public void shouldHandleMultipleLambdaTranslationsInParallel() throws Exception {
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 2b6c25f5bb..02c12eb7b7 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
@@ -34,7 +34,6 @@ import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
import org.apache.tinkerpop.gremlin.driver.simple.SimpleClient;
-import org.apache.tinkerpop.gremlin.server.channel.UnifiedChannelizer;
import org.apache.tinkerpop.gremlin.server.op.session.Session;
import org.apache.tinkerpop.gremlin.server.op.session.SessionOpProcessor;
import org.junit.After;
@@ -119,29 +118,15 @@ public class GremlinServerSessionIntegrateTest extends AbstractGremlinServerInte
processorSettings.config = new HashMap<>();
processorSettings.config.put(SessionOpProcessor.CONFIG_SESSION_TIMEOUT, 3000L);
settings.processors.add(processorSettings);
-
- // Unified setting
- settings.sessionLifetimeTimeout = 3000L;
break;
- case "shouldBlowTheSessionQueueSize":
- settings.maxSessionTaskQueueSize = 1;
case "shouldCloseSessionOnClientClose":
- case "shouldCloseSessionOnClientCloseWithStateMaintainedBetweenExceptions":
case "shouldExecuteInSessionAndSessionlessWithoutOpeningTransactionWithSingleClient":
case "shouldExecuteInSessionWithTransactionManagement":
case "shouldRollbackOnEvalExceptionForManagedTransaction":
- case "shouldNotExecuteQueuedRequestsIfOneInFrontOfItFails":
useTinkerTransactionGraph(settings);
break;
case "shouldEnsureSessionBindingsAreThreadSafe":
settings.threadPoolWorker = 2;
- break;
- case "shouldUseGlobalFunctionCache":
- // OpProcessor settings are good by default
- // UnifiedHandler settings
- settings.useCommonEngineForSessions = false;
- settings.useGlobalFunctionCacheForSessions = true;
-
break;
case "shouldNotUseGlobalFunctionCache":
settings.processors.clear();
@@ -153,108 +138,12 @@ public class GremlinServerSessionIntegrateTest extends AbstractGremlinServerInte
processorSettingsForDisableFunctionCache.config.put(SessionOpProcessor.CONFIG_GLOBAL_FUNCTION_CACHE_ENABLED, false);
settings.processors.add(processorSettingsForDisableFunctionCache);
- // UnifiedHandler settings
- settings.useCommonEngineForSessions = false;
- settings.useGlobalFunctionCacheForSessions = false;
-
break;
}
return settings;
}
- @Test
- public void shouldBlowTheSessionQueueSize() throws Exception {
- assumeThat(isUsingUnifiedChannelizer(), is(true));
-
- final Cluster cluster = TestClientFactory.open();
- final Client session = cluster.connect(name.getMethodName());
-
- // maxSessionQueueSize=1
- // we should be able to do one request at a time serially
- assertEquals("test1", session.submit("'test1'").all().get().get(0).getString());
- assertEquals("test2", session.submit("'test2'").all().get().get(0).getString());
- assertEquals("test3", session.submit("'test3'").all().get().get(0).getString());
-
- final AtomicBoolean errorTriggered = new AtomicBoolean();
- final ResultSet r1 = session.submitAsync("Thread.sleep(1000);'test4'").get();
-
- final List<CompletableFuture<List<Result>>> blockers = new ArrayList<>();
- for (int ix = 0; ix < 512 && !errorTriggered.get(); ix++) {
- blockers.add(session.submit("'test'").all().exceptionally(t -> {
- final ResponseException re = (ResponseException) t.getCause();
- errorTriggered.compareAndSet(false, ResponseStatusCode.TOO_MANY_REQUESTS == re.getResponseStatusCode());
- return null;
- }));
-
- // low resource environments like travis might need a break
- if (ix % 32 == 0) Thread.sleep(500);
- }
-
- // wait for the blockage to clear for sure
- assertEquals("test4", r1.all().get().get(0).getString());
- blockers.forEach(CompletableFuture::join);
-
- assertThat(errorTriggered.get(), is(true));
-
- // should be accepting test6 now
- assertEquals("test6", session.submit("'test6'").all().get().get(0).getString());
-
- session.close();
- cluster.close();
- }
-
- @Test
- public void shouldNotExecuteQueuedRequestsIfOneInFrontOfItFails() throws Exception {
- assumeThat(isUsingUnifiedChannelizer(), is(true));
-
- final Cluster cluster = TestClientFactory.open();
- final Client session = cluster.connect(name.getMethodName());
- final List<CompletableFuture<ResultSet>> futuresAfterFailure = new ArrayList<>();
-
- // short, sweet and successful - but will rollback after close
- final CompletableFuture<List<Result>> first = session.submit("g.addV('person').iterate();1").all();
- assertEquals(1, first.get().get(0).getInt());
-
- // leave enough time for the results to queue behind this
- final int followOnRequestCount = 8;
- final CompletableFuture<ResultSet> second = session.submitAsync("g.addV('person').sideEffect{Thread.sleep(60000)}");
-
- // these requests will queue
- for (int ix = 0; ix < followOnRequestCount; ix ++) {
- futuresAfterFailure.add(session.submitAsync("g.addV('software')"));
- }
-
- // close the session which will interrupt "second" request and flush the queue of requests behind it
- session.close();
-
- try {
- second.join().all().join();
- fail("should have been interrupted by channel close");
- } catch (Exception ex) {
- assertEquals("Connection to server is no longer active", ex.getCause().getMessage());
- }
-
- final AtomicInteger fails = new AtomicInteger(0);
- futuresAfterFailure.forEach(f -> {
- try {
- f.join().all().join();
- fail("should have been interrupted by channel close");
- } catch (Exception ex) {
- assertEquals("Connection to server is no longer active", ex.getCause().getMessage());
- fails.incrementAndGet();
- }
- });
-
- // all are failures
- assertEquals(followOnRequestCount, fails.get());
-
- // validate the rollback
- final Client client = cluster.connect();
- assertEquals(0, client.submit("g.V().count()").all().get().get(0).getInt());
- cluster.close();
- }
-
@Test
public void shouldCloseSessionOnClientClose() throws Exception {
@@ -266,12 +155,8 @@ public class GremlinServerSessionIntegrateTest extends AbstractGremlinServerInte
cluster1.close();
// the following session close log message is no longer relevant as
- if (isUsingUnifiedChannelizer()) {
- assertThat(((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().isActiveSession(name.getMethodName()), is(false));
- } else {
- assertThat(logCaptor.getLogs(), hasItem("Skipped attempt to close open graph transactions on shouldCloseSessionOnClientClose - close was forced"));
- assertThat(logCaptor.getLogs(), hasItem("Session shouldCloseSessionOnClientClose closed"));
- }
+ assertThat(logCaptor.getLogs(), hasItem("Skipped attempt to close open graph transactions on shouldCloseSessionOnClientClose - close was forced"));
+ assertThat(logCaptor.getLogs(), hasItem("Session shouldCloseSessionOnClientClose closed"));
// try to reconnect to that session and make sure no state is there
final Cluster clusterReconnect = TestClientFactory.open();
@@ -289,59 +174,9 @@ public class GremlinServerSessionIntegrateTest extends AbstractGremlinServerInte
// the commit from client1 should not have gone through so there should be no data present.
assertEquals(0, clientReconnect.submit("graph.traversal().V().count()").all().join().get(0).getInt());
-
- // must turn on maintainStateAfterException for unified channelizer
- if (!isUsingUnifiedChannelizer()) {
- assertEquals(100, clientReconnect.submit("y").all().join().get(0).getInt());
- }
-
- clusterReconnect.close();
-
- if (isUsingUnifiedChannelizer()) {
- assertEquals(0, ((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().getActiveSessionCount());
- }
- }
-
- @Test
- public void shouldCloseSessionOnClientCloseWithStateMaintainedBetweenExceptions() throws Exception {
- assumeThat("Must use UnifiedChannelizer", isUsingUnifiedChannelizer(), is(true));
-
- final Cluster cluster1 = TestClientFactory.open();
- final Client client1 = cluster1.connect(name.getMethodName());
- client1.submit("x = 1").all().join();
- client1.submit("graph.addVertex()").all().join();
- client1.close();
- cluster1.close();
-
- assertThat(((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().isActiveSession(name.getMethodName()), is(false));
-
- // try to reconnect to that session and make sure no state is there
- final Cluster clusterReconnect = TestClientFactory.open();
-
- // this configures the client to behave like OpProcessor for UnifiedChannelizer
- final Client.SessionSettings settings = Client.SessionSettings.build().
- sessionId(name.getMethodName()).maintainStateAfterException(true).create();
- final Client clientReconnect = clusterReconnect.connect(Client.Settings.build().useSession(settings).create());
-
- // should get an error because "x" is not defined as this is a new session
- try {
- clientReconnect.submit("y=100").all().join();
- clientReconnect.submit("x").all().join();
- fail("Should not have been successful as 'x' was only defined in the old session");
- } catch(Exception ex) {
- final Throwable root = ExceptionHelper.getRootCause(ex);
- assertThat(root.getMessage(), startsWith("No such property"));
- }
-
- // the commit from client1 should not have gone through so there should be no data present.
- assertEquals(0, clientReconnect.submit("graph.traversal().V().count()").all().join().get(0).getInt());
-
- // since maintainStateAfterException is enabled the UnifiedChannelizer works like OpProcessor
assertEquals(100, clientReconnect.submit("y").all().join().get(0).getInt());
clusterReconnect.close();
-
- assertEquals(0, ((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().getActiveSessionCount());
}
@Test
@@ -417,7 +252,6 @@ public class GremlinServerSessionIntegrateTest extends AbstractGremlinServerInte
final Client.SessionSettings sessionSettings = Client.SessionSettings.build().
sessionId(name.getMethodName()).
manageTransactions(true).
- maintainStateAfterException(false).
create();
final Client.Settings clientSettings = Client.Settings.build().useSession(sessionSettings).create();
final Client client = cluster.connect(clientSettings);
@@ -471,19 +305,15 @@ public class GremlinServerSessionIntegrateTest extends AbstractGremlinServerInte
cluster.close();
}
- if (isUsingUnifiedChannelizer()) {
- assertThat(((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().isActiveSession(name.getMethodName()), is(false));
- } else {
- assertEquals(1, logCaptor.getLogs().stream()
- .filter(msg -> msg.equals("Session shouldCloseSessionOnceOnRequest closed")).count());
- }
+ assertEquals(1, logCaptor.getLogs().stream()
+ .filter(msg -> msg.equals("Session shouldCloseSessionOnceOnRequest closed")).count());
}
@Test
public void shouldHaveTheSessionTimeout() throws Exception {
final Cluster cluster = TestClientFactory.open();
final Client.SessionSettings settings = Client.SessionSettings.build().
- sessionId(name.getMethodName()).maintainStateAfterException(true).create();
+ sessionId(name.getMethodName()).create();
final Client session = cluster.connect(Client.Settings.build().useSession(settings).create());
final ResultSet results1 = session.submit("x = [1,2,3,4,5,6,7,8,9]");
@@ -495,18 +325,7 @@ public class GremlinServerSessionIntegrateTest extends AbstractGremlinServerInte
// session times out in 3 seconds but slow environments like travis might not be so exacting in the
// shutdown process so be patient
- if (isUsingUnifiedChannelizer()) {
- boolean sessionAlive = true;
- for (int ix = 1; ix < 11; ix++) {
- sessionAlive = ((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().isActiveSession(name.getMethodName());
- if (!sessionAlive) break;
- Thread.sleep(ix * 500);
- }
-
- assertThat(sessionAlive, is(false));
- } else {
- Thread.sleep(5000);
- }
+ Thread.sleep(5000);
try {
// the original session should be dead so this call will open a new session with the same name but fail
@@ -524,13 +343,9 @@ public class GremlinServerSessionIntegrateTest extends AbstractGremlinServerInte
cluster.close();
}
- if (isUsingUnifiedChannelizer()) {
- assertThat(((UnifiedChannelizer) server.getChannelizer()).getUnifiedHandler().isActiveSession(name.getMethodName()), is(false));
- } else {
- // there will be one for the timeout and a second for closing the cluster
- assertEquals(2, logCaptor.getLogs().stream()
- .filter(msg -> msg.equals("Session shouldHaveTheSessionTimeout closed")).count());
- }
+ // there will be one for the timeout and a second for closing the cluster
+ assertEquals(2, logCaptor.getLogs().stream()
+ .filter(msg -> msg.equals("Session shouldHaveTheSessionTimeout closed")).count());
}
@Test
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSessionIntegrateWithUnifiedChannelizerTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSessionIntegrateWithUnifiedChannelizerTest.java
deleted file mode 100644
index bf51bc9324..0000000000
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerSessionIntegrateWithUnifiedChannelizerTest.java
+++ /dev/null
@@ -1,61 +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.server;
-
-import org.apache.tinkerpop.gremlin.driver.Cluster;
-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.UnifiedChannelizerIntegrateTest;
-import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
-import org.junit.Test;
-
-import static org.apache.tinkerpop.gremlin.process.traversal.AnonymousTraversalSource.traversal;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
-public class GremlinServerSessionIntegrateWithUnifiedChannelizerTest extends UnifiedChannelizerIntegrateTest {
-
- /**
- * Reproducer for TINKERPOP-2751
- */
- @Test(timeout=30000)
- public void shouldThrowExceptionOnTransactionUnsupportedGraph() throws Exception {
- final Cluster cluster = TestClientFactory.build().create();
- final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
-
- final GraphTraversalSource gtx = g.tx().begin();
- assertThat(gtx.tx().isOpen(), is(true));
-
- gtx.addV("person").iterate();
- assertEquals(1, (long) gtx.V().count().next());
-
- try {
- // Without Neo4j plugin this should fail on gremlin-server
- gtx.tx().commit();
- fail("commit should throw exception on non-transaction supported graph");
- } catch (Exception ex){
- final Throwable root = ExceptionHelper.getRootCause(ex);
- assertEquals("Graph does not support transactions", root.getMessage());
- }
-
- cluster.close();
- }
-}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinSessionTxIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinSessionTxIntegrateTest.java
index cf22cf9156..22ae2e8934 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinSessionTxIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinSessionTxIntegrateTest.java
@@ -71,8 +71,6 @@ public class GremlinSessionTxIntegrateTest extends AbstractGremlinServerIntegrat
processorSettings.config.put(SessionOpProcessor.CONFIG_SESSION_TIMEOUT, 3000L);
settings.processors.add(processorSettings);
- // Unified setting
- settings.sessionLifetimeTimeout = 3000L;
break;
}
@@ -260,7 +258,7 @@ public class GremlinSessionTxIntegrateTest extends AbstractGremlinServerIntegrat
// need to open significantly more sessions that we have threads in gremlinPool. if we go too obscene on
// OpProcessor this test will take too long
- final int numberOfSessions = isUsingUnifiedChannelizer() ? 1000 : 100;
+ final int numberOfSessions = 100;
for (int ix = 0; ix < numberOfSessions; ix ++) {
final Transaction tx = g.tx();
final GraphTraversalSource gtx = tx.begin();
@@ -370,7 +368,6 @@ public class GremlinSessionTxIntegrateTest extends AbstractGremlinServerIntegrat
final Client.SessionSettings sessionSettings = Client.SessionSettings.build().
sessionId(name.getMethodName()).
manageTransactions(false).
- maintainStateAfterException(false).
create();
final Client.Settings clientSettings = Client.Settings.build().useSession(sessionSettings).create();
final Client client = cluster.connect();
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/channel/UnifiedChannelizerIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/channel/UnifiedChannelizerIntegrateTest.java
deleted file mode 100644
index a090a2036a..0000000000
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/channel/UnifiedChannelizerIntegrateTest.java
+++ /dev/null
@@ -1,57 +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.server.channel;
-
-import org.apache.tinkerpop.gremlin.server.Settings;
-import org.apache.tinkerpop.gremlin.server.auth.SimpleAuthenticator;
-import org.apache.tinkerpop.gremlin.server.handler.SaslAndHttpBasicAuthenticationHandler;
-
-import java.util.HashMap;
-import java.util.Map;
-
-public class UnifiedChannelizerIntegrateTest extends AbstractGremlinServerChannelizerIntegrateTest {
-
- @Override
- public String getProtocol() {
- return WS_AND_HTTP;
- }
-
- @Override
- public String getSecureProtocol() {
- return WSS_AND_HTTPS;
- }
-
- @Override
- public String getChannelizer() {
- return UnifiedChannelizer.class.getName();
- }
-
- @Override
- public Settings.AuthenticationSettings getAuthSettings() {
- final Settings.AuthenticationSettings authSettings = new Settings.AuthenticationSettings();
- final Map<String,Object> authConfig = new HashMap<>();
- authSettings.authenticator = SimpleAuthenticator.class.getName();
- authSettings.authenticationHandler = SaslAndHttpBasicAuthenticationHandler.class.getName();
- authConfig.put(SimpleAuthenticator.CONFIG_CREDENTIALS_DB, "conf/tinkergraph-credentials.properties");
- authSettings.config = authConfig;
-
- return authSettings;
- }
-
-}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/channel/UnifiedTestChannelizer.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/channel/UnifiedTestChannelizer.java
deleted file mode 100644
index 149ecd0bb4..0000000000
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/channel/UnifiedTestChannelizer.java
+++ /dev/null
@@ -1,48 +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.server.channel;
-
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelPipeline;
-
-/**
- * A wrapper around UnifiedChannelizer which saves and exposes the ChannelHandlerContext for testing purposes
- */
-public class UnifiedTestChannelizer extends UnifiedChannelizer implements TestChannelizer {
-
- final ContextHandler contextHandler;
-
- public UnifiedTestChannelizer() {
- contextHandler = new ContextHandler();
- }
-
- @Override
- public void configure(final ChannelPipeline pipeline) {
- super.configure(pipeline);
- pipeline.addFirst(contextHandler);
- }
-
- public ChannelHandlerContext getMostRecentChannelHandlerContext() {
- return contextHandler.getMostRecentChannelHandlerContext();
- }
-
- public void resetChannelHandlerContext() {
- contextHandler.resetChannelHandlerContext();
- }
-}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/channel/WsAndHttpTestChannelizer.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/channel/WsAndHttpTestChannelizer.java
index ca2115874d..ad94fa8247 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/channel/WsAndHttpTestChannelizer.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/channel/WsAndHttpTestChannelizer.java
@@ -22,7 +22,7 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
/**
- * A wrapper around UnifiedChannelizer which saves and exposes the ChannelHandlerContext for testing purposes
+ * A wrapper around WsAndHttpTestChannelizer which saves and exposes the ChannelHandlerContext for testing purposes
*/
public class WsAndHttpTestChannelizer extends WsAndHttpChannelizer implements TestChannelizer {
diff --git a/gremlin-server/src/test/resources/logback-test.xml b/gremlin-server/src/test/resources/logback-test.xml
index b1154279dd..1cc03f6a0f 100644
--- a/gremlin-server/src/test/resources/logback-test.xml
+++ b/gremlin-server/src/test/resources/logback-test.xml
@@ -20,12 +20,10 @@ limitations under the License.
<pattern>[%p] %C - %m%n</pattern>
</encoder>
</appender>
- <logger name="org.apache.tinkerpop.gremlin.server.handler.MultiTaskSession" level="ERROR"/>
<logger name="org.apache.tinkerpop.gremlin.server.op.AbstractEvalOpProcessor" level="ERROR"/>
<logger name="org.apache.tinkerpop.gremlin.server.op.traversal.TraversalOpProcessor" level="ERROR"/>
<logger name="org.apache.tinkerpop.gremlin.server.AbstractGremlinServerIntegrationTest" level="INFO"/>
<logger name="audit.org.apache.tinkerpop.gremlin.server" level="INFO"/>
- <logger name="org.apache.tinkerpop.gremlin.server.handler.SingleTaskSession" level="ERROR"/>
<logger name="org.apache.tinkerpop.gremlin.server.AbstractChannelizer" level="ERROR"/>
<!-- this logger is noisy and we don't assert anything and the error is already tracked on the server so we can
trim the logs a bit with this. -->
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/Tokens.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/Tokens.java
index 63918524ad..8d3e96c1df 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/Tokens.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/Tokens.java
@@ -95,13 +95,6 @@ public final class Tokens {
* Should be used with {@code ARGS_MATERIALIZE_PROPERTIES}
*/
public static final String MATERIALIZE_PROPERTIES_TOKENS = "tokens";
-
- /**
- * Argument name that is intended to be used with a session which when its value is {@code true} makes it so
- * that a processing error or request timeout will not close the session, but leave it to continue processing in
- * whatever state it may hold. This argument only applies to the {@code UnifiedChannelizer}.
- */
- public static final String ARGS_MAINTAIN_STATE_AFTER_EXCEPTION = "maintainStateAfterException";
public static final String ARGS_SASL = "sasl";
public static final String ARGS_SASL_MECHANISM = "saslMechanism";