You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2015/03/06 18:19:55 UTC
incubator-tinkerpop git commit: Gremlin Server was generating extra
response messages on script evaluation errors.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master a7f941ed5 -> 0ef385747
Gremlin Server was generating extra response messages on script evaluation errors.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/0ef38574
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/0ef38574
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/0ef38574
Branch: refs/heads/master
Commit: 0ef38574709e8c0f6a0b2cc6c95a80109236fc57
Parents: a7f941e
Author: Stephen Mallette <sp...@apache.org>
Authored: Fri Mar 6 12:11:23 2015 -0500
Committer: Stephen Mallette <sp...@apache.org>
Committed: Fri Mar 6 12:11:23 2015 -0500
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../server/op/AbstractEvalOpProcessor.java | 24 +++++---
.../server/GremlinServerIntegrateTest.java | 63 ++++++++++++++++++++
3 files changed, 80 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/0ef38574/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index f99718b..9f835ad 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,7 @@ TinkerPop 3.0.0.M8 (Release Date: NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
* Added `compile` method to `GremlinExecutor` and related classes.
+* Fixed bug in Gremlin Server that was generating extra response messages on script evaluation errors.
* Changed the `Memory` API to not return the mutated value on `or`, `and`, `incr` as it is too difficult to implement faithfully in a distributed system.
* Added `SparkGraphComputer` to Hadoop-Gremlin which uses Apache Spark as the underlying computing engine.
* Renamed "Gremlin Kryo" to "Gryo".
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/0ef38574/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
index a408589..e6bc607 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
@@ -19,6 +19,7 @@
package org.apache.tinkerpop.gremlin.server.op;
import com.codahale.metrics.Timer;
+import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.tinkerpop.gremlin.driver.Tokens;
import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
import org.apache.tinkerpop.gremlin.driver.message.ResponseMessage;
@@ -152,7 +153,7 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
final CompletableFuture<Void> iterationFuture = evalFuture.thenAcceptAsync(o -> {
final Iterator itty = IteratorUtils.convertToIterator(o);
- logger.debug("Preparing to iterate results from - {} - in thread [{}]", msg, Thread.currentThread().getName());
+ logger.info("Preparing to iterate results from - {} - in thread [{}]", msg, Thread.currentThread().getName());
try {
handleIterator(context, itty);
@@ -162,15 +163,22 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
}, executor);
iterationFuture.handleAsync((r, ex) -> {
- // iteration has completed - if there was an exception then write it
+ // iteration has completed
if (ex != null) {
- final String errorMessage = String.format("Response iteration and serialization exceeded the configured threshold for request [%s] - %s", msg, ex.getCause().getMessage());
- logger.warn(errorMessage);
- ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT).statusMessage(errorMessage).create());
+ // an exception could have fired in the evalFuture - it interestingly bubbles up here so we have to
+ // check for the iteration exception (which is a TimeoutException). otherwise the exception would
+ // have been dealt with from a messaging perspective as part of the "exceptionally" completion stage
+ // of the evalFuture itself.
+ if (ExceptionUtils.getRootCause(ex).getClass().equals(TimeoutException.class)) {
+ final String errorMessage = String.format("Response iteration and serialization exceeded the configured threshold for request [%s] - %s", msg, ex.getCause().getMessage());
+ logger.warn(errorMessage);
+ ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT).statusMessage(errorMessage).create());
+ }
+ } else {
+ // since this is not an error we need to terminate. termination for errors is handled in the
+ // ResponseEncoder
+ ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SUCCESS_TERMINATOR).create());
}
-
- // either way - terminate the request
- ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SUCCESS_TERMINATOR).create());
return null;
}, executor);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/0ef38574/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
index 379b44e..3cb59de 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
@@ -24,6 +24,7 @@ import org.apache.tinkerpop.gremlin.driver.ResultSet;
import org.apache.tinkerpop.gremlin.driver.Tokens;
import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
+import org.apache.tinkerpop.gremlin.driver.message.ResponseStatus;
import org.apache.tinkerpop.gremlin.driver.message.ResponseStatusCode;
import org.apache.tinkerpop.gremlin.driver.ser.Serializers;
import org.apache.tinkerpop.gremlin.driver.simple.NioClient;
@@ -342,6 +343,68 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
cluster.close();
}
+ @Test
+ public void shouldEvalAndOnlyReturnTwoMessages() throws Exception {
+ try (SimpleClient client = new WebSocketClient()) {
+ final RequestMessage request = RequestMessage.build(Tokens.OPS_EVAL)
+ .addArg(Tokens.ARGS_GREMLIN, "10").create();
+
+ // set the latch to two as there should be two responses when you include the terminator -
+ // the error and the terminator
+ final CountDownLatch latch = new CountDownLatch(2);
+ final AtomicInteger messages = new AtomicInteger(0);
+ final AtomicBoolean errorReceived = new AtomicBoolean(false);
+ final AtomicBoolean terminatorReceived = new AtomicBoolean(false);
+ client.submit(request, r -> {
+ if (latch.getCount() == 2)
+ errorReceived.set(r.getStatus().equals(ResponseStatusCode.SUCCESS));
+ else if (latch.getCount() == 1)
+ terminatorReceived.set(r.getStatus().equals(ResponseStatusCode.SUCCESS_TERMINATOR));
+ latch.countDown();
+ messages.incrementAndGet();
+ });
+
+ assertTrue(latch.await(1500, TimeUnit.MILLISECONDS));
+
+ // make sure no extra message sneak in
+ Thread.sleep(1000);
+
+ assertEquals(2, messages.get());
+ }
+ }
+
+ @Test
+ public void shouldFailWithBadScriptEval() throws Exception {
+ try (SimpleClient client = new WebSocketClient()) {
+ final RequestMessage request = RequestMessage.build(Tokens.OPS_EVAL)
+ .addArg(Tokens.ARGS_GREMLIN, "new String().doNothingAtAllBecauseThis is a syntax error").create();
+
+ // set the latch to two as there should be two responses when you include the terminator -
+ // the error and the terminator
+ final CountDownLatch latch = new CountDownLatch(2);
+ final AtomicInteger messages = new AtomicInteger(0);
+ final AtomicBoolean errorReceived = new AtomicBoolean(false);
+ final AtomicBoolean terminatorReceived = new AtomicBoolean(false);
+ client.submit(request, r -> {
+ if (latch.getCount() == 2)
+ errorReceived.set(r.getStatus().equals(ResponseStatusCode.SERVER_ERROR_SCRIPT_EVALUATION));
+ else if (latch.getCount() == 1)
+ terminatorReceived.set(r.getStatus().equals(ResponseStatusCode.SUCCESS_TERMINATOR));
+ latch.countDown();
+ System.out.println("*******" + r);
+ messages.incrementAndGet();
+ });
+
+ assertTrue(latch.await(1500, TimeUnit.MILLISECONDS));
+
+ // make sure no extra message sneak in
+ Thread.sleep(1000);
+
+ assertEquals(2, messages.get());
+
+ }
+ }
+
// todo: get this test to pass - count connection and block incoming requests.
@Test
@org.junit.Ignore