Merge pull request #681 from zhenlineo/4.0-fix-rx-stress-it

zhenlineo · web-flow · commit a410b78a8978 · 2020-04-09T10:27:49.000+02:00
Ensure error is propagated from rx queries correctly.
diff --git a/driver/src/test/java/org/neo4j/driver/stress/AbstractStressTestBase.java b/driver/src/test/java/org/neo4j/driver/stress/AbstractStressTestBase.java
@@ -30,7 +30,6 @@
 import java.lang.management.OperatingSystemMXBean;
 import java.lang.reflect.Method;
 import java.net.URI;
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -99,7 +98,6 @@ abstract class AbstractStressTestBase<C extends AbstractContext>
 
     private static final int BIG_DATA_TEST_NODE_COUNT = Integer.getInteger( "bigDataTestNodeCount", 30_000 );
     private static final int BIG_DATA_TEST_BATCH_SIZE = Integer.getInteger( "bigDataTestBatchSize", 10_000 );
-    private static final Duration DEFAULT_BLOCKING_TIME_OUT = Duration.ofMinutes( 10 );
 
     private LoggerNameTrackingLogging logging;
     private ExecutorService executor;
@@ -638,7 +636,7 @@ private Bookmark createNodesRx( int batchCount, int batchSize, InternalDriver dr
 
         Flux.concat( Flux.range( 0, batchCount ).map( batchIndex ->
             session.writeTransaction( tx -> createNodesInTxRx( tx, batchIndex, batchSize ) )
-        ) ).blockLast( DEFAULT_BLOCKING_TIME_OUT ); // throw any error if happened
+        ) ).blockLast(); // throw any error if happened
 
         long end = System.nanoTime();
         System.out.println( "Node creation with reactive API took: " + NANOSECONDS.toMillis( end - start ) + "ms" );
@@ -673,7 +671,7 @@ private void readNodesRx( InternalDriver driver, Bookmark bookmark, int expected
             verifyNodeProperties( node );
         } ).then() );
 
-        Flux.from( readQuery ).blockLast( DEFAULT_BLOCKING_TIME_OUT );
+        Flux.from( readQuery ).blockLast();
 
         assertEquals( expectedNodeCount, nodesSeen.get() );
 
diff --git a/driver/src/test/java/org/neo4j/driver/stress/AsyncWriteQuery.java b/driver/src/test/java/org/neo4j/driver/stress/AsyncWriteQuery.java
@@ -55,6 +55,7 @@ public CompletionStage<Void> execute( C context )
                     }
                     else
                     {
+                        context.setBookmark( session.lastBookmark() );
                         assertEquals( 1, summary.counters().nodesCreated() );
                         context.nodeCreated();
                     }
