Skip to content

Commit 450135d

Browse files
lutovichlutovich
committed
Improve connection acquisition timeout error
Previously driver would propagate `TimeoutException` directly from netty's `FixedChannelPool`. This commit changes it to `ClientException`. Also added couple integration tests.
1 parent bf86eaa commit 450135d

File tree

5 files changed

+163
-16
lines changed

5 files changed

+163
-16
lines changed

‎driver/src/main/java/org/neo4j/driver/internal/async/pool/ConnectionPoolImpl.java

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,11 @@
2424
import io.netty.channel.pool.ChannelPool;
2525
import io.netty.util.concurrent.Future;
2626

27+
import java.util.concurrent.CompletionException;
2728
import java.util.concurrent.CompletionStage;
2829
import java.util.concurrent.ConcurrentHashMap;
2930
import java.util.concurrent.ConcurrentMap;
31+
import java.util.concurrent.TimeoutException;
3032
import java.util.concurrent.atomic.AtomicBoolean;
3133

3234
import org.neo4j.driver.internal.async.BoltServerAddress;
@@ -38,6 +40,7 @@
3840
import org.neo4j.driver.internal.util.Futures;
3941
import org.neo4j.driver.v1.Logger;
4042
import org.neo4j.driver.v1.Logging;
43+
import org.neo4j.driver.v1.exceptions.ClientException;
4144

4245
public class ConnectionPoolImpl implements ConnectionPool
4346
{
@@ -73,8 +76,9 @@ public CompletionStage<Connection> acquire( final BoltServerAddress address )
7376
ChannelPool pool = getOrCreatePool( address );
7477
Future<Channel> connectionFuture = pool.acquire();
7578

76-
return Futures.asCompletionStage( connectionFuture ).thenApply( channel ->
79+
return Futures.asCompletionStage( connectionFuture ).handle( ( channel, error ) ->
7780
{
81+
processAcquisitionError( error );
7882
assertNotClosed( address, channel, pool );
7983
return new NettyConnection( channel, pool, clock );
8084
} );
@@ -160,6 +164,27 @@ private EventLoopGroup eventLoopGroup()
160164
return bootstrap.config().group();
161165
}
162166

167+
private void processAcquisitionError( Throwable error )
168+
{
169+
Throwable cause = Futures.completionErrorCause( error );
170+
if ( cause != null )
171+
{
172+
if ( cause instanceof TimeoutException )
173+
{
174+
// NettyChannelPool returns future failed with TimeoutException if acquire operation takes more than
175+
// configured time, translate this exception to a prettier one and re-throw
176+
throw new ClientException(
177+
"Unable to acquire connection from the pool within configured maximum time of " +
178+
settings.connectionAcquisitionTimeout() + "ms" );
179+
}
180+
else
181+
{
182+
// some unknown error happened during connection acquisition, propagate it
183+
throw new CompletionException( cause );
184+
}
185+
}
186+
}
187+
163188
private void assertNotClosed()
164189
{
165190
if ( closed.get() )

‎driver/src/test/java/org/neo4j/driver/internal/util/Matchers.java

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,10 +29,10 @@
2929
import org.neo4j.driver.internal.InternalDriver;
3030
import org.neo4j.driver.internal.SessionFactory;
3131
import org.neo4j.driver.internal.SessionFactoryImpl;
32+
import org.neo4j.driver.internal.async.BoltServerAddress;
3233
import org.neo4j.driver.internal.cluster.AddressSet;
3334
import org.neo4j.driver.internal.cluster.RoutingTable;
3435
import org.neo4j.driver.internal.cluster.loadbalancing.LoadBalancer;
35-
import org.neo4j.driver.internal.async.BoltServerAddress;
3636
import org.neo4j.driver.internal.spi.ConnectionProvider;
3737
import org.neo4j.driver.v1.Driver;
3838
import org.neo4j.driver.v1.exceptions.ClientException;
@@ -236,6 +236,30 @@ public void describeTo( Description description )
236236
};
237237
}
238238

