Skip to content

Commit ba7b771

Browse files
injectivesinjectives
committed
feat(acquisition): apply acquisition timeout to all steps
BREAKING CHANGE: `org.neo4j.driver.Config.ConfigBuilder#withConnectionAcquisitionTimeout(long, TimeUnit)` no longer accepts negative values, matching the handling of `org.neo4j.driver.Config.ConfigBuilder#withConnectionTimeout(long, TimeUnit)`. In addition, the acquisition timeout is applied for the entire connection acquisition process that may include client-side routing and other logic. Previously, the connection acquisition timeout applied individually per every acquisition from individual connection pool and only when all connections were busy.
1 parent 072087d commit ba7b771

File tree

10 files changed

+69
-64
lines changed

10 files changed

+69
-64
lines changed

‎driver/src/main/java/org/neo4j/driver/Config.java‎

Lines changed: 17 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -554,27 +554,34 @@ public ConfigBuilder withMaxConnectionPoolSize(int value) {
554554
}
555555

556556
/**
557-
* Configure maximum amount of time connection acquisition will attempt to acquire a connection from the
558-
* connection pool. This timeout only kicks in when all existing connections are being used and no new
559-
* connections can be created because maximum connection pool size has been reached.
557+
* Sets the maximum amount of time the driver will wait to acquire a connection suitable for a given purpose.
560558
* <p>
561-
* Exception is raised when connection can't be acquired within configured time.
559+
* In some situations the driver may need to do multiple actions and potentially acquire multiple connections to
560+
* get a suitable one. For instance, when client-side routing is used, home database resolution is required,
561+
* liveness checks are needed, etc.
562562
* <p>
563-
* Default value is 60 seconds. Negative values are allowed and result in unlimited acquisition timeout. Value
564-
* of {@code 0} is allowed and results in no timeout and immediate failure when connection is unavailable.
563+
* An exception is raised when connection can't be acquired within configured time.
564+
* <p>
565+
* Timeout value should be greater or equal to zero and represent a valid {@code long} value when converted to
566+
* {@link TimeUnit#MILLISECONDS milliseconds}.
567+
* <p>
568+
* Default value is 60 seconds. {@literal 0} value disables timeout.
569+
* <p>
570+
* This timeout should be bigger than {@link ConfigBuilder#withConnectionTimeout(long, TimeUnit)}.
565571
*
566572
* @param value the acquisition timeout
567573
* @param unit the unit in which the duration is given
568574
* @return this builder
569575
* @see #withMaxConnectionPoolSize(int)
576+
* @see #withConnectionTimeout(long, TimeUnit)
570577
*/
571578
public ConfigBuilder withConnectionAcquisitionTimeout(long value, TimeUnit unit) {
572579
var valueInMillis = unit.toMillis(value);
573-
if (value >= 0) {
574-
this.connectionAcquisitionTimeoutMillis = valueInMillis;
575-
} else {
576-
this.connectionAcquisitionTimeoutMillis = -1;
580+
if (valueInMillis < 0) {
581+
throw new IllegalArgumentException(format(
582+
"The connection acquisition timeout may not be smaller than 0, but was %d %s.", value, unit));
577583
}
584+
this.connectionAcquisitionTimeoutMillis = valueInMillis;
578585
return this;
579586
}
580587

‎driver/src/main/java/org/neo4j/driver/internal/DriverFactory.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -335,6 +335,7 @@ private RoutedBoltConnectionSource createRoutedBoltConnectionProvider(
335335
clock,
336336
loggingProvider,
337337
uri,
338+
config.connectionAcquisitionTimeoutMillis(),
338339
List.of(AuthTokenManagerExecutionException.class),
339340
observationProvider);
340341
}
@@ -381,7 +382,8 @@ private BoltConnectionSourceFactory createPooledBoltConnectionSource(
381382
boltAgent,
382383
userAgent,
383384
connectTimeoutMillis,
384-
notificationConfig);
385+
notificationConfig,
386+
PooledBoltConnectionSource.TimeoutPolicy.DEFAULT);
385387
};
386388
}
387389