diff --git a/driver/src/test/java/org/neo4j/driver/stress/AsyncWriteQueryInTx.java b/driver/src/test/java/org/neo4j/driver/stress/AsyncWriteQueryInTx.java
@@ -45,8 +45,10 @@ public CompletionStage<Void> execute( C context )
 
         CompletionStage<ResultSummary> txCommitted = session.beginTransactionAsync().thenCompose( tx ->
                 tx.runAsync( "CREATE ()" ).thenCompose( cursor ->
-                        cursor.consumeAsync().thenCompose( summary ->
-                                tx.commitAsync().thenApply( ignore -> summary ) ) ) );
+                        cursor.consumeAsync().thenCompose( summary -> tx.commitAsync().thenApply( ignore -> {
+                            context.setBookmark( session.lastBookmark() );
+                            return summary;
+                        } ) ) ) );
 
         return txCommitted.handle( ( summary, error ) ->
         {
diff --git a/driver/src/test/java/org/neo4j/driver/stress/BlockingWriteQuery.java b/driver/src/test/java/org/neo4j/driver/stress/BlockingWriteQuery.java
@@ -21,7 +21,7 @@
 import org.neo4j.driver.AccessMode;
 import org.neo4j.driver.Driver;
 import org.neo4j.driver.Session;
-import org.neo4j.driver.Result;
+import org.neo4j.driver.summary.ResultSummary;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
@@ -38,12 +38,13 @@ public BlockingWriteQuery( AbstractStressTestBase<C> stressTest, Driver driver,
     @Override
     public void execute( C context )
     {
-        Result result = null;
+        ResultSummary summary = null;
         Throwable queryError = null;
 
         try ( Session session = newSession( AccessMode.WRITE, context ) )
         {
-            result = session.run( "CREATE ()" );
+            summary = session.run( "CREATE ()" ).consume();
+            context.setBookmark( session.lastBookmark() );
         }
         catch ( Throwable error )
         {
@@ -54,9 +55,9 @@ public void execute( C context )
             }
         }
 
-        if ( queryError == null && result != null )
+        if ( queryError == null && summary != null )
         {
-            assertEquals( 1, result.consume().counters().nodesCreated() );
+            assertEquals( 1, summary.counters().nodesCreated() );
             context.nodeCreated();
         }
     }
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxFailingQuery.java b/driver/src/test/java/org/neo4j/driver/stress/RxFailingQuery.java
@@ -23,6 +23,7 @@
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.neo4j.driver.AccessMode;
 import org.neo4j.driver.Driver;
@@ -51,8 +52,7 @@ public CompletionStage<Void> execute( C context )
                 RxSession::close )
                 .subscribe( record -> {
                     assertThat( record.get( 0 ).asInt(), either( equalTo( 1 ) ).or( equalTo( 2 ) ) );
-                    queryFinished.complete( null );
-                }, error -> {
+                    }, error -> {
                     Throwable cause = Futures.completionExceptionCause( error );
                     assertThat( cause, is( arithmeticError() ) );
                     queryFinished.complete( null );
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxFailingQueryInTx.java b/driver/src/test/java/org/neo4j/driver/stress/RxFailingQueryInTx.java
@@ -52,7 +52,6 @@ public CompletionStage<Void> execute( C context )
                 RxTransaction::commit, ( tx, error ) -> tx.rollback(), null )
                 .subscribe( record -> {
                     assertThat( record.get( 0 ).asInt(), either( equalTo( 1 ) ).or( equalTo( 2 ) ) );
-                    queryFinished.complete( null );
                 }, error -> {
                     Throwable cause = Futures.completionExceptionCause( error );
                     assertThat( cause, is( arithmeticError() ) );
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxFailingQueryWithRetries.java b/driver/src/test/java/org/neo4j/driver/stress/RxFailingQueryWithRetries.java
@@ -51,7 +51,6 @@ public CompletionStage<Void> execute( C context )
                 RxSession::close )
                 .subscribe( record -> {
                     assertThat( record.get( 0 ).asInt(), either( equalTo( 1 ) ).or( equalTo( 2 ) ) );
-                    queryFinished.complete( null );
                 }, error -> {
                     Throwable cause = Futures.completionExceptionCause( error );
                     assertThat( cause, is( arithmeticError() ) );
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxWriteQuery.java b/driver/src/test/java/org/neo4j/driver/stress/RxWriteQuery.java
@@ -47,24 +47,27 @@ public CompletionStage<Void> execute( C context )
     {
         CompletableFuture<Void> queryFinished = new CompletableFuture<>();
         Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.WRITE, context ) ),
-                session -> session.run( "CREATE ()" ).consume(), RxSession::close )
+                session -> Flux.from( session.run( "CREATE ()" ).consume() )
+                        .doOnComplete( () -> context.setBookmark( session.lastBookmark() ) ),
+                RxSession::close )
                 .subscribe( summary -> {
-                    queryFinished.complete( null );
                     assertEquals( 1, summary.counters().nodesCreated() );
                     context.nodeCreated();
-                }, error -> {
                     queryFinished.complete( null );
-                    handleError( Futures.completionExceptionCause( error ), context );
-                } );
+                }, error -> handleError( Futures.completionExceptionCause( error ), context, queryFinished ) );
 
         return queryFinished;
     }
 
-    private void handleError( Throwable error, C context )
+    private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )
     {
         if ( !stressTest.handleWriteFailure( error, context ) )
         {
-            throw new RuntimeException( error );
+            queryFinished.completeExceptionally( error );
+        }
+        else
+        {
+            queryFinished.complete( null );
         }
     }
 }
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxWriteQueryInTx.java b/driver/src/test/java/org/neo4j/driver/stress/RxWriteQueryInTx.java
@@ -49,22 +49,26 @@ public CompletionStage<Void> execute( C context )
         Flux.usingWhen( session.beginTransaction(), tx -> tx.run( "CREATE ()" ).consume(),
                 RxTransaction::commit, ( tx, error ) -> tx.rollback(), null ).subscribe(
                 summary -> {
+                    context.setBookmark( session.lastBookmark() );
                     assertEquals( 1, summary.counters().nodesCreated() );
                     context.nodeCreated();
                     queryFinished.complete( null );
                 }, error -> {
-                    handleError( Futures.completionExceptionCause( error ), context );
-                    queryFinished.complete( null );
+                    handleError( Futures.completionExceptionCause( error ), context, queryFinished );
                 } );
 
         return queryFinished;
     }
 
