Fixed a bug found with ClusterStressIT regarding how we send run and pull messages.

When creating the async result, we write a RUN message, followed by a PULL message.
The RUN and PULL messages shall be flushed together.

If RUN and PULL are flushed separately, the following scenario may happen:

C: RUN "RETURN Wrong" {} {mode="r"}
S: FAILURE Neo.ClientError.Statement.SyntaxError "Variable `Wrong` not defined (line 1, column 8 (offset: 7))
C: RESET
C: PULL {n=1000}
S: SUCCESS {}
S: FAILURE Neo.ClientError.Request.Invalid "Message 'PULL Map{n -> Long(1000)}' cannot be handled by a session in the READY state."

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/internal/cursor/AsyncStatementResultCursorOnlyFactory.java

@@ -60,7 +60,7 @@ public AsyncStatementResultCursorOnlyFactory( Connection connection, Message run
     public CompletionStage<AsyncStatementResultCursor> asyncResult()
     {
         // only write and flush messages when async result is wanted.
-        connection.writeAndFlush( runMessage, runHandler );
+        connection.write( runMessage, runHandler ); // queues the run message, will be flushed with pull message together
         pullAllHandler.prePopulateRecords();
 
         if ( waitForRunResponse )

driver/src/main/java/org/neo4j/driver/internal/handlers/pulln/AutoPullResponseHandler.java

@@ -48,15 +48,11 @@
 public class AutoPullResponseHandler extends BasicPullResponseHandler implements PullAllResponseHandler
 {
     private static final Queue<Record> UNINITIALIZED_RECORDS = Iterables.emptyQueue();
-
-    private static final int RECORD_BUFFER_LOW_WATERMARK = Integer.getInteger( "recordBufferLowWatermark", 300 );
-    private static final int RECORD_BUFFER_HIGH_WATERMARK = Integer.getInteger( "recordBufferHighWatermark", 1000 );
     private final long fetchSize;
 
     // initialized lazily when first record arrives
     private Queue<Record> records = UNINITIALIZED_RECORDS;
 
-//    private boolean autoReadManagementEnabled = true;
     private ResultSummary summary;
     private Throwable failure;
 
@@ -113,18 +109,6 @@ private void handleFailure( Throwable error )
             failure = error;
         }
     }
-//
-//    @Override
-//    public boolean canManageAutoRead()
-//    {
-//        return true;
-//    }
-
-//    @Override
-//    public synchronized void disableAutoReadManagement()
-//    {
-//        autoReadManagementEnabled = false;
-//    }
 
     public synchronized CompletionStage<Record> peekAsync()
     {
@@ -198,7 +182,6 @@ private synchronized CompletionStage<ResultSummary> pullAllAsync()
         }
         else
         {
-//            enableAutoRead();
             request( UNLIMITED_FETCH_SIZE );
             if ( summaryFuture == null )
             {
@@ -217,32 +200,11 @@ private void enqueueRecord( Record record )
         }
 
         records.add( record );
-
-//        boolean shouldBufferAllRecords = failureFuture != null;
-//        // when failure is requested we have to buffer all remaining records and then return the error
-//        // do not disable auto-read in this case, otherwise records will not be consumed and trailing
-//        // SUCCESS or FAILURE message will not arrive as well, so callers will get stuck waiting for the error
-//        if ( !shouldBufferAllRecords && records.size() > RECORD_BUFFER_HIGH_WATERMARK )
-//        {
-//            // more than high watermark records are already queued, tell connection to stop auto-reading from network
-//            // this is needed to deal with slow consumers, we do not want to buffer all records in memory if they are
-//            // fetched from network faster than consumed
-//            disableAutoRead();
-//        }
     }
 
     private Record dequeueRecord()
     {
-        Record record = records.poll();
-
-//        if ( records.size() < RECORD_BUFFER_LOW_WATERMARK )
-//        {
-//            // less than low watermark records are now available in the buffer, tell connection to pre-fetch more
-//            // and populate queue with new records from network
-//            enableAutoRead();
-//        }
-
-        return record;
+        return records.poll();
     }
 
     private <T> List<T> recordsAsList( Function<Record,T> mapFunction )
@@ -332,20 +294,4 @@ else if ( value == null )
             return completedFuture( value );
         }
     }
-
-//    private void enableAutoRead()
-//    {
-//        if ( autoReadManagementEnabled )
-//        {
-//            connection.enableAutoRead();
-//        }
-//    }
-//
-//    private void disableAutoRead()
-//    {
-//        if ( autoReadManagementEnabled )
-//        {
-//            connection.disableAutoRead();
-//        }
-//    }
 }

driver/src/test/java/org/neo4j/driver/internal/cursor/AsyncStatementResultCursorOnlyFactoryTest.java