239+
public static Matcher<Throwable> connectionAcquisitionTimeoutError( int timeoutMillis )
240+
{
241+
return new TypeSafeMatcher<Throwable>()
242+
{
243+
@Override
244+
protected boolean matchesSafely( Throwable error )
245+
{
246+
if ( error instanceof ClientException )
247+
{
248+
String expectedMessage = "Unable to acquire connection from the pool within " +
249+
"configured maximum time of " + timeoutMillis + "ms";
250+
return expectedMessage.equals( error.getMessage() );
251+
}
252+
return false;
253+
}
254+
255+
@Override
256+
public void describeTo( Description description )
257+
{
258+
description.appendText( "acquisition timeout error with " + timeoutMillis + "ms" );
259+
}
260+
};
261+
}
262+
239263
private static boolean contains( AddressSet set, BoltServerAddress address )
240264
{
241265
BoltServerAddress[] addresses = set.toArray();

‎driver/src/test/java/org/neo4j/driver/v1/integration/CausalClusteringIT.java

Lines changed: 49 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,6 @@
3535
import java.util.concurrent.TimeoutException;
3636

3737
import org.neo4j.driver.internal.cluster.RoutingSettings;
38-
import org.neo4j.driver.internal.logging.DevNullLogger;
3938
import org.neo4j.driver.internal.retry.RetrySettings;
4039
import org.neo4j.driver.internal.util.ChannelTrackingDriverFactory;
4140
import org.neo4j.driver.internal.util.FakeClock;
@@ -44,8 +43,6 @@
4443
import org.neo4j.driver.v1.Config;
4544
import org.neo4j.driver.v1.Driver;
4645
import org.neo4j.driver.v1.GraphDatabase;
47-
import org.neo4j.driver.v1.Logger;
48-
import org.neo4j.driver.v1.Logging;
4946
import org.neo4j.driver.v1.Record;
5047
import org.neo4j.driver.v1.Session;
5148
import org.neo4j.driver.v1.StatementResult;
@@ -64,9 +61,11 @@
6461
import org.neo4j.driver.v1.util.cc.ClusterMemberRole;
6562
import org.neo4j.driver.v1.util.cc.ClusterRule;
6663

64+
import static java.util.concurrent.TimeUnit.MILLISECONDS;
6765
import static java.util.concurrent.TimeUnit.SECONDS;
6866
import static org.hamcrest.Matchers.containsString;
6967
import static org.hamcrest.Matchers.instanceOf;
68+
import static org.hamcrest.Matchers.is;
7069
import static org.hamcrest.Matchers.startsWith;
7170
import static org.junit.Assert.assertEquals;
7271
import static org.junit.Assert.assertNotEquals;
@@ -76,6 +75,7 @@
7675
import static org.junit.Assert.fail;
7776
import static org.neo4j.driver.internal.logging.DevNullLogging.DEV_NULL_LOGGING;
7877
import static org.neo4j.driver.internal.util.Futures.getBlocking;
78+
import static org.neo4j.driver.internal.util.Matchers.connectionAcquisitionTimeoutError;
7979
import static org.neo4j.driver.v1.Values.parameters;
8080

8181
public class CausalClusteringIT
@@ -532,6 +532,46 @@ public void shouldNotReuseReadConnectionForWriteTransaction()
532532
}
533533
}
534534

