Resolve issue where pooled connections could leak if exception occurred

during close.

- close in PooledConnection calls reset and then returns the session to
  the pool. However, returning the session was not in a finally clause,
  so reset throwing an exception would leak the connection, eventually
  draining the pool from available sessions.

Author: jakewins

driver/src/main/java/org/neo4j/driver/internal/pool/PooledConnection.java

@@ -156,12 +156,15 @@ public void close()
         {
             reset( StreamCollector.NO_OP );
             sync();
-            release.accept( this );
         }
         catch (Exception ex)
         {
             dispose();
         }
+        finally
+        {
+            release.accept( this );
+        }
     }
 
     @Override

driver/src/test/java/org/neo4j/driver/internal/pool/PooledConnectionTest.java

@@ -0,0 +1,46 @@
+package org.neo4j.driver.internal.pool;
+
+import org.junit.Test;
+
+import java.util.LinkedList;
+
+import org.neo4j.driver.internal.spi.Connection;
+import org.neo4j.driver.internal.spi.StreamCollector;
+import org.neo4j.driver.internal.util.Consumer;
+import org.neo4j.driver.v1.exceptions.DatabaseException;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+
+public class PooledConnectionTest
+{
+    @Test
+    public void shouldReturnToPoolIfExceptionDuringReset() throws Throwable
+    {
+        // Given
+        final LinkedList<PooledConnection> returnedToPool = new LinkedList<>();
+        Connection conn = mock( Connection.class );
+
+        doThrow( new DatabaseException( "asd", "asd" ) ).when(conn).reset( any( StreamCollector.class) );
+
+        PooledConnection pooledConnection = new PooledConnection( conn, new Consumer<PooledConnection>()
+        {
+            @Override
+            public void accept( PooledConnection pooledConnection )
+            {
+                returnedToPool.add( pooledConnection );
+            }
+        } );
+
+        // When
+        pooledConnection.close();
+
+        // Then
+        assertThat( returnedToPool, hasItem(pooledConnection) );
+        assertThat( returnedToPool.size(), equalTo( 1 ));
+    }
+}