Do not close connections concurrently in tests

Couple CC integration tests tried to simulate connection breakage by
closing connections concurrently from a different thread. This was
wrong because connections are not thread safe. It resulted in flaky
tests.

This commit replaces concurrent closing with just throwing errors
from RUN and RESET. Connections with such errors will still be
considered broken.

Author: lutovich

driver/src/test/java/org/neo4j/driver/internal/util/ConnectionTrackingConnector.java

@@ -19,6 +19,7 @@
 package org.neo4j.driver.internal.util;
 
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.neo4j.driver.internal.net.BoltServerAddress;
 import org.neo4j.driver.internal.spi.Collector;
@@ -29,11 +30,14 @@
 public class ThrowingConnection implements Connection
 {
     private final Connection realConnection;
-    private RuntimeException nextRunFailure;
+    private final AtomicReference<RuntimeException> nextRunError;
+    private final AtomicReference<RuntimeException> nextResetError;
 
     public ThrowingConnection( Connection realConnection )
     {
         this.realConnection = realConnection;
+        this.nextRunError = new AtomicReference<>();
+        this.nextResetError = new AtomicReference<>();
     }
 
     @Override
@@ -45,12 +49,7 @@ public void init( String clientName, Map<String,Value> authToken )
     @Override
     public void run( String statement, Map<String,Value> parameters, Collector collector )
     {
-        if ( nextRunFailure != null )
-        {
-            RuntimeException error = nextRunFailure;
-            nextRunFailure = null;
-            throw error;
-        }
+        throwErrorIfExists( nextRunError );
         realConnection.run( statement, parameters, collector );
     }
 
@@ -69,6 +68,7 @@ public void pullAll( Collector collector )
     @Override
     public void reset()
     {
+        throwErrorIfExists( nextResetError );
         realConnection.reset();
     }
 
@@ -132,8 +132,22 @@ public BoltServerAddress boltServerAddress()
         return realConnection.boltServerAddress();
     }
 
-    public void setNextRunFailure( RuntimeException nextRunFailure )
+    public void setNextRunError( RuntimeException error )
+    {
+        nextRunError.set( error );
+    }
+
+    public void setNextResetError( RuntimeException error )
     {
-        this.nextRunFailure = nextRunFailure;
+        nextResetError.set( error );
+    }
+
+    private static void throwErrorIfExists( AtomicReference<RuntimeException> errorReference )
+    {
+        RuntimeException error = errorReference.getAndSet( null );
+        if ( error != null )
+        {
+            throw error;
+        }
     }
 }

driver/src/test/java/org/neo4j/driver/internal/util/ConnectionTrackingDriverFactory.java

@@ -30,8 +30,25 @@
 
 public class ThrowingConnectionDriverFactory extends DriverFactory
 {
+    private final Clock clock;
     private final List<ThrowingConnection> connections = new CopyOnWriteArrayList<>();
 
+    public ThrowingConnectionDriverFactory()
+    {
+        this( Clock.SYSTEM );
+    }
+
+    public ThrowingConnectionDriverFactory( Clock clock )
+    {
+        this.clock = clock;
+    }
+
+    @Override
+    protected Clock createClock()
+    {
+        return clock;
+    }
+
     @Override
     protected Connector createConnector( ConnectionSettings connectionSettings, SecurityPlan securityPlan,
             Logging logging )
@@ -44,4 +61,11 @@ public List<ThrowingConnection> getConnections()
     {
         return new ArrayList<>( connections );
     }
+
+    public List<ThrowingConnection> pollConnections()
+    {
+        List<ThrowingConnection> result = new ArrayList<>( connections );
+        connections.clear();
+        return result;
+    }
 }

driver/src/test/java/org/neo4j/driver/internal/util/ThrowingConnection.java

@@ -35,8 +35,6 @@
 
 import org.neo4j.driver.internal.cluster.RoutingSettings;
 import org.neo4j.driver.internal.retry.RetrySettings;
-import org.neo4j.driver.internal.util.Clock;
-import org.neo4j.driver.internal.util.ConnectionTrackingDriverFactory;
 import org.neo4j.driver.internal.util.FakeClock;
 import org.neo4j.driver.internal.util.ThrowingConnection;
 import org.neo4j.driver.internal.util.ThrowingConnectionDriverFactory;
@@ -70,6 +68,7 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.startsWith;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
@@ -236,7 +235,7 @@ public Void apply( Session session )
     }
 
     @Test