-    private void handleError( Throwable error, C context )
+    private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )
     {
         if ( !stressTest.handleWriteFailure( error, context ) )
         {
-            throw new RuntimeException( error );
+            queryFinished.completeExceptionally( error );
+        }
+        else
+        {
+            queryFinished.complete( null );
         }
     }
 }
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxWriteQueryWithRetries.java b/driver/src/test/java/org/neo4j/driver/stress/RxWriteQueryWithRetries.java
@@ -50,22 +50,23 @@ public CompletionStage<Void> execute( C context )
         Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.WRITE, context ) ),
                 session -> session.writeTransaction( tx -> tx.run( "CREATE ()" ).consume() ), RxSession::close )
                 .subscribe( summary -> {
-                    queryFinished.complete( null );
                     assertEquals( 1, summary.counters().nodesCreated() );
                     context.nodeCreated();
-                }, error -> {
                     queryFinished.complete( null );
-                    handleError( Futures.completionExceptionCause( error ), context );
-                } );
+                }, error -> handleError( Futures.completionExceptionCause( error ), context, queryFinished ) );
 
         return queryFinished;
     }
 
-    private void handleError( Throwable error, C context )
+    private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )
     {
         if ( !stressTest.handleWriteFailure( error, context ) )
         {
-            throw new RuntimeException( error );
+            queryFinished.completeExceptionally( error );
+        }
+        else
+        {
+            queryFinished.complete( null );
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -55,6 +55,7 @@ public CompletionStage<Void> execute( C context )`
`55`	`55`	`}`
`56`	`56`	`else`
`57`	`57`	`{`
	`58`	`+ context.setBookmark( session.lastBookmark() );`
`58`	`59`	`assertEquals( 1, summary.counters().nodesCreated() );`
`59`	`60`	`context.nodeCreated();`
`60`	`61`	`}`
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@`
`21`	`21`	`import org.neo4j.driver.AccessMode;`
`22`	`22`	`import org.neo4j.driver.Driver;`
`23`	`23`	`import org.neo4j.driver.Session;`
`24`		`-import org.neo4j.driver.Result;`
	`24`	`+import org.neo4j.driver.summary.ResultSummary;`
`25`	`25`
`26`	`26`	`import static org.junit.jupiter.api.Assertions.assertEquals;`
`27`	`27`
`@@ -38,12 +38,13 @@ public BlockingWriteQuery( AbstractStressTestBase<C> stressTest, Driver driver,`
`38`	`38`	`@Override`
`39`	`39`	`public void execute( C context )`
`40`	`40`	`{`
`41`		`- Result result = null;`
	`41`	`+ ResultSummary summary = null;`
`42`	`42`	`Throwable queryError = null;`
`43`	`43`
`44`	`44`	`try ( Session session = newSession( AccessMode.WRITE, context ) )`
`45`	`45`	`{`
`46`		`- result = session.run( "CREATE ()" );`
	`46`	`+ summary = session.run( "CREATE ()" ).consume();`
	`47`	`+ context.setBookmark( session.lastBookmark() );`
`47`	`48`	`}`
`48`	`49`	`catch ( Throwable error )`
`49`	`50`	`{`
`@@ -54,9 +55,9 @@ public void execute( C context )`
`54`	`55`	`}`
`55`	`56`	`}`
`56`	`57`
`57`		`- if ( queryError == null && result != null )`
	`58`	`+ if ( queryError == null && summary != null )`
`58`	`59`	`{`
`59`		`- assertEquals( 1, result.consume().counters().nodesCreated() );`
	`60`	`+ assertEquals( 1, summary.counters().nodesCreated() );`
`60`	`61`	`context.nodeCreated();`
`61`	`62`	`}`
`62`	`63`	`}`
Original file line number	Diff line number	Diff line change
`@@ -47,24 +47,27 @@ public CompletionStage<Void> execute( C context )`
`47`	`47`	`{`
`48`	`48`	`CompletableFuture<Void> queryFinished = new CompletableFuture<>();`
`49`	`49`	`Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.WRITE, context ) ),`
`50`		`- session -> session.run( "CREATE ()" ).consume(), RxSession::close )`
	`50`	`+ session -> Flux.from( session.run( "CREATE ()" ).consume() )`
	`51`	`+ .doOnComplete( () -> context.setBookmark( session.lastBookmark() ) ),`
	`52`	`+ RxSession::close )`
`51`	`53`	`.subscribe( summary -> {`
`52`		`- queryFinished.complete( null );`
`53`	`54`	`assertEquals( 1, summary.counters().nodesCreated() );`
`54`	`55`	`context.nodeCreated();`
`55`		`- }, error -> {`
`56`	`56`	`queryFinished.complete( null );`
`57`		`- handleError( Futures.completionExceptionCause( error ), context );`
`58`		`- } );`
	`57`	`+ }, error -> handleError( Futures.completionExceptionCause( error ), context, queryFinished ) );`
`59`	`58`
`60`	`59`	`return queryFinished;`
`61`	`60`	`}`
`62`	`61`
`63`		`- private void handleError( Throwable error, C context )`
	`62`	`+ private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )`
`64`	`63`	`{`
`65`	`64`	`if ( !stressTest.handleWriteFailure( error, context ) )`
`66`	`65`	`{`
`67`		`- throw new RuntimeException( error );`
	`66`	`+ queryFinished.completeExceptionally( error );`
	`67`	`+ }`
	`68`	`+ else`
	`69`	`+ {`
	`70`	`+ queryFinished.complete( null );`
`68`	`71`	`}`
`69`	`72`	`}`
`70`	`73`	`}`
Original file line number	Diff line number	Diff line change
`@@ -49,22 +49,26 @@ public CompletionStage<Void> execute( C context )`
`49`	`49`	`Flux.usingWhen( session.beginTransaction(), tx -> tx.run( "CREATE ()" ).consume(),`
`50`	`50`	`RxTransaction::commit, ( tx, error ) -> tx.rollback(), null ).subscribe(`
`51`	`51`	`summary -> {`
	`52`	`+ context.setBookmark( session.lastBookmark() );`
`52`	`53`	`assertEquals( 1, summary.counters().nodesCreated() );`
`53`	`54`	`context.nodeCreated();`
`54`	`55`	`queryFinished.complete( null );`
`55`	`56`	`}, error -> {`
`56`		`- handleError( Futures.completionExceptionCause( error ), context );`
`57`		`- queryFinished.complete( null );`
	`57`	`+ handleError( Futures.completionExceptionCause( error ), context, queryFinished );`
`58`	`58`	`} );`
`59`	`59`
`60`	`60`	`return queryFinished;`
`61`	`61`	`}`
`62`	`62`
`63`		`- private void handleError( Throwable error, C context )`
	`63`	`+ private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )`
`64`	`64`	`{`
`65`	`65`	`if ( !stressTest.handleWriteFailure( error, context ) )`
`66`	`66`	`{`
`67`		`- throw new RuntimeException( error );`
	`67`	`+ queryFinished.completeExceptionally( error );`
	`68`	`+ }`
	`69`	`+ else`
	`70`	`+ {`
	`71`	`+ queryFinished.complete( null );`
`68`	`72`	`}`
`69`	`73`	`}`
`70`	`74`	`}`
Original file line number	Diff line number	Diff line change
`@@ -50,22 +50,23 @@ public CompletionStage<Void> execute( C context )`
`50`	`50`	`Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.WRITE, context ) ),`
`51`	`51`	`session -> session.writeTransaction( tx -> tx.run( "CREATE ()" ).consume() ), RxSession::close )`
`52`	`52`	`.subscribe( summary -> {`
`53`		`- queryFinished.complete( null );`
`54`	`53`	`assertEquals( 1, summary.counters().nodesCreated() );`
`55`	`54`	`context.nodeCreated();`
`56`		`- }, error -> {`
`57`	`55`	`queryFinished.complete( null );`
`58`		`- handleError( Futures.completionExceptionCause( error ), context );`
`59`		`- } );`
	`56`	`+ }, error -> handleError( Futures.completionExceptionCause( error ), context, queryFinished ) );`
`60`	`57`
`61`	`58`	`return queryFinished;`
`62`	`59`	`}`
`63`	`60`
`64`		`- private void handleError( Throwable error, C context )`
	`61`	`+ private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )`
`65`	`62`	`{`
`66`	`63`	`if ( !stressTest.handleWriteFailure( error, context ) )`
`67`	`64`	`{`
`68`		`- throw new RuntimeException( error );`
	`65`	`+ queryFinished.completeExceptionally( error );`
	`66`	`+ }`
	`67`	`+ else`
	`68`	`+ {`
	`69`	`+ queryFinished.complete( null );`
`69`	`70`	`}`
`70`	`71`	`}`
`71`	`72`	`}`