535+
@Test
536+
public void shouldRespectMaxConnectionPoolSizePerClusterMember()
537+
{
538+
Cluster cluster = clusterRule.getCluster();
539+
ClusterMember leader = cluster.leader();
540+
541+
Config config = Config.build()
542+
.withMaxConnectionPoolSize( 2 )
543+
.withConnectionAcquisitionTimeout( 42, MILLISECONDS )
544+
.withLogging( DEV_NULL_LOGGING )
545+
.toConfig();
546+
547+
try ( Driver driver = createDriver( leader.getRoutingUri(), config ) )
548+
{
549+
Session writeSession1 = driver.session( AccessMode.WRITE );
550+
writeSession1.beginTransaction();
551+
552+
Session writeSession2 = driver.session( AccessMode.WRITE );
553+
writeSession2.beginTransaction();
554+
555+
// should not be possible to acquire more connections towards leader because limit is 2
556+
Session writeSession3 = driver.session( AccessMode.WRITE );
557+
try
558+
{
559+
writeSession3.beginTransaction();
560+
fail( "Exception expected" );
561+
}
562+
catch ( ClientException e )
563+
{
564+
assertThat( e, is( connectionAcquisitionTimeoutError( 42 ) ) );
565+
}
566+
567+
// should be possible to acquire new connection towards read server
568+
// it's a different machine, not leader, so different max connection pool size limit applies
569+
Session readSession = driver.session( AccessMode.READ );
570+
Record record = readSession.readTransaction( tx -> tx.run( "RETURN 1" ).single() );
571+
assertEquals( 1, record.get( 0 ).asInt() );
572+
}
573+
}
574+
535575
private CompletionStage<List<RecordAndSummary>> combineCursors( StatementResultCursor cursor1,
536576
StatementResultCursor cursor2 )
537577
{
@@ -702,19 +742,15 @@ else if ( role == ClusterMemberRole.READ_REPLICA )
702742

703743
private Driver createDriver( URI boltUri )
704744
{
705-
Logging devNullLogging = new Logging()
706-
{
707-
@Override
708-
public Logger getLog( String name )
709-
{
710-
return DevNullLogger.DEV_NULL_LOGGER;
711-
}
712-
};
713-
714745
Config config = Config.build()
715-
.withLogging( devNullLogging )
746+
.withLogging( DEV_NULL_LOGGING )
716747
.toConfig();
717748

749+
return createDriver( boltUri, config );
750+
}
751+
752+
private Driver createDriver( URI boltUri, Config config )
753+
{
718754
return GraphDatabase.driver( boltUri, clusterRule.getDefaultAuthToken(), config );
719755
}
720756

‎driver/src/test/java/org/neo4j/driver/v1/integration/ConnectionPoolIT.java

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,21 +30,26 @@
3030

3131
import org.neo4j.driver.internal.cluster.RoutingSettings;
3232
import org.neo4j.driver.internal.util.ChannelTrackingDriverFactory;
33+
import org.neo4j.driver.internal.util.DriverFactoryWithOneEventLoopThread;
3334
import org.neo4j.driver.internal.util.FakeClock;
3435
import org.neo4j.driver.v1.Config;
3536
import org.neo4j.driver.v1.Driver;
3637
import org.neo4j.driver.v1.GraphDatabase;
3738
import org.neo4j.driver.v1.Session;
3839
import org.neo4j.driver.v1.StatementResult;
3940
import org.neo4j.driver.v1.Transaction;
41+
import org.neo4j.driver.v1.exceptions.ClientException;
4042
import org.neo4j.driver.v1.util.TestNeo4j;
4143

4244
import static java.util.concurrent.TimeUnit.SECONDS;
43-
import static junit.framework.TestCase.fail;
45+
import static org.hamcrest.Matchers.is;
4446
import static org.junit.Assert.assertEquals;
4547
import static org.junit.Assert.assertFalse;
48+
import static org.junit.Assert.assertThat;
4649
import static org.junit.Assert.assertTrue;
50+
import static org.junit.Assert.fail;
4751
import static org.neo4j.driver.internal.retry.RetrySettings.DEFAULT;
52+
import static org.neo4j.driver.internal.util.Matchers.connectionAcquisitionTimeoutError;
4853

4954
public class ConnectionPoolIT
5055
{
@@ -113,6 +118,28 @@ public void shouldDisposeChannelsBasedOnMaxLifetime() throws Exception
113118
assertTrue( channel2.isActive() );
114119
}
115120

121+
@Test
122+
public void shouldRespectMaxConnectionPoolSize()
123+
{
124+
int maxPoolSize = 3;
125+
Config config = Config.build()
126+
.withMaxConnectionPoolSize( maxPoolSize )
127+
.withConnectionAcquisitionTimeout( 542, TimeUnit.MILLISECONDS )
128+
.toConfig();
129+
130+
driver = new DriverFactoryWithOneEventLoopThread().newInstance( neo4j.uri(), neo4j.authToken(), config );
131+
132+
try
133+
{
134+
startAndCloseTransactions( driver, maxPoolSize + 1 );
135+
fail( "Exception expected" );
136+
}
137+
catch ( ClientException e )
138+
{
139+
assertThat( e, is( connectionAcquisitionTimeoutError( 542 ) ) );
140+
}
141+
}
142+
116143
@After
117144
public void cleanup() throws Exception
118145
{

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

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
import org.neo4j.driver.internal.logging.DevNullLogging;
4646
import org.neo4j.driver.internal.retry.RetrySettings;
4747
import org.neo4j.driver.internal.util.DriverFactoryWithFixedRetryLogic;
48+
import org.neo4j.driver.internal.util.DriverFactoryWithOneEventLoopThread;
4849
import org.neo4j.driver.internal.util.ServerVersion;
4950
import org.neo4j.driver.v1.AccessMode;
5051
import org.neo4j.driver.v1.AuthToken;
@@ -73,6 +74,7 @@
7374
import static org.hamcrest.Matchers.containsInAnyOrder;
7475
import static org.hamcrest.Matchers.emptyArray;
7576
import static org.hamcrest.Matchers.greaterThan;
77+
import static org.hamcrest.Matchers.is;
7678
import static org.hamcrest.Matchers.not;
7779
import static org.junit.Assert.assertEquals;
7880
import static org.junit.Assert.assertFalse;
@@ -87,6 +89,7 @@
8789
import static org.mockito.Mockito.spy;
8890
import static org.mockito.Mockito.times;
8991
import static org.mockito.Mockito.verify;
92+
import static org.neo4j.driver.internal.util.Matchers.connectionAcquisitionTimeoutError;
9093
import static org.neo4j.driver.internal.util.ServerVersion.v3_1_0;
9194
import static org.neo4j.driver.v1.Values.parameters;
9295
import static org.neo4j.driver.v1.util.DaemonThreadFactory.daemon;
@@ -1296,6 +1299,38 @@ public void shouldConsumePreviousResultBeforeRunningNewQuery()
12961299
}
12971300
}
12981301