-    public void shouldDropBrokenOldSessions() throws Exception
+    public void shouldDropBrokenOldConnections() throws Exception
     {
         Cluster cluster = clusterRule.getCluster();
 
@@ -249,19 +248,24 @@ public void shouldDropBrokenOldSessions() throws Exception
                 .toConfig();
 
         FakeClock clock = new FakeClock();
-        ConnectionTrackingDriverFactory driverFactory = new ConnectionTrackingDriverFactory( clock );
+        ThrowingConnectionDriverFactory driverFactory = new ThrowingConnectionDriverFactory( clock );
 
         URI routingUri = cluster.leader().getRoutingUri();
         AuthToken auth = clusterRule.getDefaultAuthToken();
         RetrySettings retrySettings = RetrySettings.DEFAULT;
 
         try ( Driver driver = driverFactory.newInstance( routingUri, auth, defaultRoutingSettings(), retrySettings, config ) )
         {
-            // create nodes in different threads using different sessions
+            // create nodes in different threads using different sessions and connections
             createNodesInDifferentThreads( concurrentSessionsCount, driver );
 
-            // now pool contains many sessions, make them all invalid
-            driverFactory.closeConnections();
+            // now pool contains many connections, make them all invalid
+            List<ThrowingConnection> oldConnections = driverFactory.pollConnections();
+            for ( ThrowingConnection oldConnection : oldConnections )
+            {
+                oldConnection.setNextResetError( new ServiceUnavailableException( "Unable to reset" ) );
+            }
+
             // move clock forward more than configured liveness check timeout
             clock.progress( MINUTES.toMillis( livenessCheckTimeoutMinutes + 1 ) );
 
@@ -273,6 +277,12 @@ public void shouldDropBrokenOldSessions() throws Exception
                 assertEquals( 1, records.size() );
                 assertEquals( concurrentSessionsCount, records.get( 0 ).get( 0 ).asInt() );
             }
+
+            // all old connections failed to reset and should be closed
+            for ( ThrowingConnection connection : oldConnections )
+            {
+                assertFalse( connection.isOpen() );
+            }
         }
     }
 
@@ -573,7 +583,7 @@ public void shouldRediscoverWhenConnectionsToAllCoresBreak()
             // make all those connections throw and seem broken
             for ( ThrowingConnection connection : driverFactory.getConnections() )
             {
-                connection.setNextRunFailure( new ServiceUnavailableException( "Disconnected" ) );
+                connection.setNextRunError( new ServiceUnavailableException( "Disconnected" ) );
             }
 
             // observe that connection towards writer is broken
@@ -620,7 +630,7 @@ public void shouldKeepOperatingWhenConnectionsBreak() throws Exception
         String value = "Tony Stark";
         Cluster cluster = clusterRule.getCluster();
 
-        ConnectionTrackingDriverFactory driverFactory = new ConnectionTrackingDriverFactory( Clock.SYSTEM );
+        ThrowingConnectionDriverFactory driverFactory = new ThrowingConnectionDriverFactory();
         AtomicBoolean stop = new AtomicBoolean();
         executor = newExecutor();
 
@@ -639,11 +649,15 @@ public void shouldKeepOperatingWhenConnectionsBreak() throws Exception
                 results.add( executor.submit( createNodesCallable( driver, label, property, value, stop ) ) );
             }
 
-            // terminate connections while reads and writes are in progress
+            // make connections throw while reads and writes are in progress
             long deadline = System.currentTimeMillis() + MINUTES.toMillis( 1 );
             while ( System.currentTimeMillis() < deadline && !stop.get() )
             {
-                driverFactory.closeConnections();
+                List<ThrowingConnection> connections = driverFactory.pollConnections();
+                for ( ThrowingConnection connection : connections )
+                {
+                    connection.setNextRunError( new ServiceUnavailableException( "Unable to execute query" ) );
+                }
                 SECONDS.sleep( 5 ); // sleep a bit to allow readers and writers to progress
             }
             stop.set( true );
@@ -676,7 +690,7 @@ private static void setupLastConnectionToThrow( ThrowingConnectionDriverFactory
     {
         List<ThrowingConnection> connections = factory.getConnections();
         ThrowingConnection lastConnection = connections.get( connections.size() - 1 );
-        lastConnection.setNextRunFailure( error );
+        lastConnection.setNextRunError( error );
     }
 
     private int executeWriteAndReadThroughBolt( ClusterMember member ) throws TimeoutException, InterruptedException