@@ -42,7 +42,6 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 import static org.neo4j.driver.internal.util.Futures.completedWithNull;
 import static org.neo4j.driver.internal.util.Futures.failedFuture;
@@ -174,7 +173,7 @@ private AsyncStatementResultCursorOnlyFactory newResultCursorFactory( Completabl
 
     private void verifyRunCompleted( Connection connection, CompletionStage<AsyncStatementResultCursor> cursorFuture )
     {
-        verify( connection ).writeAndFlush( any( Message.class ), any( RunResponseHandler.class ) );
+        verify( connection ).write( any( Message.class ), any( RunResponseHandler.class ) );
         assertThat( getNow( cursorFuture ), instanceOf( AsyncStatementResultCursorImpl.class ) );
     }
 }

driver/src/test/java/org/neo4j/driver/internal/messaging/v1/BoltProtocolV1Test.java

@@ -369,7 +369,7 @@ private static ResponseHandler verifyRunInvoked( Connection connection )
         ArgumentCaptor<ResponseHandler> runHandlerCaptor = ArgumentCaptor.forClass( ResponseHandler.class );
         ArgumentCaptor<ResponseHandler> pullAllHandlerCaptor = ArgumentCaptor.forClass( ResponseHandler.class );
 
-        verify( connection ).writeAndFlush( eq( new RunMessage( QUERY, PARAMS ) ), runHandlerCaptor.capture() );
+        verify( connection ).write( eq( new RunMessage( QUERY, PARAMS ) ), runHandlerCaptor.capture() );
         verify( connection ).writeAndFlush( eq( PullAllMessage.PULL_ALL ), pullAllHandlerCaptor.capture() );
 
         assertThat( runHandlerCaptor.getValue(), instanceOf( RunResponseHandler.class ) );

driver/src/test/java/org/neo4j/driver/internal/messaging/v3/BoltProtocolV3Test.java

@@ -473,7 +473,7 @@ private static ResponseHandlers verifyRunInvoked( Connection connection, boolean
             expectedMessage = RunWithMetadataMessage.explicitTxRunMessage( STATEMENT );
         }
 
-        verify( connection ).writeAndFlush( eq( expectedMessage ), runHandlerCaptor.capture() );
+        verify( connection ).write( eq( expectedMessage ), runHandlerCaptor.capture() );
         verify( connection ).writeAndFlush( eq( PullAllMessage.PULL_ALL ), pullAllHandlerCaptor.capture() );
 
         assertThat( runHandlerCaptor.getValue(), instanceOf( RunResponseHandler.class ) );

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

@@ -22,6 +22,7 @@
 
 import java.util.Set;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.neo4j.driver.internal.BoltServerAddress;
@@ -106,6 +107,7 @@ private static class FailingConnection implements Connection
     {
         final Connection delegate;
         final AtomicReference<Throwable> nextRunFailure;
+        final AtomicInteger count = new AtomicInteger( 2 ); // one failure for run, one failure for pull
 
         FailingConnection( Connection delegate, AtomicReference<Throwable> nextRunFailure )
         {
@@ -222,13 +224,21 @@ private boolean tryFail( ResponseHandler handler1, ResponseHandler handler2 )
             Throwable failure = nextRunFailure.getAndSet( null );
             if ( failure != null )
             {
+                int reportCount = count.get();
                 if ( handler1 != null )
                 {
                     handler1.onFailure( failure );
+                    reportCount = count.decrementAndGet();
                 }
                 if ( handler2 != null )
                 {
                     handler2.onFailure( failure );
+                    reportCount = count.decrementAndGet();
+                }
+
+                if ( reportCount > 0 )
+                {
+                    nextRunFailure.compareAndSet( null, failure );
                 }
                 return true;
             }

driver/src/test/java/org/neo4j/driver/stress/AbstractStressTestBase.java

@@ -130,7 +130,6 @@ void tearDown()
         if ( driver != null )
         {
             driver.close();
-            System.out.println( driver.metrics() );
         }
     }
 

driver/src/test/java/org/neo4j/driver/stress/CausalClusteringIT.java

@@ -39,6 +39,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
+import java.util.logging.Level;
 
 import org.neo4j.driver.AccessMode;
 import org.neo4j.driver.AuthToken;
@@ -89,6 +90,7 @@
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.neo4j.driver.Logging.console;
 import static org.neo4j.driver.Values.parameters;
 import static org.neo4j.driver.SessionConfig.builder;
 import static org.neo4j.driver.internal.InternalBookmark.parse;
@@ -1018,7 +1020,7 @@ private static int runCountNodes( StatementRunner statementRunner, String label,
 
     private static Config configWithoutLogging()
     {
-        return Config.builder().withLogging( DEV_NULL_LOGGING ).build();
+        return Config.builder().withLogging( console( Level.FINE ) ).build();
     }
 
     private static ExecutorService newExecutor()