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

Zhen Li · Zhen Li · commit d63750045647 · 2019-10-11T14:26:31.000+02:00
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 -&gt; Long(1000)}' cannot be handled by a session in the READY state."
diff --git a/driver/src/main/java/org/neo4j/driver/internal/cursor/AsyncStatementResultCursorOnlyFactory.java b/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 )
diff --git a/driver/src/main/java/org/neo4j/driver/internal/cursor/StatementResultCursorFactoryImpl.java b/driver/src/main/java/org/neo4j/driver/internal/cursor/StatementResultCursorFactoryImpl.java
@@ -63,7 +63,7 @@ public StatementResultCursorFactoryImpl( Connection connection, Message runMessa
     public CompletionStage<AsyncStatementResultCursor> asyncResult()
     {
         // only write and flush messages when async result is wanted.
-        connection.writeAndFlush( runMessage, runHandler );
+        connection.write( runMessage, runHandler );
         pullAllHandler.prePopulateRecords();
 
         if ( waitForRunResponse )
diff --git a/driver/src/main/java/org/neo4j/driver/internal/handlers/pulln/AutoPullResponseHandler.java b/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();
-//        }
-//    }
 }
diff --git a/driver/src/test/java/org/neo4j/driver/internal/DirectDriverBoltKitTest.java b/driver/src/test/java/org/neo4j/driver/internal/DirectDriverBoltKitTest.java
@@ -303,6 +303,26 @@ void shouldChangeFetchSize() throws Exception
         }
     }
 
+    @Test
+    void shouldAllowPullAll() throws Exception
+    {
+        StubServer server = StubServer.start( "streaming_records_v4_all.script", 9001 );
+        try
+        {
+            try ( Driver driver = GraphDatabase.driver( "bolt://localhost:9001", insecureBuilder().withFetchSize( -1 ).build() ) )
+            {
+                Session session = driver.session();
+                StatementResult result = session.run( "MATCH (n) RETURN n.name" );
+                List<String> list = result.list( record -> record.get( "n.name" ).asString() );
+                assertEquals( list, asList( "Bob", "Alice", "Tina" ) );
+            }
+        }
+        finally
+        {
+            assertEquals( 0, server.exitStatus() );
+        }
+    }
+
     @Test
     void shouldThrowCommitErrorWhenTransactionCommit() throws Exception
     {
diff --git a/driver/src/test/java/org/neo4j/driver/internal/async/ExplicitTransactionTest.java b/driver/src/test/java/org/neo4j/driver/internal/async/ExplicitTransactionTest.java
@@ -52,9 +52,9 @@
 import static org.neo4j.driver.util.TestUtil.await;
 import static org.neo4j.driver.util.TestUtil.connectionMock;
 import static org.neo4j.driver.util.TestUtil.runMessageWithStatementMatcher;
-import static org.neo4j.driver.util.TestUtil.setupSuccessfulRun;
+import static org.neo4j.driver.util.TestUtil.setupSuccessfulRunRx;
 import static org.neo4j.driver.util.TestUtil.setupSuccessfulRunAndPull;
-import static org.neo4j.driver.util.TestUtil.verifyRun;
+import static org.neo4j.driver.util.TestUtil.verifyRunRx;
 import static org.neo4j.driver.util.TestUtil.verifyRunAndPull;
 
 class ExplicitTransactionTest
@@ -81,13 +81,13 @@ void shouldFlushOnRunRx()
         // Given
         Connection connection = connectionMock( BoltProtocolV4.INSTANCE );
         ExplicitTransaction tx = beginTx( connection );
-        setupSuccessfulRun( connection );
+        setupSuccessfulRunRx( connection );
 
         // When
         await( tx.runRx( new Statement( "RETURN 1" ) ) );
 
         // Then
-        verifyRun( connection, "RETURN 1" );
+        verifyRunRx( connection, "RETURN 1" );
     }
 
     @Test
