Skip to content

Commit dc0fa7d

Browse files
injectivesinjectives
committed
Improve connection release handling and improve flaky test (neo4j#1092)
This update ensures connection release stages are linked. In addition, it improves stability of flaky `RoutingTableAndConnectionPoolTest.shouldHandleAddAndRemoveFromRoutingTableAndConnectionPool` test.
1 parent cae2b9d commit dc0fa7d

File tree

3 files changed

+36
-18
lines changed

3 files changed

+36
-18
lines changed

‎driver/src/main/java/org/neo4j/driver/internal/async/NetworkConnection.java

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@
4848
import static java.util.Collections.emptyMap;
4949
import static org.neo4j.driver.internal.async.connection.ChannelAttributes.poolId;
5050
import static org.neo4j.driver.internal.async.connection.ChannelAttributes.setTerminationReason;
51+
import static org.neo4j.driver.internal.util.Futures.asCompletionStage;
5152

5253
/**
5354
* This connection represents a simple network connection to a remote server. It wraps a channel obtained from a connection pool. The life cycle of this
@@ -189,10 +190,14 @@ public void terminateAndRelease( String reason )
189190
if ( status.compareAndSet( Status.OPEN, Status.TERMINATED ) )
190191
{
191192
setTerminationReason( channel, reason );
192-
channel.close();
193-
channelPool.release( channel );
194-
releaseFuture.complete( null );
195-
metricsListener.afterConnectionReleased( poolId( this.channel ), this.inUseEvent );
193+
asCompletionStage( channel.close() )
194+
.exceptionally( throwable -> null )
195+
.thenCompose( ignored -> channelPool.release( channel ) )
196+
.whenComplete( ( ignored, throwable ) ->
197+
{
198+
releaseFuture.complete( null );
199+
metricsListener.afterConnectionReleased( poolId( this.channel ), this.inUseEvent );
200+
} );
196201
}
197202
}
198203

‎driver/src/main/java/org/neo4j/driver/internal/handlers/ChannelReleasingResetResponseHandler.java

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,8 @@
2828
import org.neo4j.driver.internal.util.Clock;
2929

3030
import static org.neo4j.driver.internal.async.connection.ChannelAttributes.setLastUsedTimestamp;
31+
import static org.neo4j.driver.internal.util.Futures.asCompletionStage;
32+
import static org.neo4j.driver.internal.util.Futures.completedWithNull;
3133

3234
public class ChannelReleasingResetResponseHandler extends ResetResponseHandler
3335
{
@@ -47,18 +49,20 @@ public ChannelReleasingResetResponseHandler( Channel channel, ExtendedChannelPoo
4749
@Override
4850
protected void resetCompleted( CompletableFuture<Void> completionFuture, boolean success )
4951
{
52+
CompletionStage<Void> closureStage;
5053
if ( success )
5154
{
5255
// update the last-used timestamp before returning the channel back to the pool
5356
setLastUsedTimestamp( channel, clock.millis() );
57+
closureStage = completedWithNull();
5458
}
5559
else
5660
{
5761
// close the channel before returning it back to the pool if RESET failed
58-
channel.close();
62+
closureStage = asCompletionStage( channel.close() );
5963
}
60-
61-
CompletionStage<Void> released = pool.release( channel );
62-
released.whenComplete( ( ignore, error ) -> completionFuture.complete( null ) );
64+
closureStage.exceptionally( throwable -> null )
65+
.thenCompose( ignored -> pool.release( channel ) )
66+
.whenComplete( ( ignore, error ) -> completionFuture.complete( null ) );
6367
}
6468
}

‎driver/src/test/java/org/neo4j/driver/internal/cluster/loadbalancing/RoutingTableAndConnectionPoolTest.java

Lines changed: 19 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -25,9 +25,11 @@
2525
import java.time.Duration;
2626
import java.util.ArrayList;
2727
import java.util.Arrays;
28+
import java.util.Collections;
2829
import java.util.HashSet;
2930
import java.util.LinkedList;
3031
import java.util.List;
32+
import java.util.Objects;
3133
import java.util.Random;
3234
import java.util.Set;
3335
import java.util.concurrent.CompletableFuture;
@@ -43,6 +45,7 @@
4345
import org.neo4j.driver.exceptions.FatalDiscoveryException;
4446
import org.neo4j.driver.exceptions.ProtocolException;
4547
import org.neo4j.driver.internal.BoltServerAddress;
48+
import org.neo4j.driver.internal.DatabaseNameUtil;
4649
import org.neo4j.driver.internal.async.connection.BootstrapFactory;
4750
import org.neo4j.driver.internal.async.pool.NettyChannelHealthChecker;
4851
import org.neo4j.driver.internal.async.pool.NettyChannelTracker;
@@ -85,7 +88,7 @@ class RoutingTableAndConnectionPoolTest
8588
private static final BoltServerAddress D = new BoltServerAddress( "localhost:30003" );
8689
private static final BoltServerAddress E = new BoltServerAddress( "localhost:30004" );
8790
private static final BoltServerAddress F = new BoltServerAddress( "localhost:30005" );
88-
private static final List<BoltServerAddress> SERVERS = new LinkedList<>( Arrays.asList( null, A, B, C, D, E, F ) );
91+
private static final List<BoltServerAddress> SERVERS = Collections.synchronizedList( new LinkedList<>( Arrays.asList( null, A, B, C, D, E, F ) ) );
8992

9093
private static final String[] DATABASES = new String[]{"", SYSTEM_DATABASE_NAME, "my database"};
9194

@@ -94,7 +97,7 @@ class RoutingTableAndConnectionPoolTest
9497
private final Logging logging = none();
9598

9699
@Test
97-
void shouldAddServerToRoutingTableAndConnectionPool() throws Throwable
100+
void shouldAddServerToRoutingTableAndConnectionPool()
98101
{
99102
// Given
100103
ConnectionPool connectionPool = newConnectionPool();
@@ -114,7 +117,7 @@ void shouldAddServerToRoutingTableAndConnectionPool() throws Throwable
114117
}
115118

116119
@Test
117-
void shouldNotAddToRoutingTableWhenFailedWithRoutingError() throws Throwable
120+
void shouldNotAddToRoutingTableWhenFailedWithRoutingError()
118121
{
119122
// Given
120123
ConnectionPool connectionPool = newConnectionPool();
@@ -133,7 +136,7 @@ void shouldNotAddToRoutingTableWhenFailedWithRoutingError() throws Throwable
133136
}
134137

135138
@Test
136-
void shouldNotAddToRoutingTableWhenFailedWithProtocolError() throws Throwable
139+
void shouldNotAddToRoutingTableWhenFailedWithProtocolError()
137140
{
138141
// Given
139142
ConnectionPool connectionPool = newConnectionPool();
@@ -152,7 +155,7 @@ void shouldNotAddToRoutingTableWhenFailedWithProtocolError() throws Throwable
152155
}
153156

154157
@Test
155-
void shouldNotAddToRoutingTableWhenFailedWithSecurityError() throws Throwable
158+
void shouldNotAddToRoutingTableWhenFailedWithSecurityError()
156159
{
157160
// Given
158161
ConnectionPool connectionPool = newConnectionPool();
@@ -171,7 +174,7 @@ void shouldNotAddToRoutingTableWhenFailedWithSecurityError() throws Throwable
171174
}
172175

173176
@Test
174-
void shouldNotRemoveNewlyAddedRoutingTableEvenIfItIsExpired() throws Throwable
177+
void shouldNotRemoveNewlyAddedRoutingTableEvenIfItIsExpired()
175178
{
176179
// Given
177180
ConnectionPool connectionPool = newConnectionPool();
@@ -194,7 +197,7 @@ void shouldNotRemoveNewlyAddedRoutingTableEvenIfItIsExpired() throws Throwable
194197
}
195198

196199
@Test
197-
void shouldRemoveExpiredRoutingTableAndServers() throws Throwable
200+
void shouldRemoveExpiredRoutingTableAndServers()
198201
{
199202
// Given
200203
ConnectionPool connectionPool = newConnectionPool();
@@ -219,7 +222,7 @@ void shouldRemoveExpiredRoutingTableAndServers() throws Throwable
219222
}
220223

221224
@Test
222-
void shouldRemoveExpiredRoutingTableButNotServer() throws Throwable
225+
void shouldRemoveExpiredRoutingTableButNotServer()
223226
{
224227
// Given
225228
ConnectionPool connectionPool = newConnectionPool();
@@ -256,7 +259,7 @@ void shouldHandleAddAndRemoveFromRoutingTableAndConnectionPool() throws Throwabl
256259
acquireAndReleaseConnections( loadBalancer );
257260
Set<BoltServerAddress> servers = routingTables.allServers();
258261
BoltServerAddress openServer = null;
259-
for( BoltServerAddress server: servers )
262+
for ( BoltServerAddress server : servers )
260263
{
261264
if ( connectionPool.isOpen( server ) )
262265
{
@@ -268,6 +271,8 @@ void shouldHandleAddAndRemoveFromRoutingTableAndConnectionPool() throws Throwabl
268271

269272
// if we remove the open server from servers, then the connection pool should remove the server from the pool.
270273
SERVERS.remove( openServer );
274+
// ensure rediscovery is necessary on subsequent interaction
275+
Arrays.stream( DATABASES ).map( DatabaseNameUtil::database ).forEach( routingTables::remove );
271276
acquireAndReleaseConnections( loadBalancer );
272277

273278
assertFalse( connectionPool.isOpen( openServer ) );
@@ -368,7 +373,11 @@ public CompletionStage<ClusterCompositionLookupResult> lookupClusterComposition(
368373
}
369374
if ( servers.size() == 0 )
370375
{
371-
servers.add( A );
376+
BoltServerAddress address = SERVERS.stream()
377+
.filter( Objects::nonNull )
378+
.findFirst()
379+
.orElseThrow( () -> new RuntimeException( "No non null server addresses are available" ) );
380+
servers.add( address );
372381
}
373382
ClusterComposition composition = new ClusterComposition( clock.millis() + 1, servers, servers, servers );
374383
return CompletableFuture.completedFuture( new ClusterCompositionLookupResult( composition ) );

0 commit comments

Comments
 (0)