1302+
@Test
1303+
public void shouldNotRetryOnConnectionAcquisitionTimeout()
1304+
{
1305+
int maxPoolSize = 3;
1306+
Config config = Config.build()
1307+
.withMaxConnectionPoolSize( maxPoolSize )
1308+
.withConnectionAcquisitionTimeout( 0, TimeUnit.SECONDS )
1309+
.withMaxTransactionRetryTime( 42, TimeUnit.DAYS ) // retry for a really long time
1310+
.toConfig();
1311+
1312+
driver = new DriverFactoryWithOneEventLoopThread().newInstance( neo4j.uri(), neo4j.authToken(), config );
1313+
1314+
for ( int i = 0; i < maxPoolSize; i++ )
1315+
{
1316+
driver.session().beginTransaction();
1317+
}
1318+
1319+
AtomicInteger invocations = new AtomicInteger();
1320+
try
1321+
{
1322+
driver.session().writeTransaction( tx -> invocations.incrementAndGet() );
1323+
fail( "Exception expected" );
1324+
}
1325+
catch ( ClientException e )
1326+
{
1327+
assertThat( e, is( connectionAcquisitionTimeoutError( 0 ) ) );
1328+
}
1329+
1330+
// work should never be invoked
1331+
assertEquals( 0, invocations.get() );
1332+
}
1333+
12991334
private void assumeServerIs31OrLater()
13001335
{
13011336
ServerVersion serverVersion = ServerVersion.version( neo4j.driver() );

0 commit comments

Comments
 (0)