You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by xi...@apache.org on 2022/12/19 20:06:12 UTC
[tinkerpop] branch 3.5-dev updated: Add additional information to TimeoutExceptions.
This is an automated email from the ASF dual-hosted git repository.
xiazcy pushed a commit to branch 3.5-dev
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/3.5-dev by this push:
new a3312bc830 Add additional information to TimeoutExceptions.
new d804b7dc99 Merge pull request #1909 from Bit-Quill/ken/TimeoutException
a3312bc830 is described below
commit a3312bc8309ce6f3765688da9d5fed14403da46a
Author: Ken Hu <10...@users.noreply.github.com>
AuthorDate: Fri Dec 2 19:17:22 2022 -0800
Add additional information to TimeoutExceptions.
---
.../apache/tinkerpop/gremlin/driver/Client.java | 2 +
.../tinkerpop/gremlin/driver/ConnectionPool.java | 52 ++++++-
.../gremlin/driver/SimpleSocketServer.java | 5 +
.../tinkerpop/gremlin/driver/TestChannelizers.java | 102 +++++++++++++
...bSocketServerHandler.java => TestHandlers.java} | 28 ++--
.../gremlin/driver/TestHttpServerInitializer.java | 40 ------
.../gremlin/driver/TestWSGremlinInitializer.java | 14 +-
.../gremlin/driver/TestWSNoOpInitializer.java | 33 -----
.../driver/TestWebSocketServerInitializer.java | 43 ------
.../WebSocketClientBehaviorIntegrateTest.java | 158 ++++++++++++++++++++-
.../gremlin/server/GremlinDriverIntegrateTest.java | 35 +++++
11 files changed, 376 insertions(+), 136 deletions(-)
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 0226d32e18..513ec92096 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
@@ -65,6 +65,8 @@ import java.util.stream.IntStream;
public abstract class Client {
private static final Logger logger = LoggerFactory.getLogger(Client.class);
+ public static final String TOO_MANY_IN_FLIGHT_REQUESTS = "Number of active requests exceeds pool size. " +
+ "Consider increasing the value for maxConnectionPoolSize.";
protected final Cluster cluster;
protected volatile boolean initialized;
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java
index 66fb815dfd..c393106dd0 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionPool.java
@@ -22,6 +22,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.tinkerpop.gremlin.driver.exception.ConnectionException;
import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
+import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
import org.apache.tinkerpop.gremlin.util.TimeUtil;
import java.util.ArrayList;
@@ -52,6 +53,8 @@ final class ConnectionPool {
public static final int MAX_POOL_SIZE = 8;
public static final int MIN_SIMULTANEOUS_USAGE_PER_CONNECTION = 8;
public static final int MAX_SIMULTANEOUS_USAGE_PER_CONNECTION = 16;
+ // A small buffer in millis used for comparing if a connection was created within a certain amount of time.
+ private static final int CONNECTION_SETUP_TIME_DELTA = 25;
public final Host host;
private final Cluster cluster;
@@ -67,6 +70,8 @@ final class ConnectionPool {
private final String poolLabel;
private final AtomicInteger scheduledForCreation = new AtomicInteger();
+ private final AtomicReference<ConnectionResult> latestConnectionResult = new AtomicReference<>(new ConnectionResult());
+
private final AtomicReference<CompletableFuture<Void>> closeFuture = new AtomicReference<>();
private volatile int waiter = 0;
@@ -74,6 +79,21 @@ final class ConnectionPool {
private final Condition hasAvailableConnection = waitLock.newCondition();
ConnectionFactory connectionFactory;
+ /**
+ * The result of a connection attempt. A null value for failureCause means that the connection attempt was successful.
+ */
+ public class ConnectionResult {
+ private long timeOfConnectionAttempt;
+ private Throwable failureCause;
+
+ public ConnectionResult() {}
+
+ public Throwable getFailureCause() { return failureCause; }
+ public long getTime() { return timeOfConnectionAttempt; }
+ public void setFailureCause(Throwable cause) { failureCause = cause; }
+ public void setTimeNow() { timeOfConnectionAttempt = System.currentTimeMillis(); }
+ }
+
public ConnectionPool(final Host host, final Client client) {
this(host, client, Optional.empty(), Optional.empty());
}
@@ -105,11 +125,19 @@ final class ConnectionPool {
final List<CompletableFuture<Void>> connectionCreationFutures = new ArrayList<>();
for (int i = 0; i < minPoolSize; i++) {
connectionCreationFutures.add(CompletableFuture.runAsync(() -> {
+ final ConnectionResult result = new ConnectionResult();
try {
this.connections.add(connectionFactory.create(this));
this.open.incrementAndGet();
} catch (ConnectionException e) {
+ result.setFailureCause(e);
throw new CompletionException(e);
+ } finally {
+ result.setTimeNow();
+ if ((latestConnectionResult.get() == null) ||
+ (latestConnectionResult.get().getTime() < result.getTime())) {
+ latestConnectionResult.set(result);
+ }
}
}, cluster.connectionScheduler()));
}
@@ -361,6 +389,7 @@ final class ConnectionPool {
return false;
}
+ final ConnectionResult result = new ConnectionResult();
try {
connections.add(connectionFactory.create(this));
} catch (Exception ex) {
@@ -370,7 +399,13 @@ final class ConnectionPool {
openCountToActOn, maxPoolSize),
ex);
considerHostUnavailable();
+ result.setFailureCause(ex);
return false;
+ } finally {
+ result.setTimeNow();
+ if (latestConnectionResult.get().getTime() < result.getTime()) {
+ latestConnectionResult.set(result);
+ }
}
announceAvailableConnection();
@@ -448,6 +483,19 @@ final class ConnectionPool {
logger.debug("Continue to wait for connection on {} if {} > 0", host, remaining);
} while (remaining > 0);
+ final StringBuilder cause = new StringBuilder("Potential Cause: ");
+ final ConnectionResult res = latestConnectionResult.get();
+ // If a connection attempt failed within a request's maxWaitForConnection then it likely contributed to that
+ // request getting a TimeoutException.
+ if (((System.currentTimeMillis() - res.getTime()) < (cluster.getMaxWaitForConnection() + CONNECTION_SETUP_TIME_DELTA)) &&
+ (res.getFailureCause() != null)) {
+ // Assumes that the root cause will give better information about why the connection failed.
+ cause.append(ExceptionHelper.getRootCause(res.getFailureCause()).getMessage());
+ } else if (open.get() >= maxPoolSize) {
+ cause.append(Client.TOO_MANY_IN_FLIGHT_REQUESTS);
+ } else {
+ cause.setLength(0);
+ }
// if we get to this point, we waited up to maxWaitForConnection from the pool and did not get a Connection.
// this can either mean the pool is exhausted or the host is unavailable. for the former, this could mean
// the connection pool and/or server is not sized correctly for the workload as connections are not freeing up
@@ -456,8 +504,8 @@ final class ConnectionPool {
// message written to the log should provide some insight into the state of the connection pool telling the
// user if the pool was running at maximum.
final String timeoutErrorMessage = String.format(
- "Timed-out (%s %s) waiting for connection on %s%s%s",
- timeout, unit, host, System.lineSeparator(), this.getPoolInfo());
+ "Timed-out (%s %s) waiting for connection on %s. %s%s%s",
+ timeout, unit, host, cause.toString(), System.lineSeparator(), this.getPoolInfo());
logger.error(timeoutErrorMessage);
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SimpleSocketServer.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SimpleSocketServer.java
index 84eaf4a534..dc9c513f00 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SimpleSocketServer.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SimpleSocketServer.java
@@ -51,4 +51,9 @@ public class SimpleSocketServer {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
+
+ public void stopSync() throws InterruptedException {
+ bossGroup.shutdownGracefully().sync();
+ workerGroup.shutdownGracefully().sync();
+ }
}
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestChannelizers.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestChannelizers.java
new file mode 100644
index 0000000000..8e05ff3688
--- /dev/null
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestChannelizers.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.driver;
+
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
+import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
+
+/**
+ * Class that holds the channelizers that can used for testing with the SimpleSocketServer.
+ */
+public class TestChannelizers {
+
+ /**
+ * A base ChannelInitializer that setups the pipeline for HTTP handling. This class should be sub-classed by a
+ * handler that handles the actual data being received.
+ */
+ public static abstract class TestHttpServerInitializer extends ChannelInitializer<SocketChannel> {
+ protected static final String WEBSOCKET_PATH = "/gremlin";
+
+ @Override
+ public void initChannel(SocketChannel ch) {
+ final ChannelPipeline pipeline = ch.pipeline();
+ pipeline.addLast(new HttpServerCodec());
+ pipeline.addLast(new HttpObjectAggregator(65536));
+ }
+ }
+
+ /**
+ * A vanilla WebSocket server Initializer implementation using Netty. This initializer would configure the server
+ * for WebSocket handshake and decoding incoming WebSocket frames.
+ */
+ public static abstract class TestWebSocketServerInitializer extends TestChannelizers.TestHttpServerInitializer {
+
+ @Override
+ public void initChannel(SocketChannel ch) {
+ super.initChannel(ch);
+
+ final ChannelPipeline pipeline = ch.pipeline();
+ pipeline.addLast(new WebSocketServerCompressionHandler());
+ pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
+ this.postInit(ch.pipeline());
+ }
+
+ public abstract void postInit(final ChannelPipeline ch);
+ }
+
+ /**
+ * An initializer that adds a handler that will drop WebSocket frames.
+ */
+ public static class TestWSNoOpInitializer extends TestHttpServerInitializer {
+
+ @Override
+ public void initChannel(SocketChannel ch) {
+ super.initChannel(ch);
+ ch.pipeline().addLast(new TestHandlers.NoOpWebSocketServerHandler(WEBSOCKET_PATH));
+ }
+ }
+
+ /**
+ * An initializer that adds a handler to mimics a server that is throttling connections.
+ */
+ public static class TestConnectionThrottlingInitializer extends TestHttpServerInitializer {
+ int connectionCount = 0;
+
+ @Override
+ public void initChannel(SocketChannel ch) {
+ super.initChannel(ch);
+
+ final ChannelPipeline pipeline = ch.pipeline();
+ if (connectionCount < 1) {
+ pipeline.addLast(new WebSocketServerCompressionHandler());
+ pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
+ pipeline.addLast(new TestWSGremlinInitializer.ClientTestConfigurableHandler());
+ } else {
+ pipeline.addLast(new TestHandlers.NoOpWebSocketServerHandler(WEBSOCKET_PATH));
+ }
+
+ connectionCount++;
+ }
+ }
+}
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/NoOpWebSocketServerHandler.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestHandlers.java
similarity index 60%
rename from gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/NoOpWebSocketServerHandler.java
rename to gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestHandlers.java
index 8c22c723fa..e373b023db 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/NoOpWebSocketServerHandler.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestHandlers.java
@@ -24,19 +24,25 @@ import io.netty.handler.codec.http.HttpRequest;
import io.netty.util.ReferenceCountUtil;
/**
-* Handler that will drop requests to the WebSocket path.
-*/
-public class NoOpWebSocketServerHandler extends ChannelInboundHandlerAdapter {
- private String websocketPath;
+ * Class that holds the handlers that can used by the initializers in TestChannelizers.
+ */
+public class TestHandlers {
- public NoOpWebSocketServerHandler(String websocketPath) {
- this.websocketPath = websocketPath;
- }
+ /**
+ * Handler that will drop requests to the WebSocket path.
+ */
+ public static class NoOpWebSocketServerHandler extends ChannelInboundHandlerAdapter {
+ private String websocketPath;
+
+ public NoOpWebSocketServerHandler(String websocketPath) {
+ this.websocketPath = websocketPath;
+ }
- @Override
- public void channelRead(ChannelHandlerContext ctx, Object msg) {
- if ((msg instanceof HttpRequest) && ((HttpRequest) msg).uri().endsWith(websocketPath)) {
- ReferenceCountUtil.release(msg);
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) {
+ if ((msg instanceof HttpRequest) && ((HttpRequest) msg).uri().endsWith(websocketPath)) {
+ ReferenceCountUtil.release(msg);
+ }
}
}
}
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestHttpServerInitializer.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestHttpServerInitializer.java
deleted file mode 100644
index 9985b70353..0000000000
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestHttpServerInitializer.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.tinkerpop.gremlin.driver;
-
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelPipeline;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.codec.http.HttpObjectAggregator;
-import io.netty.handler.codec.http.HttpServerCodec;
-
-/**
- * A base ChannelInitializer that setups the pipeline for HTTP handling. This class should be sub-classed by a handler
- * that handles the actual data being received.
- */
-public class TestHttpServerInitializer extends ChannelInitializer<SocketChannel> {
- protected static final String WEBSOCKET_PATH = "/gremlin";
-
- @Override
- public void initChannel(SocketChannel ch) {
- final ChannelPipeline pipeline = ch.pipeline();
- pipeline.addLast(new HttpServerCodec());
- pipeline.addLast(new HttpObjectAggregator(65536));
- }
-}
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestWSGremlinInitializer.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestWSGremlinInitializer.java
index 44d314caf8..b9e9b520c4 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestWSGremlinInitializer.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestWSGremlinInitializer.java
@@ -47,7 +47,7 @@ import java.util.UUID;
* Initializer which partially mimics the Gremlin Server. This initializer injects a handler in the
* server pipeline that can be modified to send the desired response for a test case.
*/
-public class TestWSGremlinInitializer extends TestWebSocketServerInitializer {
+public class TestWSGremlinInitializer extends TestChannelizers.TestWebSocketServerInitializer {
private static final Logger logger = LoggerFactory.getLogger(TestWSGremlinInitializer.class);
/**
* If a request with this ID comes to the server, the server responds back with a single vertex picked from Modern
@@ -91,7 +91,7 @@ public class TestWSGremlinInitializer extends TestWebSocketServerInitializer {
/**
* Handler introduced in the server pipeline to configure expected response for test cases.
*/
- private static class ClientTestConfigurableHandler extends MessageToMessageDecoder<BinaryWebSocketFrame> {
+ static class ClientTestConfigurableHandler extends MessageToMessageDecoder<BinaryWebSocketFrame> {
private String userAgent = "";
@Override
protected void decode(final ChannelHandlerContext ctx, final BinaryWebSocketFrame frame, final List<Object> objects)
@@ -135,9 +135,15 @@ public class TestWSGremlinInitializer extends TestWebSocketServerInitializer {
} else if (msg.getRequestId().equals(RESPONSE_CONTAINS_SERVER_ERROR_REQUEST_ID)) {
Thread.sleep(1000);
ctx.channel().writeAndFlush(new CloseWebSocketFrame());
- }
- else if (msg.getRequestId().equals(USER_AGENT_REQUEST_ID)) {
+ } else if (msg.getRequestId().equals(USER_AGENT_REQUEST_ID)) {
ctx.channel().writeAndFlush(new TextWebSocketFrame(returnSimpleStringResponse(USER_AGENT_REQUEST_ID, userAgent)));
+ } else {
+ try {
+ Thread.sleep(Long.parseLong((String) msg.getArgs().get("gremlin")));
+ ctx.channel().writeAndFlush(new TextWebSocketFrame(returnSingleVertexResponse(msg.getRequestId())));
+ } catch (NumberFormatException nfe) {
+ // Ignore. Only return a vertex if the query was a long value.
+ }
}
}
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestWSNoOpInitializer.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestWSNoOpInitializer.java
deleted file mode 100644
index 268144a7af..0000000000
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestWSNoOpInitializer.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.tinkerpop.gremlin.driver;
-
-import io.netty.channel.socket.SocketChannel;
-
-/**
- * An initializer that adds a handler that will drop WebSocket frames.
- */
-public class TestWSNoOpInitializer extends TestHttpServerInitializer {
-
- @Override
- public void initChannel(SocketChannel ch) {
- super.initChannel(ch);
- ch.pipeline().addLast(new NoOpWebSocketServerHandler(WEBSOCKET_PATH));
- }
-}
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestWebSocketServerInitializer.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestWebSocketServerInitializer.java
deleted file mode 100644
index adfa723549..0000000000
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/TestWebSocketServerInitializer.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.driver;
-
-import io.netty.channel.ChannelPipeline;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
-import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
-
-/**
- * A vanilla WebSocket server Initializer implementation using Netty. This initializer would configure the server for
- * WebSocket handshake and decoding incoming WebSocket frames.
- */
-public abstract class TestWebSocketServerInitializer extends TestHttpServerInitializer {
-
- @Override
- public void initChannel(SocketChannel ch) {
- super.initChannel(ch);
-
- final ChannelPipeline pipeline = ch.pipeline();
- pipeline.addLast(new WebSocketServerCompressionHandler());
- pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
- this.postInit(ch.pipeline());
- }
-
- public abstract void postInit(final ChannelPipeline ch);
-}
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/WebSocketClientBehaviorIntegrateTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/WebSocketClientBehaviorIntegrateTest.java
index 10e3996288..cd01ede88a 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/WebSocketClientBehaviorIntegrateTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/WebSocketClientBehaviorIntegrateTest.java
@@ -29,13 +29,18 @@ import org.apache.log4j.Level;
import org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
import org.apache.tinkerpop.gremlin.driver.ser.Serializers;
import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
import org.apache.tinkerpop.gremlin.util.Log4jRecordingAppender;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.hamcrest.core.Is.is;
@@ -71,7 +76,10 @@ public class WebSocketClientBehaviorIntegrateTest {
server = new SimpleSocketServer();
if (name.getMethodName().equals("shouldAttemptHandshakeForLongerThanDefaultNettySslHandshakeTimeout") ||
name.getMethodName().equals("shouldPrintCorrectErrorForRegularWebSocketHandshakeTimeout")) {
- server.start(new TestWSNoOpInitializer());
+ server.start(new TestChannelizers.TestWSNoOpInitializer());
+ } else if (name.getMethodName().equals("shouldContinueRunningRemainingConnectionsIfServerThrottlesNewConnections") ||
+ name.getMethodName().equals("shouldReturnCorrectExceptionIfServerThrottlesNewConnectionsAndMaxWaitExceeded")) {
+ server.start(new TestChannelizers.TestConnectionThrottlingInitializer());
} else {
server.start(new TestWSGremlinInitializer());
}
@@ -349,17 +357,19 @@ public class WebSocketClientBehaviorIntegrateTest {
assertTrue(caught instanceof NoHostAvailableException);
assertTrue(recordingAppender.getMessages().stream().anyMatch(str -> str.contains("SSL handshake not completed")));
}
+
+ cluster.close();
}
/**
* Tests to make sure that the correct error message is logged when a non-SSL connection attempt times out.
*/
@Test
- public void shouldPrintCorrectErrorForRegularWebSocketHandshakeTimeout() {
+ public void shouldPrintCorrectErrorForRegularWebSocketHandshakeTimeout() throws InterruptedException {
final Cluster cluster = Cluster.build("localhost").port(SimpleSocketServer.PORT)
.minConnectionPoolSize(1)
.maxConnectionPoolSize(1)
- .connectionSetupTimeoutMillis(100)
+ .connectionSetupTimeoutMillis(120)
.create();
final Client.ClusteredClient client = cluster.connect();
@@ -372,7 +382,149 @@ public class WebSocketClientBehaviorIntegrateTest {
} finally {
assertTrue(caught != null);
assertTrue(caught instanceof NoHostAvailableException);
+ Thread.sleep(150);
assertTrue(recordingAppender.getMessages().stream().anyMatch(str -> str.contains("WebSocket handshake not completed")));
}
+
+ cluster.close();
+ }
+
+ /**
+ * Tests that if a server throttles new connections (doesn't allow new connections to be made) then all requests
+ * will run and complete on the connections that are already open.
+ */
+ @Test
+ public void shouldContinueRunningRemainingConnectionsIfServerThrottlesNewConnections() throws ExecutionException, InterruptedException, TimeoutException {
+ final Cluster cluster = Cluster.build("localhost").port(SimpleSocketServer.PORT)
+ .minConnectionPoolSize(1)
+ .maxConnectionPoolSize(5)
+ .maxWaitForConnection(15000) // large value ensures that request will eventually find a connection.
+ .connectionSetupTimeoutMillis(1000)
+ .minInProcessPerConnection(0)
+ .maxInProcessPerConnection(1)
+ .minSimultaneousUsagePerConnection(0)
+ .maxSimultaneousUsagePerConnection(1)
+ .serializer(Serializers.GRAPHSON_V2D0)
+ .create();
+
+ final Client.ClusteredClient client = cluster.connect();
+
+ final List<CompletableFuture<ResultSet>> results = new ArrayList<CompletableFuture<ResultSet>>();
+ for (int i = 0; i < 5; i++) {
+ results.add(client.submitAsync("200"));
+ Thread.sleep(150); // Delay needed in 3.5-dev to ensure attempt to create new connection
+ }
+
+ for (CompletableFuture<ResultSet> result : results) {
+ assertNotNull(result.get(60000, TimeUnit.MILLISECONDS).one().getVertex());
+ }
+
+ cluster.close();
+ }
+
+ /**
+ * Tests that if a server throttles new connections (doesn't allow new connections to be made) then any request
+ * that can't find a connection within its maxWaitForConnection will return an informative exception regarding
+ * the inability to open new connections.
+ */
+ @Test
+ public void shouldReturnCorrectExceptionIfServerThrottlesNewConnectionsAndMaxWaitExceeded() {
+ final Cluster cluster = Cluster.build("localhost").port(SimpleSocketServer.PORT)
+ .minConnectionPoolSize(1)
+ .maxConnectionPoolSize(5)
+ .maxWaitForConnection(250) // small value ensures that requests will return TimeoutException.
+ .connectionSetupTimeoutMillis(100)
+ .minInProcessPerConnection(0)
+ .maxInProcessPerConnection(1)
+ .minSimultaneousUsagePerConnection(0)
+ .maxSimultaneousUsagePerConnection(1)
+ .serializer(Serializers.GRAPHSON_V2D0)
+ .create();
+
+ final Client.ClusteredClient client = cluster.connect();
+
+ for (int i = 0; i < 5; i++) {
+ try {
+ client.submitAsync("3000");
+ } catch (Exception e) {
+ final Throwable rootCause = ExceptionHelper.getRootCause(e);
+ assertTrue(rootCause instanceof TimeoutException);
+ assertTrue(rootCause.getMessage().contains("WebSocket handshake not completed"));
+ }
+
+ try {
+ Thread.sleep(25); // Delay needed in 3.5-dev to ensure attempt to create new connection
+ } catch (Exception ignored) {
+ // Ignored.
+ }
+ }
+
+ cluster.close();
+ }
+
+ /**
+ * Tests that the client continues to work if the server temporarily goes down between two requests.
+ */
+ @Test
+ public void shouldContinueRunningIfServerGoesDownTemporarily() throws InterruptedException {
+ final Cluster cluster = Cluster.build("localhost").port(SimpleSocketServer.PORT)
+ .minConnectionPoolSize(1)
+ .serializer(Serializers.GRAPHSON_V2D0)
+ .create();
+
+ final Client.ClusteredClient client = cluster.connect();
+ final Object lock = new Object();
+
+ final ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(1);
+ scheduledPool.schedule(() -> {
+ try {
+ server.stopSync();
+ server = new SimpleSocketServer();
+ server.start(new TestWSGremlinInitializer());
+ synchronized (lock) {
+ lock.notify();
+ }
+ } catch (InterruptedException ignored) {
+ // Ignored.
+ }
+ }, 1000, TimeUnit.MILLISECONDS);
+
+ synchronized (lock) {
+ assertNotNull(client.submit("1").one().getVertex());
+ lock.wait(30000);
+ }
+
+ assertNotNull(client.submit("1").one().getVertex());
+
+ cluster.close();
+ }
+
+ /**
+ * Tests that if the host is unavailable then the client will return an exception that contains information about
+ * the status of the host.
+ */
+ @Test
+ public void shouldReturnCorrectExceptionIfServerGoesDown() throws InterruptedException {
+ final Cluster cluster = Cluster.build("localhost").port(SimpleSocketServer.PORT)
+ .minConnectionPoolSize(1)
+ .maxWaitForConnection(500)
+ .connectionSetupTimeoutMillis(100)
+ .serializer(Serializers.GRAPHSON_V2D0)
+ .create();
+
+ final Client.ClusteredClient client = cluster.connect();
+ client.submit("1");
+
+ server.stopSync();
+
+ try {
+ client.submit("1");
+ } catch (Exception e) {
+ final Throwable rootCause = ExceptionHelper.getRootCause(e);
+ assertTrue(rootCause instanceof TimeoutException);
+ assertTrue(rootCause.getMessage().contains("Connection refused"));
+ }
+
+ cluster.close();
}
}
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 43a53bae7a..ab8c432da2 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
@@ -79,6 +79,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
@@ -1832,6 +1833,40 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
}
}
+ /**
+ * Tests to make sure that the stack trace contains an informative cause when the request times out because the
+ * client was unable to get a connection and the pool is already maxed out.
+ */
+ @Test
+ public void shouldReturnClearExceptionCauseWhenClientIsTooBusyAndConnectionPoolIsFull() throws InterruptedException {
+ final Cluster cluster = TestClientFactory.build()
+ .minConnectionPoolSize(1)
+ .maxConnectionPoolSize(1)
+ .connectionSetupTimeoutMillis(100)
+ .maxWaitForConnection(150)
+ .minInProcessPerConnection(0)
+ .maxInProcessPerConnection(1)
+ .minSimultaneousUsagePerConnection(0)
+ .maxSimultaneousUsagePerConnection(1)
+ .create();
+
+ final Client.ClusteredClient client = cluster.connect();
+
+ for (int i = 0; i < 3; i++) {
+ try {
+ client.submitAsync("Thread.sleep(5000);");
+ } catch (Exception e) {
+ final Throwable root = ExceptionHelper.getRootCause(e);
+ assertTrue(root instanceof TimeoutException);
+ assertTrue(root.getMessage().contains(Client.TOO_MANY_IN_FLIGHT_REQUESTS));
+ }
+
+ Thread.sleep(300);
+ }
+
+ cluster.close();
+ }
+
/**
* Client created on an initially dead host should fail initially, and recover after the dead host has restarted
* @param testClusterClient - boolean flag set to test clustered client if true and sessioned client if false.