When receiving unrecoverable errors, do not ack failure or queuing any more message but just wait to be closed when putting back to session pool

Author: Zhen

driver/src/main/java/org/neo4j/driver/internal/InternalSession.java

@@ -148,9 +148,14 @@ public void run()
                 //must check if transaction has been closed
                 if (currentTransaction != null)
                 {
-                    currentTransaction.failure();
-                    currentTransaction = null;
-                    connection.onError( null );
+                    if( connection.hasUnrecoverableErrors() )
+                    {
+                        currentTransaction.markToClose();
+                    }
+                    else
+                    {
+                        currentTransaction.failure();
+                    }
                 }
             }
         });
@@ -165,12 +170,14 @@ public TypeSystem typeSystem()
 
     private void ensureConnectionIsValidBeforeRunningSession()
     {
+        ensureNoUnrecoverableError();
         ensureNoOpenTransactionBeforeRunningSession();
         ensureConnectionIsOpen();
     }
 
     private void ensureConnectionIsValidBeforeOpeningTransaction()
     {
+        ensureNoUnrecoverableError();
         ensureNoOpenTransactionBeforeOpeningTransaction();
         ensureConnectionIsOpen();
     }
@@ -187,6 +194,16 @@ protected void finalize() throws Throwable
         super.finalize();
     }
 
+    private void ensureNoUnrecoverableError()
+    {
+        if( connection.hasUnrecoverableErrors() )
+        {
+            throw new ClientException( "Cannot run more statements in the current session as unrecoverable errors " +
+                                       "has happened. Please close the currect session and re-run your statement in a" +
+                                       " new session." );
+        }
+    }
+
     private void ensureNoOpenTransactionBeforeRunningSession()
     {
         if ( currentTransaction != null )

driver/src/main/java/org/neo4j/driver/internal/InternalTransaction.java

@@ -200,5 +200,8 @@ public TypeSystem typeSystem()
         return InternalTypeSystem.TYPE_SYSTEM;
     }
 
-
+    public void markToClose()
+    {
+        state = State.FAILED;
+    }
 }

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

@@ -195,6 +195,12 @@ public void onError( Runnable runnable )
         delegate.onError( runnable );
     }
 
+    @Override
+    public boolean hasUnrecoverableErrors()
+    {
+        return delegate.hasUnrecoverableErrors();
+    }
+
     private void markAsAvailable()
     {
         inUse.set( false );

driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketConnection.java

@@ -153,7 +153,6 @@ private void assertNoServerFailure()
     {
         if ( responseHandler.serverFailureOccurred() )
         {
-            ackFailure( StreamCollector.NO_OP );
             Neo4jException exception = responseHandler.serverFailure();
             responseHandler.clearError();
             throw exception;
@@ -200,4 +199,10 @@ public void onError( Runnable runnable )
     {
         throw new UnsupportedOperationException( "Error subscribers are not supported on SocketConnection." );
     }
+
+    @Override
+    public boolean hasUnrecoverableErrors()
+    {
+        throw new UnsupportedOperationException( "Unrecoverable error detection is not supported on SocketConnection." );
+    }
 }

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

@@ -228,6 +228,10 @@ private void onDelegateException( RuntimeException e )
         {
             unrecoverableErrorsOccurred = true;
         }
+        else
+        {
+            ackFailure( StreamCollector.NO_OP );
+        }
         if( onError != null )
         {
             onError.run();

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

@@ -87,8 +87,10 @@ else if ( driverStopped.get() )
 
     boolean validConnection( PooledConnection pooledConnection )
     {
-        return reset(pooledConnection) &&
-               !pooledConnection.hasUnrecoverableErrors() &&
+        // once the pooledConn has marked to have unrecoverable errors, there is no way to remove the error
+        // and we should close the conn without bother to reset the conn at all
+        return !pooledConnection.hasUnrecoverableErrors() &&
+               reset(pooledConnection) &&
                (pooledConnection.idleTime() <= minIdleBeforeConnectionTest || ping( pooledConnection ));
     }
 

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

@@ -102,4 +102,7 @@ public interface Connection extends AutoCloseable
      * @param runnable
      */
     void onError( Runnable runnable );
+
+
+    boolean hasUnrecoverableErrors();
 }