diff --git a/driver/src/test/java/org/neo4j/driver/internal/async/NetworkSessionTest.java b/driver/src/test/java/org/neo4j/driver/internal/async/NetworkSessionTest.java
@@ -65,11 +65,11 @@
 import static org.neo4j.driver.util.TestUtil.connectionMock;
 import static org.neo4j.driver.util.TestUtil.newSession;
 import static org.neo4j.driver.util.TestUtil.setupFailingBegin;
-import static org.neo4j.driver.util.TestUtil.setupSuccessfulRun;
+import static org.neo4j.driver.util.TestUtil.setupSuccessfulRunRx;
 import static org.neo4j.driver.util.TestUtil.setupSuccessfulRunAndPull;
 import static org.neo4j.driver.util.TestUtil.verifyBeginTx;
 import static org.neo4j.driver.util.TestUtil.verifyRollbackTx;
-import static org.neo4j.driver.util.TestUtil.verifyRun;
+import static org.neo4j.driver.util.TestUtil.verifyRunRx;
 import static org.neo4j.driver.util.TestUtil.verifyRunAndPull;
 
 class NetworkSessionTest
@@ -101,10 +101,10 @@ void shouldFlushOnRunAsync( boolean waitForResponse )
     @Test
     void shouldFlushOnRunRx()
     {
-        setupSuccessfulRun( connection );
+        setupSuccessfulRunRx( connection );
         await( session.runRx( new Statement( "RETURN 1" ), TransactionConfig.empty() ) );
 
-        verifyRun( connection, "RETURN 1" );
+        verifyRunRx( connection, "RETURN 1" );
     }
 
     @Test
@@ -195,7 +195,7 @@ void releasesOpenConnectionUsedForRunWhenSessionIsClosed()
         close( session );
 
         InOrder inOrder = inOrder( connection );
-        inOrder.verify( connection ).writeAndFlush( any( RunWithMetadataMessage.class ), any() );
+        inOrder.verify( connection ).write( any( RunWithMetadataMessage.class ), any() );
         inOrder.verify( connection ).writeAndFlush( any( PullMessage.class ), any() );
         inOrder.verify( connection, atLeastOnce() ).release();
     }
diff --git a/driver/src/test/java/org/neo4j/driver/internal/cursor/AsyncStatementResultCursorOnlyFactoryTest.java b/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 ) );
     }
 }
diff --git a/driver/src/test/java/org/neo4j/driver/internal/cursor/StatementResultCursorFactoryImplTest.java b/driver/src/test/java/org/neo4j/driver/internal/cursor/StatementResultCursorFactoryImplTest.java
@@ -221,7 +221,7 @@ private StatementResultCursorFactoryImpl newResultCursorFactory( CompletableFutu
 
     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 ) );
     }
 
diff --git a/driver/src/test/java/org/neo4j/driver/internal/messaging/v1/BoltProtocolV1Test.java b/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 ) );
diff --git a/driver/src/test/java/org/neo4j/driver/internal/messaging/v3/BoltProtocolV3Test.java b/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 ) );
diff --git a/driver/src/test/java/org/neo4j/driver/internal/messaging/v4/BoltProtocolV4Test.java b/driver/src/test/java/org/neo4j/driver/internal/messaging/v4/BoltProtocolV4Test.java
@@ -224,7 +224,7 @@ private ResponseHandler verifyRunInvoked( Connection connection, RunWithMetadata
         ArgumentCaptor<ResponseHandler> runHandlerCaptor = ArgumentCaptor.forClass( ResponseHandler.class );
         ArgumentCaptor<ResponseHandler> pullHandlerCaptor = ArgumentCaptor.forClass( ResponseHandler.class );
 
-        verify( connection ).writeAndFlush( eq( runMessage ), runHandlerCaptor.capture() );
+        verify( connection ).write( eq( runMessage ), runHandlerCaptor.capture() );
         verify( connection ).writeAndFlush( any( PullMessage.class ), pullHandlerCaptor.capture() );
 
         assertThat( runHandlerCaptor.getValue(), instanceOf( RunResponseHandler.class ) );
diff --git a/driver/src/test/java/org/neo4j/driver/internal/util/FailingConnectionDriverFactory.java b/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;
             }