‎driver/src/main/java/org/neo4j/driver/internal/SessionFactoryImpl.java‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,9 @@
2222
import java.util.Objects;
2323
import java.util.Optional;
2424
import java.util.Set;
25+
import java.util.concurrent.CompletionException;
2526
import java.util.concurrent.CompletionStage;
27+
import java.util.concurrent.TimeoutException;
2628
import org.neo4j.bolt.connection.DatabaseName;
2729
import org.neo4j.bolt.connection.SecurityPlan;
2830
import org.neo4j.driver.AccessMode;
@@ -35,13 +37,15 @@
3537
import org.neo4j.driver.NotificationConfig;
3638
import org.neo4j.driver.SessionConfig;
3739
import org.neo4j.driver.Value;
40+
import org.neo4j.driver.exceptions.ClientException;
3841
import org.neo4j.driver.internal.adaptedbolt.DriverBoltConnectionSource;
3942
import org.neo4j.driver.internal.async.LeakLoggingNetworkSession;
4043
import org.neo4j.driver.internal.async.NetworkSession;
4144
import org.neo4j.driver.internal.homedb.HomeDatabaseCache;
4245
import org.neo4j.driver.internal.observation.DriverObservationProvider;
4346
import org.neo4j.driver.internal.retry.RetryLogic;
4447
import org.neo4j.driver.internal.security.BoltSecurityPlanManager;
48+
import org.neo4j.driver.internal.util.Futures;
4549

