Made reset method sync so that once isInterrupted atomic filed get set, no more messages will be enqueued or flushed until the current sent ones has been synced.

Always ackFail

Author: Zhen

driver/src/main/java/org/neo4j/driver/internal/net/ConcurrencyGuardingConnection.java

@@ -207,12 +207,6 @@ public void resetAsync()
         delegate.resetAsync();
     }
 
-    @Override
-    public boolean isInterrupted()
-    {
-        return delegate.isInterrupted();
-    }
-
     private void markAsAvailable()
     {
         inUse.set( false );

driver/src/main/java/org/neo4j/driver/internal/net/SocketConnection.java

@@ -46,7 +46,7 @@ public class SocketConnection implements Connection
 {
     private final Queue<Message> pendingMessages = new LinkedList<>();
     private final SocketResponseHandler responseHandler;
-    private AtomicBoolean interrupted = new AtomicBoolean( false );
+    private AtomicBoolean isInterrupted = new AtomicBoolean( false );
     private final Collector.InitCollector initCollector = new Collector.InitCollector();
 
     private final SocketClient socket;
@@ -115,6 +115,8 @@ public void sync()
     @Override
     public synchronized void flush()
     {
+        ensureNotInterrupted();
+
         try
         {
             socket.send( pendingMessages );
@@ -126,6 +128,26 @@ public synchronized void flush()
         }
     }
 
+    private void ensureNotInterrupted()
+    {
+        try
+        {
+            if( isInterrupted.get() )
+            {
+                receiveAll();
+            }
+        }
+        catch ( Neo4jException e )
+        {
+            throw new ClientException(
+                    "Failed to execute more statements as the session has been reset " +
+                    "and an error has occurred due to the cancellation of executing the previous statement. " +
+                    "You received this error probably because you did not consume the result immediately after " +
+                    "running the statement which get cancelled in this session.", e );
+        }
+
+    }
+
     private void receiveAll()
     {
         try
@@ -159,6 +181,7 @@ private void assertNoServerFailure()
         {
             Neo4jException exception = responseHandler.serverFailure();
             responseHandler.clearError();
+            isInterrupted.set( false );
             throw exception;
         }
     }
@@ -182,6 +205,8 @@ else if ( e instanceof SocketTimeoutException )
 
     private synchronized void queueMessage( Message msg, Collector collector )
     {
+        ensureNotInterrupted();
+
         pendingMessages.add( msg );
         responseHandler.appendResultCollector( collector );
     }
@@ -211,26 +236,18 @@ public boolean hasUnrecoverableErrors()
     }
 
     @Override
-    public void resetAsync()
+    public synchronized void resetAsync()
     {
-        if( interrupted.compareAndSet( false, true ) )
+        queueMessage( RESET, new Collector.ResetCollector( new Runnable()
         {
-            queueMessage( RESET, new Collector.ResetCollector( new Runnable()
+            @Override
+            public void run()
             {
-                @Override
-                public void run()
-                {
-                    interrupted.set( false );
-                }
-            } ) );
-            flush();
-        }
-    }
-
-    @Override
-    public boolean isInterrupted()
-    {
-        return interrupted.get();
+                isInterrupted.set( false );
+            }
+        } ) );
+        flush();
+        isInterrupted.set( true );
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/net/pooling/PooledConnection.java

@@ -225,12 +225,6 @@ public void resetAsync()
         }
     }
 
-    @Override
-    public boolean isInterrupted()
-    {
-        return delegate.isInterrupted();
-    }
-
     @Override
     public String server()
     {
@@ -260,7 +254,7 @@ private void onDelegateException( RuntimeException e )
         {
             unrecoverableErrorsOccurred = true;
         }
-        else if( !isInterrupted() )
+        else
         {
             ackFailure();
         }

driver/src/main/java/org/neo4j/driver/internal/spi/Connection.java

@@ -113,12 +113,6 @@ public interface Connection extends AutoCloseable
      */
     void resetAsync();
 
-    /**
-     * Return true if the current session statement execution has been interrupted by another thread, otherwise false.
-     * @return true if the current session statement execution has been interrupted by another thread, otherwise false
-     */
-    boolean isInterrupted();
-
     /**
      * Returns the version of the server connected to.
      * @return The version of the server connected to.

driver/src/test/java/org/neo4j/driver/v1/integration/TransactionIT.java

@@ -32,9 +32,11 @@
 import org.neo4j.driver.v1.Transaction;
 import org.neo4j.driver.v1.Value;
 import org.neo4j.driver.v1.exceptions.ClientException;
+import org.neo4j.driver.v1.exceptions.TransientException;
 import org.neo4j.driver.v1.util.TestNeo4jSession;
 
 import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.startsWith;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -215,24 +217,64 @@ public void shouldHandleNullMapParameters() throws Throwable
         // Then it wasn't the end of the world as we know it
     }
 
+
     @Test
-    public void shouldHandleReset() throws Throwable
+    public void shouldThrowExceptionDueToNotConsumingResetFailure() throws Throwable
     {
-            Transaction tx = session.beginTransaction();
-            tx.run( "UNWIND range(1,1) AS n RETURN n AS number" );
-            //TODO I believe there is a bug server side, if the RESET comes in
-            //too fast, stuff breaks
-            Thread.sleep( 100 );
-            session.reset();
-            tx = session.beginTransaction();
-            tx.run( "CREATE (n:FirstNode)" );
-            tx.success();
-            tx.close();
+        // Given
+        Transaction tx = session.beginTransaction();
+        tx.run( "UNWIND range(1,1) AS n RETURN n AS number" );
+        session.reset();
+
+        exception.expect( ClientException.class );
+        exception.expectMessage( startsWith(
+                "Failed to execute more statements as the session has been reset " +
+                "and an error has occurred due to the cancellation of executing the previous statement." ) );
+
+        // When & Then
+        tx = session.beginTransaction();
+    }
+
+    @Test
+    public void shouldBeAbleToRunMoreStatementAfterConsumingResetFailure() throws Throwable
+    {
+        Transaction tx = session.beginTransaction();
+        StatementResult failingResult = tx.run( "UNWIND range(1,1) AS n RETURN n AS number" );
+        session.reset();
+
+        exception.expect( TransientException.class );
+        exception.expectMessage( startsWith(
+                "The transaction has been terminated." ) );
+
+        failingResult.consume(); // fail with errors
+
+        tx = session.beginTransaction();
+        tx.run( "CREATE (n:FirstNode)" );
+        tx.success();
+        tx.close();
 
-            // Then the outcome of both statements should be visible
-            StatementResult result = session.run( "MATCH (n) RETURN count(n)" );
-            long nodes = result.single().get( "count(n)" ).asLong();
-            assertThat( nodes, equalTo( 1L ) );
+        // Then the outcome of both statements should be visible
+        StatementResult result = session.run( "MATCH (n) RETURN count(n)" );
+        long nodes = result.single().get( "count(n)" ).asLong();
+        assertThat( nodes, equalTo( 1L ) );
+    }
+
+    @Test
+    public void shouldBeAbleToRunMoreStatementAfterEmptyReset() throws Throwable
+    {
+        // Given
+        session.reset();
+
+        // When
+        Transaction tx = session.beginTransaction();
+        tx.run( "CREATE (n:FirstNode)" );
+        tx.success();
+        tx.close();
+
+        // Then the outcome of both statements should be visible
+        StatementResult result = session.run( "MATCH (n) RETURN count(n)" );
+        long nodes = result.single().get( "count(n)" ).asLong();
+        assertThat( nodes, equalTo( 1L ) );
     }
 
     @Test