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 2022/12/01 18:28:02 UTC
[tinkerpop] branch TINKERPOP-2813 updated (aa130721ba -> 1c85f3770e)
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a change to branch TINKERPOP-2813
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
discard aa130721ba TINKERPOP-2813 Updated upgrade docs and reference docs
discard 4401f0163a TINKERPOP-2813 Fixed some issues with Cluster.close()
discard 942c6b8f61 TINKERPOP-2813 Removed fast NoHostAvailableException
discard 5dead2df32 TINKERPOP-2813 Refactored job scheduling and thread pools
discard aae18500bc TINKERPOP-2813 Improved logging around server-side close
add e403ad40d9 TINKERPOP-2807 Added documentation around aiohttp heartbeat
add 4528ec33ab Added test to demonstrate tx() issue of TINKERPOP-2832 CTR
new 320ba48b95 TINKERPOP-2813 Improved logging around server-side close
new 2d5f9fee63 TINKERPOP-2813 Refactored job scheduling and thread pools
new 96274275a8 TINKERPOP-2813 Removed fast NoHostAvailableException
new 5c9d0e0327 TINKERPOP-2813 Fixed some issues with Cluster.close()
new 1c85f3770e TINKERPOP-2813 Updated upgrade docs and reference docs
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (aa130721ba)
\
N -- N -- N refs/heads/TINKERPOP-2813 (1c85f3770e)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 5 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.
Summary of changes:
docs/src/reference/gremlin-variants.asciidoc | 14 ++++--
.../gremlin_python/driver/aiohttp/transport.py | 9 ++--
.../server/GremlinSessionTxIntegrateTest.java | 50 ++++++++++++++++++++++
3 files changed, 65 insertions(+), 8 deletions(-)
[tinkerpop] 04/05: TINKERPOP-2813 Fixed some issues with Cluster.close()
Posted by sp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch TINKERPOP-2813
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 5c9d0e0327ba710b3ae432d9d90b299f6b7c545d
Author: Stephen Mallette <st...@amazon.com>
AuthorDate: Mon Nov 28 13:30:08 2022 -0500
TINKERPOP-2813 Fixed some issues with Cluster.close()
After refactoring to add the new thread pools, some situations seemed to arise where connections didn't close cleanly. Not clear if this came as a result of the thread pool changes or if they simply were always present but hidden behind use of fork/join pool and the normal gremlin pool. Still don't quite have a good async shutdown going but that wasn't quite working that way prior to these changes so perhaps that is better saved for a dedicated body of work.
---
.../apache/tinkerpop/gremlin/driver/Cluster.java | 38 +++++++++++++++-------
.../tinkerpop/gremlin/driver/Connection.java | 28 +++++++++-------
.../tinkerpop/gremlin/driver/ConnectionPool.java | 19 ++++++++---
.../driver/ClientConnectionIntegrateTest.java | 4 ++-
4 files changed, 61 insertions(+), 28 deletions(-)
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
index e93074d3fc..7392deb7d5 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
@@ -24,6 +24,7 @@ import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SslProvider;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import io.netty.util.concurrent.Future;
import org.apache.commons.configuration2.Configuration;
import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
import org.apache.tinkerpop.gremlin.driver.ser.Serializers;
@@ -60,6 +61,7 @@ import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -1021,10 +1023,14 @@ public final class Cluster {
return b;
}
- void shutdown() {
+ /**
+ * Gracefully shutsdown the event loop and returns the termination future which signals that all jobs are done.
+ */
+ Future<?> shutdown() {
// Do not provide a quiet period (default is 2s) to accept more requests. Once we have decided to shutdown,
// no new requests should be accepted.
- group.shutdownGracefully(/*quiet period*/0, /*timeout*/2, TimeUnit.SECONDS).awaitUninterruptibly();
+ group.shutdownGracefully(/*quiet period*/0, /*timeout*/2, TimeUnit.SECONDS);
+ return group.terminationFuture();
}
}
@@ -1219,25 +1225,33 @@ public final class Cluster {
if (closeFuture.get() != null)
return closeFuture.get();
+ final List<CompletableFuture<Void>> clientCloseFutures = new ArrayList<>(openedClients.size());
for (WeakReference<Client> openedClient : openedClients) {
final Client client = openedClient.get();
- if (client != null && !client.isClosing()) {
- client.close();
+ if (client != null) {
+ // best to call close() even if the Client is already closing so that we can be sure that
+ // any background client closing operations are included in this shutdown future
+ clientCloseFutures.add(client.closeAsync());
}
}
- final CompletableFuture<Void> closeIt = new CompletableFuture<>();
- closeFuture.set(closeIt);
+ // when all the clients are fully closed then shutdown the netty event loop. not sure why this needs to
+ // block here, but if it doesn't then factory.shutdown() below doesn't seem to want to ever complete.
+ // ideally, this should all be async, but i guess it wasn't before this change so just going to leave it
+ // for now as this really isn't the focus on this change
+ CompletableFuture.allOf(clientCloseFutures.toArray(new CompletableFuture[0])).join();
- hostScheduler.submit(() -> {
- factory.shutdown();
+ final CompletableFuture<Void> closeIt = new CompletableFuture<>();
+ // shutdown the event loop. that shutdown can trigger some final jobs to get scheduled so add a listener
+ // to the termination event to shutdown remaining thread pools
+ factory.shutdown().awaitUninterruptibly().addListener(f -> {
+ executor.shutdown();
+ hostScheduler.shutdown();
+ connectionScheduler.shutdown();
closeIt.complete(null);
});
- // Prevent the executor from accepting new tasks while still allowing enqueued tasks to complete
- executor.shutdown();
- connectionScheduler.shutdown();
- hostScheduler.shutdown();
+ closeFuture.set(closeIt);
return closeIt;
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
index 0dfe0dbab3..1183a3bdc4 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
@@ -29,6 +29,7 @@ import io.netty.channel.ChannelPromise;
import io.netty.channel.socket.nio.NioSocketChannel;
import java.net.URI;
+import java.time.Instant;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -56,6 +57,8 @@ final class Connection {
private final Cluster cluster;
private final Client client;
private final ConnectionPool pool;
+ private final String creatingThread;
+ private final String createdTimestamp;
public static final int MAX_IN_PROCESS = 4;
public static final int MIN_IN_PROCESS = 1;
@@ -95,7 +98,8 @@ final class Connection {
this.client = pool.getClient();
this.pool = pool;
this.maxInProcess = maxInProcess;
-
+ this.creatingThread = Thread.currentThread().getName();
+ this.createdTimestamp = Instant.now().toString();
connectionLabel = "Connection{host=" + pool.host + "}";
if (cluster.isClosing())
@@ -302,8 +306,6 @@ final class Connection {
// guess). that seems to put the executor thread in a monitor state that it doesn't recover from. since all
// the code in here is behind shutdownInitiated the synchronized doesn't seem necessary
if (shutdownInitiated.compareAndSet(false, true)) {
- final String connectionInfo = this.getConnectionInfo();
-
// the session close message was removed in 3.5.0 after deprecation at 3.3.11. That removal was perhaps
// a bit hasty as session semantics may still require this message in certain cases. Until we can look
// at this in more detail, it seems best to bring back the old functionality to the driver.
@@ -342,7 +344,8 @@ final class Connection {
channelizer.close(channel);
// seems possible that the channelizer could initialize but fail to produce a channel, so worth checking
- // null before proceeding here
+ // null before proceeding here. also if the cluster is in shutdown then the event loop could be shutdown
+ // already and there will be no way to get a new promise out there.
if (channel != null) {
final ChannelPromise promise = channel.newPromise();
promise.addListener(f -> {
@@ -350,7 +353,7 @@ final class Connection {
future.completeExceptionally(f.cause());
} else {
if (logger.isDebugEnabled())
- logger.debug("{} destroyed successfully.", connectionInfo);
+ logger.debug("{} destroyed successfully.", this.getConnectionInfo());
future.complete(null);
}
@@ -365,9 +368,12 @@ final class Connection {
}
}
} else {
- logger.debug("Connection {} is shutting down but the channel was not initialized to begin with",
- getConnectionInfo());
+ // if we dont handle the supplied future it can hang the close
+ future.complete(null);
}
+ } else {
+ // if we dont handle the supplied future it can hang the close
+ future.complete(null);
}
}
@@ -385,10 +391,10 @@ final class Connection {
*/
public String getConnectionInfo(final boolean showHost) {
return showHost ?
- String.format("Connection{channel=%s host=%s isDead=%s borrowed=%s pending=%s markedReplaced=%s closing=%s}",
- getChannelId(), pool.host.toString(), isDead(), this.borrowed.get(), getPending().size(), this.isBeingReplaced, isClosing()) :
- String.format("Connection{channel=%s isDead=%s borrowed=%s pending=%s markedReplaced=%s closing=%s}",
- getChannelId(), isDead(), this.borrowed.get(), getPending().size(), this.isBeingReplaced, isClosing());
+ String.format("Connection{channel=%s host=%s isDead=%s borrowed=%s pending=%s markedReplaced=%s closing=%s created=%s thread=%s}",
+ getChannelId(), pool.host.toString(), isDead(), this.borrowed.get(), getPending().size(), this.isBeingReplaced, isClosing(), createdTimestamp, creatingThread) :
+ String.format("Connection{channel=%s isDead=%s borrowed=%s pending=%s markedReplaced=%s closing=%s created=%s thread=%s}",
+ getChannelId(), isDead(), this.borrowed.get(), getPending().size(), this.isBeingReplaced, isClosing(), createdTimestamp, creatingThread);
}
/**
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 4e5f019273..0ec3a4c3af 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
@@ -272,14 +272,25 @@ final class ConnectionPool {
return bin.size();
}
+ /**
+ * Calls close on connections in the pool gathering close futures from both active connections and ones in the
+ * bin.
+ */
private CompletableFuture<Void> killAvailableConnections() {
- final List<CompletableFuture<Void>> futures = new ArrayList<>(connections.size());
+ final List<CompletableFuture<Void>> futures = new ArrayList<>(connections.size() + bin.size());
for (Connection connection : connections) {
final CompletableFuture<Void> future = connection.closeAsync();
future.thenRun(open::decrementAndGet);
futures.add(future);
}
+ // Without the ones in the bin the close for the ConnectionPool won't account for their shutdown and could
+ // lead to scenario where the bin connections stay open after the channel executor is closed which then
+ // leads to close operation getting rejected in Connection.close() for channel.newPromise().
+ for (Connection connection : bin) {
+ futures.add(connection.closeAsync());
+ }
+
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
}
@@ -368,7 +379,7 @@ final class ConnectionPool {
private boolean destroyConnection(final Connection connection) {
while (true) {
- int opened = open.get();
+ final int opened = open.get();
if (opened <= minPoolSize)
return false;
@@ -598,8 +609,8 @@ final class ConnectionPool {
} else {
final int connectionCount = connections.size();
sb.append(System.lineSeparator());
- sb.append(String.format("Connection Pool Status (size=%s max=%s min=%s toCreate=%s markedOpen=%s bin=%s)",
- connectionCount, maxPoolSize, minPoolSize, this.scheduledForCreation.get(), this.open.get(), bin.size()));
+ sb.append(String.format("Connection Pool Status (size=%s max=%s min=%s toCreate=%s bin=%s)",
+ connectionCount, maxPoolSize, minPoolSize, this.scheduledForCreation.get(), bin.size()));
sb.append(System.lineSeparator());
appendConnections(sb, connectionToCallout, connections);
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java
index 1417409c47..8899db2e9e 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java
@@ -184,6 +184,8 @@ public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrat
// if there was a exception in the worker thread, then it had better be a TimeoutException
assertThat(hadFailOtherThanTimeout.get(), is(false));
+ connectionFactory.jittery = false;
+
cluster.close();
}
@@ -212,7 +214,7 @@ public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrat
if (jittery && connectionsCreated.incrementAndGet() % numberOfConnectionsBetweenErrors == 0) {
connectionFailures.incrementAndGet();
throw new ConnectionException(pool.host.getHostUri(),
- new SSLHandshakeException("SSL on the funk - server is big mad"));
+ new SSLHandshakeException("SSL on the funk - server is big mad with the jitters"));
}
return ConnectionFactory.super.create(pool);
[tinkerpop] 01/05: TINKERPOP-2813 Improved logging around server-side close
Posted by sp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch TINKERPOP-2813
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 320ba48b9596ce685dee68d7215397991383742b
Author: Stephen Mallette <st...@amazon.com>
AuthorDate: Fri Oct 14 13:12:07 2022 -0400
TINKERPOP-2813 Improved logging around server-side close
Prior to this change it was impossible to know if the connection closed as a result of a client-side or server-side issue. The log messages now make that more clear. Also, improved the output of the info from Connection and ConnectionPool so that it logged a bit more nicely.
---
CHANGELOG.asciidoc | 1 +
.../tinkerpop/gremlin/driver/Connection.java | 27 ++++++++-
.../tinkerpop/gremlin/driver/ConnectionPool.java | 64 ++++++++++++++++++----
3 files changed, 79 insertions(+), 13 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 30fec09ff2..7212285da6 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -41,6 +41,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Bumped `jackson-databind` to 2.14.0 to fix security vulnerability.
* Bumped to Groovy 2.5.19.
* Bumped to Netty 4.1.85.
+* Improved logging for `gremlin-driver`.
==== Bugs
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
index 00536e63ca..e2f5717714 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
@@ -123,8 +123,14 @@ final class Connection {
channel.closeFuture().addListener((ChannelFutureListener) future -> {
logger.debug("OnChannelClose callback called for channel {}", channel);
- // Replace the channel if it was not intentionally closed using CloseAsync method.
+ // if the closeFuture is not set, it means that closeAsync() wasn't called which means that the
+ // close did not come from the client side. it means the server closed the channel for some reason.
+ // it's important to distinguish that difference in debugging
if (thisConnection.closeFuture.get() == null) {
+ logger.error(String.format(
+ "Server closed the Connection on channel %s - scheduling removal from %s",
+ channel.id().asShortText(), thisConnection.pool.getPoolInfo(thisConnection)));
+
// delegate the task to worker thread and free up the event loop
thisConnection.cluster.executor().submit(() -> thisConnection.pool.definitelyDestroyConnection(thisConnection));
}
@@ -355,9 +361,24 @@ final class Connection {
}
}
+ /**
+ * Gets a message that describes the state of the connection.
+ */
public String getConnectionInfo() {
- return String.format("Connection{channel=%s, host=%s, isDead=%s, borrowed=%s, pending=%s}",
- channel, pool.host, isDead(), borrowed, pending.size());
+ return getConnectionInfo(true);
+ }
+
+ /**
+ * Gets a message that describes the state of the connection.
+ *
+ * @param showHost determines if the {@link Host} should be displayed in the message.
+ */
+ public String getConnectionInfo(final boolean showHost) {
+ return showHost ?
+ String.format("Connection{channel=%s, host=%s, isDead=%s, borrowed=%s, pending=%s}",
+ channel.id().asShortText(), pool.host, isDead(), borrowed, pending.size()) :
+ String.format("Connection{channel=%s, isDead=%s, borrowed=%s, pending=%s}",
+ channel.id().asShortText(), isDead(), borrowed, pending.size());
}
/**
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 ca42dbc791..f1d8f69eb7 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
@@ -371,9 +371,10 @@ final class ConnectionPool {
// only close the connection for good once it is done being borrowed or when it is dead
if (connection.isDead() || connection.borrowed.get() == 0) {
if (bin.remove(connection)) {
- connection.closeAsync();
- // TODO: Log the following message on completion of the future returned by closeAsync.
- logger.debug("{} destroyed", connection.getConnectionInfo());
+ final CompletableFuture<Void> closeFuture = connection.closeAsync();
+ closeFuture.whenComplete((v, t) -> {
+ logger.debug("Destroyed {}{}{}", connection.getConnectionInfo(), System.lineSeparator(), this.getPoolInfo());
+ });
}
}
}
@@ -416,13 +417,24 @@ final class ConnectionPool {
logger.debug("Continue to wait for connection on {} if {} > 0", host, remaining);
} while (remaining > 0);
- logger.error("Timed-out ({} {}) waiting for connection on {} - possibly unavailable", timeout, unit, host);
+ // 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
+ // to keep up requests and for the latter it could mean anything from a network hiccup to the server simply
+ // being down. it is critical that the caller be able to discern between these two. the following error
+ // 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());
+
+ logger.error(timeoutErrorMessage);
// if we timeout borrowing a connection that might mean the host is dead (or the timeout was super short).
// either way supply a function to reconnect
this.considerHostUnavailable();
- throw new TimeoutException("Timed-out waiting for connection on " + host + " - possibly unavailable");
+ throw new TimeoutException(timeoutErrorMessage);
}
public void considerHostUnavailable() {
@@ -521,14 +533,46 @@ final class ConnectionPool {
return connections.stream().map(Connection::getChannelId).collect(Collectors.toSet());
}
+ /**
+ * Gets a message that describes the state of the connection pool.
+ */
public String getPoolInfo() {
+ return getPoolInfo(null);
+ }
+
+ /**
+ * Gets a message that describes the state of the connection pool.
+ *
+ * @param connectionToCallout the connection from the pool to identify more clearly in the message
+ */
+ public String getPoolInfo(final Connection connectionToCallout) {
final StringBuilder sb = new StringBuilder("ConnectionPool (");
sb.append(host);
- sb.append(") - ");
- connections.forEach(c -> {
- sb.append(c);
- sb.append(",");
- });
+ sb.append(")");
+
+ if (connections.isEmpty()) {
+ sb.append("- no connections in pool");
+ } else {
+ final int connectionCount = connections.size();
+ sb.append(System.lineSeparator());
+ sb.append(String.format("Connection Pool Status (size=%s max=%s min=%s)", connectionCount, maxPoolSize, minPoolSize));
+ sb.append(System.lineSeparator());
+
+ for (int ix = 0; ix < connectionCount; ix++) {
+ final Connection c = connections.get(ix);
+
+ if (c == connectionToCallout)
+ sb.append("==> ");
+ else
+ sb.append("> ");
+
+ sb.append(c.getConnectionInfo(false));
+
+ if (ix < connectionCount - 1)
+ sb.append(System.lineSeparator());
+ }
+ }
+
return sb.toString().trim();
}
[tinkerpop] 02/05: TINKERPOP-2813 Refactored job scheduling and thread pools
Posted by sp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch TINKERPOP-2813
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 2d5f9fee630eeaa2b782060ed0db9eba851c0a17
Author: Stephen Mallette <st...@amazon.com>
AuthorDate: Mon Oct 17 08:47:01 2022 -0400
TINKERPOP-2813 Refactored job scheduling and thread pools
Went back to parallelized ConnectionPool creation and factored out that single threaded executor initialization which was put in long ago to prevent deadlock situations from using the default thread pool alone. Stopped using the fork-join pool for all jobs. Added a new thread pool distinguishing between Connection jobs and Host jobs to prevent deadlock problems. The single thread approach to the old scheduler thread pool seemed wrong because it seems possible that reconnect retries sc [...]
---
CHANGELOG.asciidoc | 3 ++
.../apache/tinkerpop/gremlin/driver/Client.java | 36 ++++++--------------
.../apache/tinkerpop/gremlin/driver/Cluster.java | 39 +++++++++++++++++-----
.../tinkerpop/gremlin/driver/Connection.java | 10 +++---
.../tinkerpop/gremlin/driver/ConnectionPool.java | 4 +--
.../org/apache/tinkerpop/gremlin/driver/Host.java | 2 +-
.../apache/tinkerpop/gremlin/driver/Settings.java | 1 -
.../tinkerpop/gremlin/driver/ClientTest.java | 2 +-
.../WebSocketClientBehaviorIntegrateTest.java | 2 +-
9 files changed, 54 insertions(+), 45 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 7212285da6..a95744a692 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -42,6 +42,9 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Bumped to Groovy 2.5.19.
* Bumped to Netty 4.1.85.
* Improved logging for `gremlin-driver`.
+* Modified `Connection` and `Host` job scheduling in `gremlin-driver` by dividing their work to two different thread pools and sparing work from the primary pool responsible for submitting requests and reading results.
+* Prevented usage of the fork-join pool for `gremlin-driver` job scheduling.
+* Changed `Host` initialization within a `Client` to be parallel again in `gremlin-driver`.
==== Bugs
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 89dfcc780e..bca2932d7a 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
@@ -18,7 +18,6 @@
*/
package org.apache.tinkerpop.gremlin.driver;
-import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,8 +45,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
@@ -518,26 +515,15 @@ public abstract class Client {
*/
@Override
protected void initializeImplementation() {
- // use a special executor here to initialize the Host instances as the worker thread pool may be
- // insufficiently sized for this task and the parallel initialization of the ConnectionPool. if too small
- // tasks may be schedule in such a way as to produce a deadlock: TINKERPOP-2550
- //
- // the cost of this single threaded executor here should be fairly small because it is only used once at
- // initialization and shutdown. since users will typically construct a Client once for the life of their
- // application there shouldn't be tons of thread pools being created and destroyed.
- final BasicThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("gremlin-driver-initializer").build();
- final ExecutorService hostExecutor = Executors.newSingleThreadExecutor(threadFactory);
-
try {
CompletableFuture.allOf(cluster.allHosts().stream()
- .map(host -> CompletableFuture.runAsync(() -> initializeConnectionSetupForHost.accept(host), hostExecutor))
+ .map(host -> CompletableFuture.runAsync(
+ () -> initializeConnectionSetupForHost.accept(host), cluster.hostScheduler()))
.toArray(CompletableFuture[]::new))
.join();
} catch (CompletionException ex) {
logger.error("Initialization failed", ex);
this.initializationFailure = ex;
- } finally {
- hostExecutor.shutdown();
}
// throw an error if there is no host available after initializing connection pool.
@@ -548,7 +534,7 @@ public abstract class Client {
final List<Host> unavailableHosts = cluster.allHosts()
.stream().filter(host -> !host.isAvailable()).collect(Collectors.toList());
if (!unavailableHosts.isEmpty()) {
- CompletableFuture.runAsync(() -> handleUnavailableHosts(unavailableHosts));
+ handleUnavailableHosts(unavailableHosts);
}
}
@@ -596,15 +582,15 @@ public abstract class Client {
}
};
- private void handleUnavailableHosts(List<Host> unavailableHosts) {
+ private void handleUnavailableHosts(final List<Host> unavailableHosts) {
// start the re-initialization attempt for each of the unavailable hosts through Host.makeUnavailable().
- try {
- CompletableFuture.allOf(unavailableHosts.stream()
- .map(host -> CompletableFuture.runAsync(() -> host.makeUnavailable(this::tryReInitializeHost)))
- .toArray(CompletableFuture[]::new))
- .join();
- } catch (CompletionException ex) {
- logger.error("", (ex.getCause() == null) ? ex : ex.getCause());
+ for (Host host : unavailableHosts) {
+ final CompletableFuture<Void> f = CompletableFuture.runAsync(
+ () -> host.makeUnavailable(this::tryReInitializeHost), cluster.hostScheduler());
+ f.exceptionally(t -> {
+ logger.error("", (t.getCause() == null) ? t : t.getCause());
+ return null;
+ });
}
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
index adc20f7063..e93074d3fc 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
@@ -34,7 +34,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.net.ssl.KeyManagerFactory;
-import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;
import java.io.File;
@@ -476,8 +475,12 @@ public final class Cluster {
return manager.executor;
}
- ScheduledExecutorService scheduler() {
- return manager.scheduler;
+ ScheduledExecutorService hostScheduler() {
+ return manager.hostScheduler;
+ }
+
+ ScheduledExecutorService connectionScheduler() {
+ return manager.connectionScheduler;
}
Settings.ConnectionPoolSettings connectionPoolSettings() {
@@ -1038,8 +1041,20 @@ public final class Cluster {
private final Supplier<RequestMessage.Builder> validationRequest;
private final HandshakeInterceptor interceptor;
+ /**
+ * Thread pool for requests.
+ */
private final ScheduledThreadPoolExecutor executor;
- private final ScheduledThreadPoolExecutor scheduler;
+
+ /**
+ * Thread pool for background work related to the {@link Host}.
+ */
+ private final ScheduledThreadPoolExecutor hostScheduler;
+
+ /**
+ * Thread pool for background work related to the {@link Connection} and {@link ConnectionPool}.
+ */
+ private final ScheduledThreadPoolExecutor connectionScheduler;
private final int nioPoolSize;
private final int workerPoolSize;
@@ -1102,9 +1117,14 @@ public final class Cluster {
// the executor above should be reserved for reading/writing background tasks that wont interfere with each
// other if the thread pool is 1 otherwise tasks may be schedule in such a way as to produce a deadlock
// as in TINKERPOP-2550. not sure if there is a way to only require the worker pool for all of this. as it
- // sits now the worker pool probably doesn't need to be a scheduled executor type
- this.scheduler = new ScheduledThreadPoolExecutor(1,
- new BasicThreadFactory.Builder().namingPattern("gremlin-driver-scheduler").build());
+ // sits now the worker pool probably doesn't need to be a scheduled executor type.
+ this.hostScheduler = new ScheduledThreadPoolExecutor(contactPoints.size() + 1,
+ new BasicThreadFactory.Builder().namingPattern("gremlin-driver-host-scheduler-%d").build());
+
+ // we distinguish between the hostScheduler and the connectionScheduler because you can end in deadlock
+ // if all the possible jobs the driver allows for go to a single thread pool.
+ this.connectionScheduler = new ScheduledThreadPoolExecutor(contactPoints.size() + 1,
+ new BasicThreadFactory.Builder().namingPattern("gremlin-driver-conn-scheduler-%d").build());
validationRequest = () -> RequestMessage.build(Tokens.OPS_EVAL).add(Tokens.ARGS_GREMLIN, builder.validationRequest);
}
@@ -1209,14 +1229,15 @@ public final class Cluster {
final CompletableFuture<Void> closeIt = new CompletableFuture<>();
closeFuture.set(closeIt);
- scheduler.submit(() -> {
+ hostScheduler.submit(() -> {
factory.shutdown();
closeIt.complete(null);
});
// Prevent the executor from accepting new tasks while still allowing enqueued tasks to complete
executor.shutdown();
- scheduler.shutdown();
+ connectionScheduler.shutdown();
+ hostScheduler.shutdown();
return closeIt;
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
index e2f5717714..81dd440891 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
@@ -131,14 +131,14 @@ final class Connection {
"Server closed the Connection on channel %s - scheduling removal from %s",
channel.id().asShortText(), thisConnection.pool.getPoolInfo(thisConnection)));
- // delegate the task to worker thread and free up the event loop
- thisConnection.cluster.executor().submit(() -> thisConnection.pool.definitelyDestroyConnection(thisConnection));
+ // delegate the task to scheduler thread and free up the event loop
+ thisConnection.cluster.connectionScheduler().submit(() -> thisConnection.pool.definitelyDestroyConnection(thisConnection));
}
});
logger.info("Created new connection for {}", uri);
} catch (Exception ex) {
- throw new ConnectionException(uri, "Could not open " + this.toString(), ex);
+ throw new ConnectionException(uri, "Could not open " + this.getConnectionInfo(true), ex);
}
}
@@ -201,7 +201,7 @@ final class Connection {
shutdown(future);
} else {
// there may be some pending requests. schedule a job to wait for those to complete and then shutdown
- new CheckForPending(future).runUntilDone(cluster.executor());
+ new CheckForPending(future).runUntilDone(cluster.connectionScheduler());
}
return future;
@@ -376,7 +376,7 @@ final class Connection {
public String getConnectionInfo(final boolean showHost) {
return showHost ?
String.format("Connection{channel=%s, host=%s, isDead=%s, borrowed=%s, pending=%s}",
- channel.id().asShortText(), pool.host, isDead(), borrowed, pending.size()) :
+ getChannelId(), pool.host, isDead(), borrowed, pending.size()) :
String.format("Connection{channel=%s, isDead=%s, borrowed=%s, pending=%s}",
channel.id().asShortText(), isDead(), borrowed, pending.size());
}
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 f1d8f69eb7..b1a9632738 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
@@ -105,7 +105,7 @@ final class ConnectionPool {
} catch (ConnectionException e) {
throw new CompletionException(e);
}
- }, cluster.executor()));
+ }, cluster.connectionScheduler()));
}
CompletableFuture.allOf(connectionCreationFutures.toArray(new CompletableFuture[0])).join();
@@ -311,7 +311,7 @@ final class ConnectionPool {
}
private void newConnection() {
- cluster.executor().submit(() -> {
+ cluster.connectionScheduler().submit(() -> {
addConnectionIfUnderMaximum();
scheduledForCreation.decrementAndGet();
return null;
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Host.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Host.java
index afac54ce5e..fb0a9e333a 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Host.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Host.java
@@ -77,7 +77,7 @@ public final class Host {
// only do a connection re-attempt if one is not already in progress
if (retryInProgress.compareAndSet(Boolean.FALSE, Boolean.TRUE)) {
- retryThread = this.cluster.scheduler().scheduleAtFixedRate(() -> {
+ retryThread = this.cluster.hostScheduler().scheduleAtFixedRate(() -> {
logger.debug("Trying to reconnect to dead host at {}", this);
if (reconnect.apply(this)) reconnected();
}, cluster.connectionPoolSettings().reconnectInterval,
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java
index 0e8ae0fdf3..38608769ca 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java
@@ -395,7 +395,6 @@ final class Settings {
public String validationRequest = "''";
/**
- *
* Duration of time in milliseconds provided for connection setup to complete which includes WebSocket
* handshake and SSL handshake. Beyond this duration an exception would be thrown if the handshake is not
* complete by then.
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientTest.java
index 1ca2aeb845..1ab61b3cfc 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientTest.java
@@ -58,7 +58,7 @@ public class ClientTest {
when(mockAvailableHost.isAvailable()).thenReturn(true);
when(cluster.allHosts()).thenReturn(Collections.singletonList(mockAvailableHost));
when(cluster.executor()).thenReturn(executor);
- when(cluster.scheduler()).thenReturn(scheduler);
+ when(cluster.hostScheduler()).thenReturn(scheduler);
}
@After
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 0eb34f6867..549aa75855 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
@@ -96,7 +96,7 @@ public class WebSocketClientBehaviorIntegrateTest {
*/
@Test
public void shouldNotDeadlockOnInitialization() throws Exception {
- // it seems you cah add the same host more than once so while kinda weird it is helpful in faithfully
+ // it seems you can add the same host more than once so while kinda weird it is helpful in faithfully
// recreating the deadlock situation, though it can/will happen with just one host. workerPoolSize at
// "1" also helps faithfully reproduce the problem though it can happen at larger pool sizes depending
// on the timing/interleaving of tasks. the larger connection pool sizes may not be required given the
[tinkerpop] 03/05: TINKERPOP-2813 Removed fast NoHostAvailableException
Posted by sp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch TINKERPOP-2813
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 96274275a84411ae2d06acf6b3eca47ee5ca1314
Author: Stephen Mallette <st...@amazon.com>
AuthorDate: Fri Nov 4 15:16:19 2022 -0400
TINKERPOP-2813 Removed fast NoHostAvailableException
Took a more optimistic approach to determining Host availability which prevents what may be intermittent network/server problems from being interpreted as the Host not being reachable. Takes a faster approach to reconnecting when a host is found to "maybe" be unavailable.
---
CHANGELOG.asciidoc | 3 +
.../apache/tinkerpop/gremlin/driver/Client.java | 27 +++--
.../tinkerpop/gremlin/driver/Connection.java | 66 ++++++-----
.../gremlin/driver/ConnectionFactory.java | 41 +++++++
.../tinkerpop/gremlin/driver/ConnectionPool.java | 123 +++++++++++++++------
.../org/apache/tinkerpop/gremlin/driver/Host.java | 18 +++
.../driver/exception/NoHostAvailableException.java | 8 +-
.../driver/ClientConnectionIntegrateTest.java | 108 ++++++++++++++++++
.../AbstractGremlinServerIntegrationTest.java | 8 +-
.../gremlin/server/GremlinDriverIntegrateTest.java | 2 -
10 files changed, 328 insertions(+), 76 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index a95744a692..c026be3a76 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -45,6 +45,9 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Modified `Connection` and `Host` job scheduling in `gremlin-driver` by dividing their work to two different thread pools and sparing work from the primary pool responsible for submitting requests and reading results.
* Prevented usage of the fork-join pool for `gremlin-driver` job scheduling.
* Changed `Host` initialization within a `Client` to be parallel again in `gremlin-driver`.
+* Changed mechanism for determining `Host` health which should make the driver more resilient to intermittent network failures.
+* Removed the delay for reconnecting to a potentially unhealthy `Host` only marking it as unavailable after that initial retry fails.
+* Prevented fast `NoHostAvailableException` in favor of more direct exceptions when borrowing connections from the `ConnectionPool`.
==== Bugs
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 bca2932d7a..0226d32e18 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
@@ -40,6 +40,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -50,6 +51,7 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
/**
* A {@code Client} is constructed from a {@link Cluster} and represents a way to send messages to Gremlin Server.
@@ -68,6 +70,8 @@ public abstract class Client {
protected volatile boolean initialized;
protected final Client.Settings settings;
+ private static final Random random = new Random();
+
Client(final Cluster cluster, final Client.Settings settings) {
this.cluster = cluster;
this.settings = settings;
@@ -423,7 +427,7 @@ public abstract class Client {
*/
public final static class ClusteredClient extends Client {
- protected ConcurrentMap<Host, ConnectionPool> hostConnectionPools = new ConcurrentHashMap<>();
+ ConcurrentMap<Host, ConnectionPool> hostConnectionPools = new ConcurrentHashMap<>();
private final AtomicReference<CompletableFuture<Void>> closing = new AtomicReference<>(null);
private Throwable initializationFailure = null;
@@ -492,7 +496,11 @@ public abstract class Client {
protected Connection chooseConnection(final RequestMessage msg) throws TimeoutException, ConnectionException {
final Iterator<Host> possibleHosts;
if (msg.optionalArgs(Tokens.ARGS_HOST).isPresent()) {
- // TODO: not sure what should be done if unavailable - select new host and re-submit traversal?
+ // looking at this code about putting the Host on the RequestMessage in light of 3.5.4, not sure
+ // this is being used as intended here. server side usage is to place the channel.remoteAddress
+ // in this token in the status metadata for the response. can't remember why it is being used this
+ // way here exactly. created TINKERPOP-2821 to examine this more carefully to clean this up in a
+ // future version.
final Host host = (Host) msg.getArgs().get(Tokens.ARGS_HOST);
msg.getArgs().remove(Tokens.ARGS_HOST);
possibleHosts = IteratorUtils.of(host);
@@ -500,16 +508,19 @@ public abstract class Client {
possibleHosts = this.cluster.loadBalancingStrategy().select(msg);
}
- // you can get no possible hosts in more than a few situations. perhaps the servers are just all down.
- // or perhaps the client is not configured properly (disables ssl when ssl is enabled on the server).
- if (!possibleHosts.hasNext())
- throwNoHostAvailableException();
-
- final Host bestHost = possibleHosts.next();
+ // try a random host if none are marked available. maybe it will reconnect in the meantime. better than
+ // going straight to a fast NoHostAvailableException as was the case in versions 3.5.4 and earlier
+ final Host bestHost = possibleHosts.hasNext() ? possibleHosts.next() : chooseRandomHost();
final ConnectionPool pool = hostConnectionPools.get(bestHost);
return pool.borrowConnection(cluster.connectionPoolSettings().maxWaitForConnection, TimeUnit.MILLISECONDS);
}
+ private Host chooseRandomHost() {
+ final List<Host> hosts = new ArrayList<>(cluster.allHosts());
+ final int ix = random.nextInt(hosts.size());
+ return hosts.get(ix);
+ }
+
/**
* Initializes the connection pools on all hosts.
*/
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
index 81dd440891..0dfe0dbab3 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
@@ -114,11 +114,8 @@ final class Connection {
channel = b.connect(uri.getHost(), uri.getPort()).sync().channel();
channelizer.connected();
- /* Configure behaviour on close of this channel.
- *
- * This callback would trigger the workflow to destroy this connection, so that a new request doesn't pick
- * this closed connection.
- */
+ // Configure behaviour on close of this channel. This callback would trigger the workflow to destroy this
+ // connection, so that a new request doesn't pick this closed connection.
final Connection thisConnection = this;
channel.closeFuture().addListener((ChannelFutureListener) future -> {
logger.debug("OnChannelClose callback called for channel {}", channel);
@@ -138,7 +135,7 @@ final class Connection {
logger.info("Created new connection for {}", uri);
} catch (Exception ex) {
- throw new ConnectionException(uri, "Could not open " + this.getConnectionInfo(true), ex);
+ throw new ConnectionException(uri, "Could not open " + getConnectionInfo(true), ex);
}
}
@@ -316,6 +313,9 @@ final class Connection {
RequestMessage.build(Tokens.OPS_CLOSE).addArg(Tokens.ARGS_FORCE, forceClose)).create();
final CompletableFuture<ResultSet> closed = new CompletableFuture<>();
+
+ // TINKERPOP-2822 should investigate this write more carefully to check for sensible behavior
+ // in the event the Channel was not created but we try to send the close message
write(closeMessage, closed);
try {
@@ -336,27 +336,37 @@ final class Connection {
}
}
- channelizer.close(channel);
+ // take a defensive posture here in the event the channelizer didn't get initialized somehow and a
+ // close() on the Connection is still called
+ if (channelizer != null)
+ channelizer.close(channel);
+
+ // seems possible that the channelizer could initialize but fail to produce a channel, so worth checking
+ // null before proceeding here
+ if (channel != null) {
+ final ChannelPromise promise = channel.newPromise();
+ promise.addListener(f -> {
+ if (f.cause() != null) {
+ future.completeExceptionally(f.cause());
+ } else {
+ if (logger.isDebugEnabled())
+ logger.debug("{} destroyed successfully.", connectionInfo);
- final ChannelPromise promise = channel.newPromise();
- promise.addListener(f -> {
- if (f.cause() != null) {
- future.completeExceptionally(f.cause());
- } else {
- if (logger.isDebugEnabled())
- logger.debug("{} destroyed successfully.", connectionInfo);
+ future.complete(null);
+ }
+ });
- future.complete(null);
+ // close the netty channel, if not already closed
+ if (!channel.closeFuture().isDone()) {
+ channel.close(promise);
+ } else {
+ if (!promise.trySuccess()) {
+ logger.warn("Failed to mark a promise as success because it is done already: {}", promise);
+ }
}
- });
-
- // close the netty channel, if not already closed
- if (!channel.closeFuture().isDone()) {
- channel.close(promise);
} else {
- if (!promise.trySuccess()) {
- logger.warn("Failed to mark a promise as success because it is done already: {}", promise);
- }
+ logger.debug("Connection {} is shutting down but the channel was not initialized to begin with",
+ getConnectionInfo());
}
}
}
@@ -365,7 +375,7 @@ final class Connection {
* Gets a message that describes the state of the connection.
*/
public String getConnectionInfo() {
- return getConnectionInfo(true);
+ return this.getConnectionInfo(true);
}
/**
@@ -375,10 +385,10 @@ final class Connection {
*/
public String getConnectionInfo(final boolean showHost) {
return showHost ?
- String.format("Connection{channel=%s, host=%s, isDead=%s, borrowed=%s, pending=%s}",
- getChannelId(), pool.host, isDead(), borrowed, pending.size()) :
- String.format("Connection{channel=%s, isDead=%s, borrowed=%s, pending=%s}",
- channel.id().asShortText(), isDead(), borrowed, pending.size());
+ String.format("Connection{channel=%s host=%s isDead=%s borrowed=%s pending=%s markedReplaced=%s closing=%s}",
+ getChannelId(), pool.host.toString(), isDead(), this.borrowed.get(), getPending().size(), this.isBeingReplaced, isClosing()) :
+ String.format("Connection{channel=%s isDead=%s borrowed=%s pending=%s markedReplaced=%s closing=%s}",
+ getChannelId(), isDead(), this.borrowed.get(), getPending().size(), this.isBeingReplaced, isClosing());
}
/**
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionFactory.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionFactory.java
new file mode 100644
index 0000000000..0771402c70
--- /dev/null
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ConnectionFactory.java
@@ -0,0 +1,41 @@
+/*
+ * 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 org.apache.tinkerpop.gremlin.driver.exception.ConnectionException;
+
+/**
+ * A factory that is responsible for creating a {@link Connection} instance. The {@link DefaultConnectionFactory}
+ * simply news up a {@code Connection} using its default constructor. This interface is mostly present to help
+ * enable better testing of the driver internals and likely shouldn't be used otherwise.
+ */
+interface ConnectionFactory {
+
+ /**
+ * Create a connection for the specified {@link ConnectionPool}.
+ */
+ public default Connection create(final ConnectionPool pool) throws ConnectionException {
+ return new Connection(pool.host.getHostUri(), pool, pool.settings().maxInProcessPerConnection);
+ }
+
+ /**
+ * Default implementation.
+ */
+ public static class DefaultConnectionFactory implements ConnectionFactory { }
+}
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 b1a9632738..4e5f019273 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
@@ -67,12 +67,12 @@ final class ConnectionPool {
private final String poolLabel;
private final AtomicInteger scheduledForCreation = new AtomicInteger();
-
private final AtomicReference<CompletableFuture<Void>> closeFuture = new AtomicReference<>();
private volatile int waiter = 0;
private final Lock waitLock = new ReentrantLock(true);
private final Condition hasAvailableConnection = waitLock.newCondition();
+ ConnectionFactory connectionFactory;
public ConnectionPool(final Host host, final Client client) {
this(host, client, Optional.empty(), Optional.empty());
@@ -80,9 +80,15 @@ final class ConnectionPool {
public ConnectionPool(final Host host, final Client client, final Optional<Integer> overrideMinPoolSize,
final Optional<Integer> overrideMaxPoolSize) {
+ this(host, client, overrideMinPoolSize, overrideMaxPoolSize, new ConnectionFactory.DefaultConnectionFactory());
+ }
+
+ ConnectionPool(final Host host, final Client client, final Optional<Integer> overrideMinPoolSize,
+ final Optional<Integer> overrideMaxPoolSize, final ConnectionFactory connectionFactory) {
this.host = host;
this.client = client;
this.cluster = client.cluster;
+ this.connectionFactory = connectionFactory;
poolLabel = "Connection Pool {host=" + host + "}";
final Settings.ConnectionPoolSettings settings = settings();
@@ -100,7 +106,7 @@ final class ConnectionPool {
for (int i = 0; i < minPoolSize; i++) {
connectionCreationFutures.add(CompletableFuture.runAsync(() -> {
try {
- this.connections.add(new Connection(host.getHostUri(), this, settings.maxInProcessPerConnection));
+ this.connections.add(connectionFactory.create(this));
this.open.incrementAndGet();
} catch (ConnectionException e) {
throw new CompletionException(e);
@@ -110,7 +116,7 @@ final class ConnectionPool {
CompletableFuture.allOf(connectionCreationFutures.toArray(new CompletableFuture[0])).join();
} catch (CancellationException ce) {
- logger.warn("Initialization of connections cancelled for {}", getPoolInfo(), ce);
+ logger.warn("Initialization of connections cancelled for {}", this.getPoolInfo(), ce);
throw ce;
} catch (CompletionException ce) {
// Some connections might have been initialized. Close the connection pool gracefully to close them.
@@ -312,20 +318,31 @@ final class ConnectionPool {
private void newConnection() {
cluster.connectionScheduler().submit(() -> {
- addConnectionIfUnderMaximum();
+ // seems like this should be decremented first because if addConnectionIfUnderMaximum fails there is
+ // nothing that wants to decrement this number and so it leaves things in a state where you could
+ // newConnection() doesn't seem to get called at all because it believes connections are being currently
+ // created. this seems to lead to situations where the client can never borrow a connection and it's as
+ // though it can't reconnect at all. this was hard to test but it seemed to happen regularly after
+ // introduced the ConnectionFactory that enabled a way to introduce connection jitters (i.e. failures in
+ // creation of a Connection) at which point it seemed to happen with some regularity.
scheduledForCreation.decrementAndGet();
+ addConnectionIfUnderMaximum();
return null;
});
}
private boolean addConnectionIfUnderMaximum() {
+ final int openCountToActOn;
+
while (true) {
- int opened = open.get();
+ final int opened = open.get();
if (opened >= maxPoolSize)
return false;
- if (open.compareAndSet(opened, opened + 1))
+ if (open.compareAndSet(opened, opened + 1)) {
+ openCountToActOn = opened;
break;
+ }
}
if (isClosed()) {
@@ -334,10 +351,13 @@ final class ConnectionPool {
}
try {
- connections.add(new Connection(host.getHostUri(), this, settings().maxInProcessPerConnection));
- } catch (ConnectionException ce) {
- logger.error("Connections were under max, but there was an error creating the connection.", ce);
+ connections.add(connectionFactory.create(this));
+ } catch (Exception ce) {
open.decrementAndGet();
+ logger.error(String.format(
+ "Connections[%s] were under maximum allowed[%s], but there was an error creating a new connection",
+ openCountToActOn, maxPoolSize),
+ ce);
considerHostUnavailable();
return false;
}
@@ -432,20 +452,33 @@ final class ConnectionPool {
// if we timeout borrowing a connection that might mean the host is dead (or the timeout was super short).
// either way supply a function to reconnect
+ final TimeoutException timeoutException = new TimeoutException(timeoutErrorMessage);
this.considerHostUnavailable();
- throw new TimeoutException(timeoutErrorMessage);
+ throw timeoutException;
}
+ /**
+ * On a failure to get a {@link Connection} this method is called to determine if the {@link Host} should be
+ * marked as unavailable and to establish a background reconnect operation.
+ */
public void considerHostUnavailable() {
- // called when a connection is "dead" due to a non-recoverable error.
- host.makeUnavailable(this::tryReconnect);
-
- // if the host is unavailable then we should release the connections
- connections.forEach(this::definitelyDestroyConnection);
-
- // let the load-balancer know that the host is acting poorly
- this.cluster.loadBalancingStrategy().onUnavailable(host);
+ // if there is at least one available connection the host has to still be around (or is perhaps on its way out
+ // but we'll stay optimistic in this check). no connections also means "unhealthy". unsure if there is an ok
+ // "no connections" state we can get into. in any event if there are no connections then we'd just try to
+ // immediately reconnect below anyway so perhaps that state isn't really something to worry about.
+ final boolean maybeUnhealthy = connections.stream().allMatch(Connection::isDead);
+ if (maybeUnhealthy) {
+ // immediately fire off an attempt to reconnect because there are no active connections.
+ host.tryReconnectingImmediately(this::tryReconnect);
+
+ // let the load-balancer know that the host is acting poorly
+ if (!host.isAvailable()) {
+ // if the host is unavailable then we should release the connections
+ connections.forEach(this::definitelyDestroyConnection);
+ this.cluster.loadBalancingStrategy().onUnavailable(host);
+ }
+ }
}
/**
@@ -457,7 +490,13 @@ final class ConnectionPool {
Connection connection = null;
try {
- connection = borrowConnection(cluster.connectionPoolSettings().maxWaitForConnection, TimeUnit.MILLISECONDS);
+ // rather than rely on borrowConnection() infrastructure and the pool create a brand new Connection
+ // instance solely for the purpose of this ping. this ensures that if the pool is overloaded that we
+ // make an honest attempt at validating host health without failing over some timeout waiting for a
+ // connection in the pool. not sure if we should try to keep this connection if it succeeds and if the
+ // pool needs it. for now that seems like an unnecessary added bit of complexity for dealing with this
+ // error state
+ connection = connectionFactory.create(this);
final RequestMessage ping = client.buildMessage(cluster.validationRequest()).create();
final CompletableFuture<ResultSet> f = new CompletableFuture<>();
connection.write(ping, f);
@@ -467,9 +506,13 @@ final class ConnectionPool {
this.cluster.loadBalancingStrategy().onAvailable(h);
return true;
} catch (Exception ex) {
- logger.debug("Failed reconnect attempt on {}", h, ex);
- if (connection != null) definitelyDestroyConnection(connection);
+ logger.error(String.format("Failed reconnect attempt on %s%s%s",
+ h, System.lineSeparator(), this.getPoolInfo()), ex);
return false;
+ } finally {
+ if (connection != null) {
+ connection.closeAsync();
+ }
}
}
@@ -547,7 +590,7 @@ final class ConnectionPool {
*/
public String getPoolInfo(final Connection connectionToCallout) {
final StringBuilder sb = new StringBuilder("ConnectionPool (");
- sb.append(host);
+ sb.append(host.toString());
sb.append(")");
if (connections.isEmpty()) {
@@ -555,25 +598,35 @@ final class ConnectionPool {
} else {
final int connectionCount = connections.size();
sb.append(System.lineSeparator());
- sb.append(String.format("Connection Pool Status (size=%s max=%s min=%s)", connectionCount, maxPoolSize, minPoolSize));
+ sb.append(String.format("Connection Pool Status (size=%s max=%s min=%s toCreate=%s markedOpen=%s bin=%s)",
+ connectionCount, maxPoolSize, minPoolSize, this.scheduledForCreation.get(), this.open.get(), bin.size()));
+ sb.append(System.lineSeparator());
+
+ appendConnections(sb, connectionToCallout, connections);
+ sb.append(System.lineSeparator());
+ sb.append("-- bin --");
sb.append(System.lineSeparator());
+ appendConnections(sb, connectionToCallout, new ArrayList<>(bin));
+ }
- for (int ix = 0; ix < connectionCount; ix++) {
- final Connection c = connections.get(ix);
+ return sb.toString().trim();
+ }
- if (c == connectionToCallout)
- sb.append("==> ");
- else
- sb.append("> ");
+ private void appendConnections(final StringBuilder sb, final Connection connectionToCallout,
+ final List<Connection> connections) {
+ final int connectionCount = connections.size();
+ for (int ix = 0; ix < connectionCount; ix++) {
+ final Connection c = connections.get(ix);
+ if (c.equals(connectionToCallout))
+ sb.append("==> ");
+ else
+ sb.append("> ");
- sb.append(c.getConnectionInfo(false));
+ sb.append(c.getConnectionInfo(false));
- if (ix < connectionCount - 1)
- sb.append(System.lineSeparator());
- }
+ if (ix < connectionCount - 1)
+ sb.append(System.lineSeparator());
}
-
- return sb.toString().trim();
}
@Override
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Host.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Host.java
index fb0a9e333a..3d492fa612 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Host.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Host.java
@@ -26,6 +26,7 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
@@ -85,6 +86,23 @@ public final class Host {
}
}
+ void tryReconnectingImmediately(final Function<Host, Boolean> reconnect) {
+ // only do a connection re-attempt if one is not already in progress
+ if (retryInProgress.compareAndSet(Boolean.FALSE, Boolean.TRUE)) {
+ retryThread = this.cluster.hostScheduler().scheduleAtFixedRate(() -> {
+ logger.debug("Trying to reconnect to host at {}", this);
+ final boolean reconnected = reconnect.apply(this);
+ if (reconnected)
+ reconnected();
+ else {
+ logger.warn("Marking {} as unavailable. Trying to reconnect.", this);
+ isAvailable = false;
+ }
+ }, 0,
+ cluster.connectionPoolSettings().reconnectInterval, TimeUnit.MILLISECONDS);
+ }
+ }
+
private void reconnected() {
// race condition! retry boolean could be set to false, a new retryThread created above
// and then cancelled here. But we're only executing this at all because we *have* reconnected
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/NoHostAvailableException.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/NoHostAvailableException.java
index 1d2a70f7ea..9ccea2dffa 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/NoHostAvailableException.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/exception/NoHostAvailableException.java
@@ -21,10 +21,14 @@ package org.apache.tinkerpop.gremlin.driver.exception;
public class NoHostAvailableException extends RuntimeException {
public NoHostAvailableException() {
- super("All hosts are considered unavailable due to previous exceptions. Check the error log to find the actual reason.");
+ this("All hosts are considered unavailable due to previous exceptions. Check the error log to find the actual reason.");
}
- public NoHostAvailableException(Throwable ex) {
+ public NoHostAvailableException(final String message) {
+ super(message);
+ }
+
+ public NoHostAvailableException(final Throwable ex) {
super(ex);
}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java
index 8ebad7809e..1417409c47 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/ClientConnectionIntegrateTest.java
@@ -21,6 +21,8 @@ package org.apache.tinkerpop.gremlin.driver;
import io.netty.handler.codec.CorruptedFrameException;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
+import org.apache.tinkerpop.gremlin.driver.exception.ConnectionException;
+import org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
import org.apache.tinkerpop.gremlin.driver.ser.Serializers;
import org.apache.tinkerpop.gremlin.server.AbstractGremlinServerIntegrationTest;
import org.apache.tinkerpop.gremlin.server.TestClientFactory;
@@ -28,13 +30,25 @@ import org.apache.tinkerpop.gremlin.util.Log4jRecordingAppender;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.SSLHandshakeException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.IntStream;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.number.OrderingComparison.greaterThan;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrationTest {
+ private static final org.slf4j.Logger logger = LoggerFactory.getLogger(ClientConnectionIntegrateTest.class);
private Log4jRecordingAppender recordingAppender = null;
private Level previousLogLevel;
@@ -110,4 +124,98 @@ public class ClientConnectionIntegrateTest extends AbstractGremlinServerIntegrat
assertThat(recordingAppender.logContainsAny("^(?!.*(isDead=false)).*isDead=true.*destroyed successfully.$"), is(true));
}
+
+ /**
+ * Added for TINKERPOP-2813 - this scenario would have previously thrown tons of
+ * {@link NoHostAvailableException}.
+ */
+ @Test
+ public void shouldSucceedWithJitteryConnection() throws Exception {
+ final Cluster cluster = TestClientFactory.build().minConnectionPoolSize(1).maxConnectionPoolSize(4).
+ reconnectInterval(1000).
+ maxWaitForConnection(4000).validationRequest("g.inject()").create();
+ final Client.ClusteredClient client = cluster.connect();
+
+ client.init();
+
+ // every 10 connections let's have some problems
+ final JitteryConnectionFactory connectionFactory = new JitteryConnectionFactory(3);
+ client.hostConnectionPools.forEach((h, pool) -> pool.connectionFactory = connectionFactory);
+
+ // get an initial connection which marks the host as available
+ assertEquals(2, client.submit("1+1").all().join().get(0).getInt());
+
+ // network is gonna get fishy - ConnectionPool should try to grow during the workload below and when it
+ // does some connections will fail to create in the background which should log some errors but not tank
+ // the submit() as connections that are currently still working and active should be able to handle the load.
+ connectionFactory.jittery = true;
+
+ // load up a hella ton of requests
+ final int requests = 1000;
+ final CountDownLatch latch = new CountDownLatch(requests);
+ final AtomicBoolean hadFailOtherThanTimeout = new AtomicBoolean(false);
+
+ new Thread(() -> {
+ IntStream.range(0, requests).forEach(i -> {
+ try {
+ client.submitAsync("1 + " + i);
+ } catch (Exception ex) {
+ // we could catch a TimeoutException here in some cases if the jitters cause a borrow of a
+ // connection to take too long. submitAsync() will wrap in a RuntimeException. can't assert
+ // this condition inside this thread or the test locks up
+ hadFailOtherThanTimeout.compareAndSet(false, !(ex.getCause() instanceof TimeoutException));
+ } finally {
+ latch.countDown();
+ }
+ });
+ }, "worker-shouldSucceedWithJitteryConnection").start();
+
+ // wait long enough for the jitters to kick in at least a little
+ while (latch.getCount() > 500) {
+ TimeUnit.MILLISECONDS.sleep(50);
+ }
+
+ // wait for requests to complete
+ assertTrue(latch.await(30000, TimeUnit.MILLISECONDS));
+
+ // make sure we had some failures for sure coming out the factory
+ assertThat(connectionFactory.getNumberOfFailures(), is(greaterThan(0L)));
+
+ // if there was a exception in the worker thread, then it had better be a TimeoutException
+ assertThat(hadFailOtherThanTimeout.get(), is(false));
+
+ cluster.close();
+ }
+
+ /**
+ * Introduces random failures when creating a {@link Connection} for the {@link ConnectionPool}.
+ */
+ public static class JitteryConnectionFactory implements ConnectionFactory {
+
+ private volatile boolean jittery = false;
+ private final AtomicLong connectionsCreated = new AtomicLong(0);
+ private final AtomicLong connectionFailures = new AtomicLong(0);
+ private final int numberOfConnectionsBetweenErrors;
+
+ public JitteryConnectionFactory(final int numberOfConnectionsBetweenErrors) {
+ this.numberOfConnectionsBetweenErrors = numberOfConnectionsBetweenErrors;
+ }
+
+ public long getNumberOfFailures() {
+ return connectionFailures.get();
+ }
+
+ @Override
+ public Connection create(final ConnectionPool pool) {
+
+ // fail creating a connection every 10 attempts or so when jittery
+ if (jittery && connectionsCreated.incrementAndGet() % numberOfConnectionsBetweenErrors == 0) {
+ connectionFailures.incrementAndGet();
+ throw new ConnectionException(pool.host.getHostUri(),
+ new SSLHandshakeException("SSL on the funk - server is big mad"));
+ }
+
+ return ConnectionFactory.super.create(pool);
+ }
+ }
}
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 2b66ce1de1..6a8acbad15 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
@@ -21,6 +21,7 @@ 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.TinkerGraph;
import org.junit.After;
import org.junit.Before;
@@ -31,6 +32,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.InputStream;
+import java.util.concurrent.CompletableFuture;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assume.assumeThat;
@@ -117,6 +119,10 @@ public abstract class AbstractGremlinServerIntegrationTest {
}
public void startServer() throws Exception {
+ startServerAsync().join();
+ }
+
+ public CompletableFuture<ServerGremlinExecutor> startServerAsync() throws Exception {
final InputStream stream = getSettingsInputStream();
final Settings settings = Settings.read(stream);
overriddenSettings = overrideSettings(settings);
@@ -131,7 +137,7 @@ public abstract class AbstractGremlinServerIntegrationTest {
this.server = new GremlinServer(overriddenSettings);
- server.start().join();
+ return server.start();
}
@After
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 e44660ee50..43a53bae7a 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
@@ -55,7 +55,6 @@ import org.apache.tinkerpop.gremlin.util.function.FunctionUtils;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import org.junit.After;
import org.junit.Before;
-import org.junit.Ignore;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
@@ -63,7 +62,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.awt.Color;
-import java.io.File;
import java.net.ConnectException;
import java.time.Instant;
import java.util.ArrayList;
[tinkerpop] 05/05: TINKERPOP-2813 Updated upgrade docs and reference docs
Posted by sp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch TINKERPOP-2813
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 1c85f3770e47880cfebb428aa541b59e7e041500
Author: Stephen Mallette <st...@amazon.com>
AuthorDate: Thu Dec 1 13:27:03 2022 -0500
TINKERPOP-2813 Updated upgrade docs and reference docs
---
docs/src/reference/gremlin-variants.asciidoc | 25 +++++++++++++++++++
docs/src/upgrade/release-3.5.x.asciidoc | 37 ++++++++++++++++++++++++++++
2 files changed, 62 insertions(+)
diff --git a/docs/src/reference/gremlin-variants.asciidoc b/docs/src/reference/gremlin-variants.asciidoc
index 545c872b86..2e5134d626 100644
--- a/docs/src/reference/gremlin-variants.asciidoc
+++ b/docs/src/reference/gremlin-variants.asciidoc
@@ -657,6 +657,31 @@ social.persons("marko").knows("josh");
NOTE: Using Maven, as shown in the `gremlin-archetype-dsl` module, makes developing DSLs with the annotation processor
straightforward in that it sets up appropriate paths to the generated code automatically.
+[[gremlin-java-troubleshooting]]
+=== Troubleshooting
+
+*Max frame length of 65536 has been exceeded*
+
+This error occurs when the driver attempts to process a request/response that exceeds the configured maximum size.
+The most direct way to fix this problem is to increase the `maxContentLength` setting in the driver. Ideally, the
+`maxContentLength` set for the driver should match the setting defined on the server.
+
+*TimeoutException*
+
+A `TimeoutException` is thrown by the driver when the time limit assigned by the `maxWaitForConnection` is exceeded
+when trying to borrow a connection from the connection pool for a particular host. There are generally two scenarios
+where this occurs:
+
+1. The server has actually reached its maximum capacity or the driver has just learned that the server is unreachable.
+2. The client is throttling requests.
+
+The latter of the two can be addressed from the driver side in the following ways:
+
+* Increase the `maxWaitForConnection` allowing the client to wait a bit longer for a connection to become available.
+* Increase the number of requests allowed per connection by increasing the `maxSimultaneousUsagePerConnection` and
+`maxInProcessPerConnection` settings.
+* Increase the number of connections available in the connection pool by increasing the `maxConnectionPoolSize`.
+
anchor:java-application-examples[]
anchor:gremlin-archetypes[]
[[gremlin-java-examples]]
diff --git a/docs/src/upgrade/release-3.5.x.asciidoc b/docs/src/upgrade/release-3.5.x.asciidoc
index 369c32ec84..52e2b66f53 100644
--- a/docs/src/upgrade/release-3.5.x.asciidoc
+++ b/docs/src/upgrade/release-3.5.x.asciidoc
@@ -50,6 +50,43 @@ var client = new GremlinClient(new GremlinServer("localhost", 8182), loggerFacto
See: link:https://issues.apache.org/jira/browse/TINKERPOP-2471[TINKERPOP-2471]
+==== gremlin-driver Host Availability
+
+The Java drivers approach to determine the health of the host to which it is connected is important to how it behaves.
+The driver has generally taken a pessimistic approach to make this determination, where a failure to borrow a
+connection from the pool would constitute enough evidence to mark the host as dead. Unfortunately a failure to borrow
+from the pool is not the best indicator for a dead host. Intermittent network failure, an overly busy client, or other
+temporary issues could have been at play while the server was perfectly available.
+
+The consequences for marking a host dead are fairly severe as the connection pool is shutdown. While it is shutdown
+in an orderly fashion so as to allow for existing requests to complete if at all possible, it immediately blocks new
+requests producing an immediate `NoHostAvailableException` (assuming there are actually no other hosts to route
+requests to). There is then some delay to reconnect to the server and re-establish each of the connections in the pool,
+which might take some time especially if there was a large pool. If an application were sending hundreds of request
+a second, it would quickly equate to hundreds of exceptions a second, filling logs quickly and making it hard to
+decipher the original root cause from that initial inability to simply borrow a connection.
+
+For 3.5.5, the driver has dropped some of its pessimistic ways and has taken a more optimistic posture when it comes
+to considering host availability. When the driver now has a problem borrowing a connection from the pool, it does not
+assume the worst of a failed host. It instead checks if the other connections in the pool are still alive. If at least
+one is alive it assumes the host is still available and the user simply gets a `TimeoutException` for waiting beyond
+the `maxWaitForConnection` setting. In the event that there are no connections deemed alive, the driver will try to
+issue an immediate reconnect to determine host health. If the reconnect succeeds then the host remains in an available
+state for future requests and if it fails it is finally marked as dead and thus unavailable.
+
+If a host is marked unavailable, the driver, like before, will try to route requests to other configured hosts that
+are healthy. If there are no such other hosts and all are unhealthy, the driver will now make an immediate attempt to
+route the request to a random unavailable host on the off chance it will come back online and the connection pool thus
+re-established. So, rather than a `NoHostAvailableException` flood, the user would instead see a `TimeoutException`
+per request made.
+
+Since the driver was already throwing `NoHostAvailableException` and `TimeoutException`, there shouldn't be too
+many code changes required on upgrade to this version. The main difference to consider is that
+`NoHostAvailableException` should appear less often, so any code that mitigated against a flood of these exceptions,
+such defenses might not be necessary anymore.
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-2813[TINKERPOP-2813]
+
== TinkerPop 3.5.4
*Release Date: July 18, 2022*