4650
public class SessionFactoryImpl implements SessionFactory {
4751
private final BoltSecurityPlanManager securityPlanManager;
@@ -125,7 +129,20 @@ private DatabaseName parseDatabaseName(SessionConfig sessionConfig) {
125129

126130
@Override
127131
public CompletionStage<Void> verifyConnectivity() {
128-
return connectionSource.verifyConnectivity();
132+
return connectionSource.verifyConnectivity().exceptionally(throwable -> {
133+
throwable = Futures.completionExceptionCause(throwable);
134+
if (throwable instanceof TimeoutException) {
135+
throw new ClientException(
136+
GqlStatusError.UNKNOWN.getStatus(),
137+
GqlStatusError.UNKNOWN.getStatusDescription(throwable.getMessage()),
138+
"N/A",
139+
throwable.getMessage(),
140+
GqlStatusError.DIAGNOSTIC_RECORD,
141+
throwable);
142+
} else {
143+
throw new CompletionException(throwable);
144+
}
145+
});
129146
}
130147

131148
@Override

‎driver/src/main/java/org/neo4j/driver/internal/boltlistener/ListeningBoltConnectionProvider.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ public CompletionStage<BoltConnection> connect(
4545
BoltAgent boltAgent,
4646
String userAgent,
4747
int connectTimeoutMillis,
48+
long initialisationTimeoutMillis,
4849
SecurityPlan securityPlan,
4950
AuthToken authToken,
5051
BoltProtocolVersion minVersion,
@@ -56,6 +57,7 @@ public CompletionStage<BoltConnection> connect(
5657
boltAgent,
5758
userAgent,
5859
connectTimeoutMillis,
60+
initialisationTimeoutMillis,
5961
securityPlan,
6062
authToken,
6163
minVersion,

‎driver/src/test/java/org/neo4j/driver/ConfigTest.java‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -246,12 +246,11 @@ void shouldAllowPositiveConnectionAcquisitionTimeout() {
246246
}
247247

248248
@Test
249-
void shouldAllowNegativeConnectionAcquisitionTimeout() {
250-
var config = Config.builder()
251-
.withConnectionAcquisitionTimeout(-42, TimeUnit.HOURS)
252-
.build();
249+
void shouldRejectNegativeConnectionAcquisitionTimeout() {
250+
var builder = Config.builder();
253251

254-
assertEquals(-1, config.connectionAcquisitionTimeoutMillis());
252+
assertThrows(
253+
IllegalArgumentException.class, () -> builder.withConnectionAcquisitionTimeout(-42, TimeUnit.HOURS));
255254
}
256255

257256
@Test

‎driver/src/test/java/org/neo4j/driver/GraphDatabaseTest.java‎

Lines changed: 20 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -20,16 +20,17 @@
2020
import static java.util.concurrent.TimeUnit.SECONDS;
2121
import static org.hamcrest.MatcherAssert.assertThat;
2222
import static org.hamcrest.Matchers.containsString;
23-
import static org.junit.jupiter.api.Assertions.assertEquals;
23+
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
2424
import static org.junit.jupiter.api.Assertions.assertNotNull;
2525
import static org.junit.jupiter.api.Assertions.assertThrows;
26-
import static org.neo4j.driver.internal.logging.DevNullLogging.DEV_NULL_LOGGING;
2726

2827
import java.io.IOException;
2928
import java.net.ServerSocket;
3029
import java.net.URI;
30+
import java.util.concurrent.TimeoutException;
3131
import org.junit.jupiter.api.Disabled;
3232
import org.junit.jupiter.api.Test;
33+
import org.neo4j.driver.exceptions.Neo4jException;
3334
import org.neo4j.driver.exceptions.ServiceUnavailableException;
3435
import org.neo4j.driver.internal.security.StaticAuthTokenManager;
3536
import org.neo4j.driver.testutil.TestUtil;
@@ -56,9 +57,12 @@ void shouldRespondToInterruptsWhenConnectingToUnresponsiveServer() throws Except
5657
@SuppressWarnings("resource")
5758
final var driver = GraphDatabase.driver(
5859
"bolt://localhost:" + serverSocket.getLocalPort(),
59-
Config.builder().withConnectionTimeout(1, SECONDS).build());
60+
Config.builder()
61+
.withConnectionTimeout(1, SECONDS)
62+
.withConnectionAcquisitionTimeout(1, SECONDS)
63+
.build());
6064
try {
61-
assertThrows(ServiceUnavailableException.class, driver::verifyConnectivity);
65+
assertThrows(Neo4jException.class, driver::verifyConnectivity);
6266
} finally {
6367
// clear interrupted flag
6468
Thread.interrupted();
@@ -171,27 +175,20 @@ private static void testFailureWhenServerDoesNotRespond(boolean encrypted) throw
171175
try (var server = new ServerSocket(0)) // server that accepts connections but does not reply
172176
{
173177
var connectionTimeoutMillis = 1_000;
174-
var config = createConfig(encrypted, connectionTimeoutMillis);
178+
var configBuilder = Config.builder()
179+
.withConnectionTimeout(connectionTimeoutMillis, MILLISECONDS)
180+
.withConnectionAcquisitionTimeout(connectionTimeoutMillis, MILLISECONDS);
181+
if (encrypted) {
182+
configBuilder.withEncryption();
183+
} else {
184+
configBuilder.withoutEncryption();
185+
}
175186
@SuppressWarnings("resource")
176-
final var driver = GraphDatabase.driver(URI.create("bolt://localhost:" + server.getLocalPort()), config);
177-
178-
var e = assertThrows(ServiceUnavailableException.class, driver::verifyConnectivity);
179-
assertEquals(e.getMessage(), "Unable to establish connection in " + connectionTimeoutMillis + "ms");
180-
}
181-
}
182-
183-
private static Config createConfig(boolean encrypted, int timeoutMillis) {
184-
@SuppressWarnings("deprecation")
185-
var configBuilder = Config.builder()
186-
.withConnectionTimeout(timeoutMillis, MILLISECONDS)
187-
.withLogging(DEV_NULL_LOGGING);
187+
final var driver = GraphDatabase.driver(
188+
URI.create("bolt://localhost:" + server.getLocalPort()), configBuilder.build());
188189

189-
if (encrypted) {
190-
configBuilder.withEncryption();
191-
} else {
192-
configBuilder.withoutEncryption();
190+
var e = assertThrows(Neo4jException.class, driver::verifyConnectivity);
191+
assertInstanceOf(TimeoutException.class, e.getCause());
193192
}
194-
195-
return configBuilder.build();
196193
}
197194
}

‎driver/src/test/java/org/neo4j/driver/integration/EncryptionIT.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ void shouldOperateWithEncryptionWhenConfiguredUsingBoltSscURI() {
6868

6969
@Test
7070
void shouldFailWithEncryptionWhenItIsDisabledInTheDatabase() {
71-
testMismatchingEncryption(BoltTlsLevel.DISABLED, true, "Unable to write Bolt handshake to");
71+
testMismatchingEncryption(BoltTlsLevel.DISABLED, true, "SSL handshake with");
7272
}
7373

7474
@Test

‎driver/src/test/java/org/neo4j/driver/integration/SessionIT.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -646,7 +646,7 @@ void shouldNotRetryOnConnectionAcquisitionTimeout() {
646646
var maxPoolSize = 3;
647647
var config = Config.builder()
648648
.withMaxConnectionPoolSize(maxPoolSize)
649-
.withConnectionAcquisitionTimeout(0, TimeUnit.SECONDS)
649+
.withConnectionAcquisitionTimeout(100, TimeUnit.MILLISECONDS)
650650
.withMaxTransactionRetryTime(42, TimeUnit.DAYS) // retry for a really long time
651651
.withEventLoopThreads(1)
652652
.build();
@@ -660,7 +660,7 @@ void shouldNotRetryOnConnectionAcquisitionTimeout() {
660660
var invocations = new AtomicInteger();
661661
var e = assertThrows(
662662
ClientException.class, () -> driver.session().executeWrite(tx -> invocations.incrementAndGet()));
663-
assertThat(e, is(connectionAcquisitionTimeoutError(0)));
663+
assertThat(e, is(connectionAcquisitionTimeoutError(100)));
664664

665665
// work should never be invoked
666666
assertEquals(0, invocations.get());

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@
3434
<maven.deploy.skip>true</maven.deploy.skip>
3535

3636
<!-- Versions -->
37-
<neo4j-bolt-connection-bom.version>7.0.0</neo4j-bolt-connection-bom.version>
37+
<neo4j-bolt-connection-bom.version>8.0.1</neo4j-bolt-connection-bom.version>
3838
<reactive-streams.version>1.0.4</reactive-streams.version>
3939
<!-- Please note that when updating this dependency -->
4040
<!-- (i.e. due to a security vulnerability or bug) that the -->

‎testkit-backend/src/main/java/neo4j/org/testkit/backend/messages/requests/StartTest.java‎

Lines changed: 1 addition & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -61,26 +61,7 @@ public class StartTest implements TestkitRequest {
6161
COMMON_SKIP_PATTERN_TO_REASON.put(
6262
"^.*\\.test_partial_summary_contains_updates$", "Does not contain updates because value is zero");
6363
COMMON_SKIP_PATTERN_TO_REASON.put("^.*\\.test_supports_multi_db$", "Database is None");
64-
var skipMessage = "Driver handles connection acquisition timeout differently";
65-
COMMON_SKIP_PATTERN_TO_REASON.put(
66-
"^.*\\.TestConnectionAcquisitionTimeoutMs\\.test_should_encompass_the_handshake_time.*$", skipMessage);
67-
COMMON_SKIP_PATTERN_TO_REASON.put(
68-
"^.*\\.TestConnectionAcquisitionTimeoutMs\\.test_router_handshake_has_own_timeout_too_slow$",
69-
skipMessage);
70-
COMMON_SKIP_PATTERN_TO_REASON.put(
71-
"^.*\\.TestConnectionAcquisitionTimeoutMs\\.test_should_fail_when_acquisition_timeout_is_reached_first.*$",
72-
skipMessage);
73-
COMMON_SKIP_PATTERN_TO_REASON.put(
74-
"^.*\\.TestConnectionAcquisitionTimeoutMs\\.test_should_encompass_the_version_handshake_(in_time|time_out)$",
75-
skipMessage);
76-
COMMON_SKIP_PATTERN_TO_REASON.put(
77-
"^.*\\.TestHomeDbMixedCluster\\.test_connection_acquisition_timeout_during_fallback$", skipMessage);
78-
COMMON_SKIP_PATTERN_TO_REASON.put(
79-
"^.*\\.TestConnectionAcquisitionTimeoutMs\\.test_does_encompass_router_route_response$", skipMessage);
80-
COMMON_SKIP_PATTERN_TO_REASON.put(
81-
"^.*\\.TestConnectionAcquisitionTimeoutMs\\.test_router_handshake_shares_acquisition_timeout$",
82-
skipMessage);
83-
skipMessage = "This test needs updating to implement expected behaviour";
64+
var skipMessage = "This test needs updating to implement expected behaviour";
8465
COMMON_SKIP_PATTERN_TO_REASON.put(
8566
"^.*\\.TestAuthenticationSchemes[^.]+\\.test_custom_scheme_empty$", skipMessage);
8667
skipMessage = "Driver does not implement optimization for qid in explicit transaction";

0 commit comments

Comments
 (0)