diff --git a/driver/src/test/java/org/neo4j/driver/stress/AbstractStressTestBase.java b/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() );
         }
     }
 
diff --git a/driver/src/test/java/org/neo4j/driver/stress/CausalClusteringIT.java b/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()
diff --git a/driver/src/test/java/org/neo4j/driver/util/TestUtil.java b/driver/src/test/java/org/neo4j/driver/util/TestUtil.java
@@ -272,14 +272,14 @@ public static NetworkSession newSession( ConnectionProvider connectionProvider,
                 DEV_NULL_LOGGING );
     }
 
-    public static void verifyRun( Connection connection, String query )
+    public static void verifyRunRx( Connection connection, String query )
     {
         verify( connection ).writeAndFlush( argThat( runWithMetaMessageWithStatementMatcher( query ) ), any() );
     }
 
     public static void verifyRunAndPull( Connection connection, String query )
     {
-        verify( connection ).writeAndFlush( argThat( runWithMetaMessageWithStatementMatcher( query ) ), any() );
+        verify( connection ).write( argThat( runWithMetaMessageWithStatementMatcher( query ) ), any() );
         verify( connection ).writeAndFlush( any( PullMessage.class ), any() );
     }
 
@@ -412,7 +412,7 @@ public static void setupSuccessfulRunAndPull( Connection connection )
             ResponseHandler runHandler = invocation.getArgument( 1 );
             runHandler.onSuccess( emptyMap() );
             return null;
-        } ).when( connection ).writeAndFlush( any( RunWithMetadataMessage.class ), any() );
+        } ).when( connection ).write( any( RunWithMetadataMessage.class ), any() );
 
         doAnswer( invocation ->
         {
@@ -422,7 +422,7 @@ public static void setupSuccessfulRunAndPull( Connection connection )
         } ).when( connection ).writeAndFlush( any( PullMessage.class ), any() );
     }
 
-    public static void setupSuccessfulRun( Connection connection )
+    public static void setupSuccessfulRunRx( Connection connection )
     {
         doAnswer( invocation ->
         {
@@ -439,7 +439,7 @@ public static void setupSuccessfulRunAndPull( Connection connection, String quer
             ResponseHandler runHandler = invocation.getArgument( 1 );
             runHandler.onSuccess( emptyMap() );
             return null;
-        } ).when( connection ).writeAndFlush( argThat( runWithMetaMessageWithStatementMatcher( query ) ), any() );
+        } ).when( connection ).write( argThat( runWithMetaMessageWithStatementMatcher( query ) ), any() );
 
         doAnswer( invocation ->
         {
diff --git a/driver/src/test/resources/streaming_records_v4_all.script b/driver/src/test/resources/streaming_records_v4_all.script
@@ -0,0 +1,12 @@
+!: BOLT 4
+!: AUTO RESET
+!: AUTO HELLO
+!: AUTO GOODBYE
+
+C: RUN "MATCH (n) RETURN n.name" {} {}
+   PULL { "n": -1 }
+S: SUCCESS {"fields": ["n.name"]}
+   RECORD ["Bob"]
+   RECORD ["Alice"]
+   RECORD ["Tina"]
+   SUCCESS {}

Original file line number	Diff line number	Diff line change
`@@ -60,7 +60,7 @@ public AsyncStatementResultCursorOnlyFactory( Connection connection, Message run`
`60`	`60`	`public CompletionStage<AsyncStatementResultCursor> asyncResult()`
`61`	`61`	`{`
`62`	`62`	`// only write and flush messages when async result is wanted.`
`63`		`- connection.writeAndFlush( runMessage, runHandler );`
	`63`	`+ connection.write( runMessage, runHandler ); // queues the run message, will be flushed with pull message together`
`64`	`64`	`pullAllHandler.prePopulateRecords();`
`65`	`65`
`66`	`66`	`if ( waitForRunResponse )`
Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ public StatementResultCursorFactoryImpl( Connection connection, Message runMessa`
`63`	`63`	`public CompletionStage<AsyncStatementResultCursor> asyncResult()`
`64`	`64`	`{`
`65`	`65`	`// only write and flush messages when async result is wanted.`
`66`		`- connection.writeAndFlush( runMessage, runHandler );`
	`66`	`+ connection.write( runMessage, runHandler );`
`67`	`67`	`pullAllHandler.prePopulateRecords();`
`68`	`68`
`69`	`69`	`if ( waitForRunResponse )`
Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,6 @@`
`42`	`42`	`import static org.mockito.ArgumentMatchers.any;`
`43`	`43`	`import static org.mockito.Mockito.mock;`
`44`	`44`	`import static org.mockito.Mockito.verify;`
`45`		`-import static org.mockito.Mockito.verifyNoMoreInteractions;`
`46`	`45`	`import static org.mockito.Mockito.when;`
`47`	`46`	`import static org.neo4j.driver.internal.util.Futures.completedWithNull;`
`48`	`47`	`import static org.neo4j.driver.internal.util.Futures.failedFuture;`
`@@ -174,7 +173,7 @@ private AsyncStatementResultCursorOnlyFactory newResultCursorFactory( Completabl`
`174`	`173`
`175`	`174`	`private void verifyRunCompleted( Connection connection, CompletionStage<AsyncStatementResultCursor> cursorFuture )`
`176`	`175`	`{`
`177`		`- verify( connection ).writeAndFlush( any( Message.class ), any( RunResponseHandler.class ) );`
	`176`	`+ verify( connection ).write( any( Message.class ), any( RunResponseHandler.class ) );`
`178`	`177`	`assertThat( getNow( cursorFuture ), instanceOf( AsyncStatementResultCursorImpl.class ) );`
`179`	`178`	`}`
`180`	`179`	`}`
Original file line number	Diff line number	Diff line change
`@@ -221,7 +221,7 @@ private StatementResultCursorFactoryImpl newResultCursorFactory( CompletableFutu`
`221`	`221`
`222`	`222`	`private void verifyRunCompleted( Connection connection, CompletionStage<AsyncStatementResultCursor> cursorFuture )`
`223`	`223`	`{`
`224`		`- verify( connection ).writeAndFlush( any( Message.class ), any( RunResponseHandler.class ) );`
	`224`	`+ verify( connection ).write( any( Message.class ), any( RunResponseHandler.class ) );`
`225`	`225`	`assertThat( getNow( cursorFuture ), instanceOf( AsyncStatementResultCursorImpl.class ) );`
`226`	`226`	`}`
`227`	`227`
Original file line number	Diff line number	Diff line change
`@@ -473,7 +473,7 @@ private static ResponseHandlers verifyRunInvoked( Connection connection, boolean`
`473`	`473`	`expectedMessage = RunWithMetadataMessage.explicitTxRunMessage( STATEMENT );`
`474`	`474`	`}`
`475`	`475`
`476`		`- verify( connection ).writeAndFlush( eq( expectedMessage ), runHandlerCaptor.capture() );`
	`476`	`+ verify( connection ).write( eq( expectedMessage ), runHandlerCaptor.capture() );`
`477`	`477`	`verify( connection ).writeAndFlush( eq( PullAllMessage.PULL_ALL ), pullAllHandlerCaptor.capture() );`
`478`	`478`
`479`	`479`	`assertThat( runHandlerCaptor.getValue(), instanceOf( RunResponseHandler.class ) );`