Updating stress tests to improve stability when testing clustered setup

Author: injectives

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

@@ -31,7 +31,7 @@
 import java.lang.reflect.Method;
 import java.net.URI;
 import java.util.ArrayList;
-import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -159,7 +159,7 @@ void asyncApiBigDataTest() throws Throwable
     }
 
     @Test
-    void rxApiBigDataTest() throws Throwable
+    void rxApiBigDataTest()
     {
         assertRxIsAvailable();
         Bookmark bookmark = createNodesRx( bigDataTestBatchCount(), BIG_DATA_TEST_BATCH_SIZE, driver );
@@ -221,7 +221,17 @@ Config config()
 
     abstract C createContext();
 
-    abstract List<BlockingCommand<C>> createTestSpecificBlockingCommands();
+    List<BlockingCommand<C>> createTestSpecificBlockingCommands() {
+        return Collections.emptyList();
+    }
+
+    List<AsyncCommand<C>> createTestSpecificAsyncCommands() {
+        return Collections.emptyList();
+    }
+
+    List<RxCommand<C>> createTestSpecificRxCommands() {
+        return Collections.emptyList();
+    }
 
     abstract boolean handleWriteFailure( Throwable error, C context );
 
@@ -245,23 +255,15 @@ private List<BlockingCommand<C>> createBlockingCommands()
     {
         List<BlockingCommand<C>> commands = new ArrayList<>();
 
-        commands.add( new BlockingReadQuery<>( driver, false ) );
-        commands.add( new BlockingReadQuery<>( driver, true ) );
-
-        commands.add( new BlockingReadQueryInTx<>( driver, false ) );
-        commands.add( new BlockingReadQueryInTx<>( driver, true ) );
+        commands.add( new BlockingReadQueryWithRetries<>( driver, false ) );
+        commands.add( new BlockingReadQueryWithRetries<>( driver, true ) );
 
-        commands.add( new BlockingWriteQuery<>( this, driver, false ) );
-        commands.add( new BlockingWriteQuery<>( this, driver, true ) );
+        commands.add( new BlockingWriteQueryWithRetries<>( this, driver, false ) );
+        commands.add( new BlockingWriteQueryWithRetries<>( this, driver, true ) );
 
-        commands.add( new BlockingWriteQueryInTx<>( this, driver, false ) );
-        commands.add( new BlockingWriteQueryInTx<>( this, driver, true ) );
+        commands.add( new BlockingWrongQueryWithRetries<>( driver ) );
 
-        commands.add( new BlockingWrongQuery<>( driver ) );
-        commands.add( new BlockingWrongQueryInTx<>( driver ) );
-
-        commands.add( new BlockingFailingQuery<>( driver ) );
-        commands.add( new BlockingFailingQueryInTx<>( driver ) );
+        commands.add( new BlockingFailingQueryWithRetries<>( driver ) );
 
         commands.add( new FailedAuth<>( databaseUri(), config() ) );
 
@@ -299,29 +301,19 @@ private List<Future<?>> launchRxWorkerThreads( C context )
 
     private List<RxCommand<C>> createRxCommands()
     {
-        return Arrays.asList(
-                new RxReadQuery<>( driver, false ),
-                new RxReadQuery<>( driver, true ),
-
-                new RxWriteQuery<>( this, driver, false ),
-                new RxWriteQuery<>( this, driver, true ),
+        List<RxCommand<C>> commands = new ArrayList<>();
 
-                new RxReadQueryInTx<>( driver, false ),
-                new RxReadQueryInTx<>( driver, true ),
+        commands.add( new RxReadQueryWithRetries<>( driver, false ) );
+        commands.add( new RxReadQueryWithRetries<>( driver, true ) );
 
-                new RxWriteQueryInTx<>( this, driver, false ),
-                new RxWriteQueryInTx<>( this, driver, true ),
+        commands.add( new RxWriteQueryWithRetries<>( this, driver, false ) );
+        commands.add( new RxWriteQueryWithRetries<>( this, driver, true ) );
 
-                new RxReadQueryWithRetries<>( driver, false ),
-                new RxReadQueryWithRetries<>( driver, false ),
+        commands.add( new RxFailingQueryWithRetries<>( driver ) );
 
-                new RxWriteQueryWithRetries<>( this, driver, false ),
-                new RxWriteQueryWithRetries<>( this, driver, true ),
+        commands.addAll( createTestSpecificRxCommands() );
 
-                new RxFailingQuery<>( driver ),
-                new RxFailingQueryInTx<>( driver ),
-                new RxFailingQueryWithRetries<>( driver )
-        );
+        return commands;
     }
 
     private Future<Void> launchRxWorkerThread( ExecutorService executor, List<RxCommand<C>> commands, C context )
@@ -367,23 +359,15 @@ private List<AsyncCommand<C>> createAsyncCommands()
     {
         List<AsyncCommand<C>> commands = new ArrayList<>();
 
-        commands.add( new AsyncReadQuery<>( driver, false ) );
-        commands.add( new AsyncReadQuery<>( driver, true ) );
-
-        commands.add( new AsyncReadQueryInTx<>( driver, false ) );
-        commands.add( new AsyncReadQueryInTx<>( driver, true ) );
-
-        commands.add( new AsyncWriteQuery<>( this, driver, false ) );
-        commands.add( new AsyncWriteQuery<>( this, driver, true ) );
+        commands.add( new AsyncReadQueryWithRetries<>( driver, false ) );
+        commands.add( new AsyncReadQueryWithRetries<>( driver, true ) );
 
-        commands.add( new AsyncWriteQueryInTx<>( this, driver, false ) );
-        commands.add( new AsyncWriteQueryInTx<>( this, driver, true ) );
+        commands.add( new AsyncWriteQueryWithRetries<>( this, driver, false ) );
+        commands.add( new AsyncWriteQueryWithRetries<>( this, driver, true ) );
 
-        commands.add( new AsyncWrongQuery<>( driver ) );
-        commands.add( new AsyncWrongQueryInTx<>( driver ) );
+        commands.add( new AsyncWrongQueryWithRetries<>( driver ) );
 
-        commands.add( new AsyncFailingQuery<>( driver ) );
-        commands.add( new AsyncFailingQueryInTx<>( driver ) );
+        commands.add( new AsyncFailingQueryWithRetries<>( driver ) );
 
         return commands;
     }

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

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2002-2020 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.stress;
+
+import java.util.concurrent.CompletionStage;
+
+import org.neo4j.driver.AccessMode;
+import org.neo4j.driver.Driver;
+import org.neo4j.driver.async.AsyncSession;
+import org.neo4j.driver.async.ResultCursor;
+import org.neo4j.driver.internal.util.Futures;
+
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.junit.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.neo4j.driver.internal.util.Matchers.arithmeticError;
+
+public class AsyncFailingQueryWithRetries<C extends AbstractContext> extends AbstractAsyncQuery<C>
+{
+    public AsyncFailingQueryWithRetries( Driver driver )
+    {
+        super( driver, false );
+    }
+
+    @Override
+    public CompletionStage<Void> execute( C context )
+    {
+        AsyncSession session = newSession( AccessMode.READ, context );
+
+        return session.readTransactionAsync( tx -> tx.runAsync( "UNWIND [10, 5, 0] AS x RETURN 10 / x" )
+                      .thenCompose( ResultCursor::listAsync )
+                      .handle( ( records, error ) ->
+                               {
+                                   session.closeAsync();
+
+                                   assertNull( records );
+                                   Throwable cause = Futures.completionExceptionCause( error );
+                                   assertThat( cause, is( arithmeticError() ) );
+
+                                   return null;
+                               } ));
+    }
+}

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

@@ -0,0 +1,72 @@
+/*
+ * Copyright (c) 2002-2020 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.stress;
+
+import java.util.concurrent.CompletionStage;
+
+import org.neo4j.driver.AccessMode;
+import org.neo4j.driver.Driver;
+import org.neo4j.driver.Record;
+import org.neo4j.driver.async.AsyncSession;
+import org.neo4j.driver.async.ResultCursor;
+import org.neo4j.driver.summary.ResultSummary;
+import org.neo4j.driver.types.Node;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class AsyncReadQueryWithRetries<C extends AbstractContext> extends AbstractAsyncQuery<C>
+{
+    public AsyncReadQueryWithRetries( Driver driver, boolean useBookmark )
+    {
+        super( driver, useBookmark );
+    }
+
+    @Override
+    public CompletionStage<Void> execute( C context )
+    {
+        AsyncSession session = newSession( AccessMode.READ, context );
+
+        CompletionStage<ResultSummary> queryFinished = session.readTransactionAsync(
+                tx -> tx.runAsync( "MATCH (n) RETURN n LIMIT 1" )
+                        .thenCompose(
+                                cursor -> cursor.nextAsync()
+                                                .thenCompose( record -> processAndGetSummary( record, cursor ) ) ) );
+
+        queryFinished.whenComplete( ( summary, error ) ->
+                                    {
+                                        if ( summary != null )
+                                        {
+                                            context.readCompleted( summary );
+                                        }
+                                        session.closeAsync();
+                                    } );
+
+        return queryFinished.thenApply( summary -> null );
+    }
+
+    private CompletionStage<ResultSummary> processAndGetSummary( Record record, ResultCursor cursor )
+    {
+        if ( record != null )
+        {
+            Node node = record.get( 0 ).asNode();
+            assertNotNull( node );
+        }
+        return cursor.consumeAsync();
+    }
+}

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

@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) 2002-2020 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.stress;
+
+import java.util.concurrent.CompletionStage;
+
+import org.neo4j.driver.AccessMode;
+import org.neo4j.driver.Driver;
+import org.neo4j.driver.async.AsyncSession;
+import org.neo4j.driver.async.ResultCursor;
+import org.neo4j.driver.internal.util.Futures;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class AsyncWriteQueryWithRetries<C extends AbstractContext> extends AbstractAsyncQuery<C>
+{
+    private final AbstractStressTestBase<C> stressTest;
+
+    public AsyncWriteQueryWithRetries( AbstractStressTestBase<C> stressTest, Driver driver, boolean useBookmark )
+    {
+        super( driver, useBookmark );
+        this.stressTest = stressTest;
+    }
+
+    @Override
+    public CompletionStage<Void> execute( C context )
+    {
+        AsyncSession session = newSession( AccessMode.WRITE, context );
+
+        return session.writeTransactionAsync(
+                tx -> tx.runAsync( "CREATE ()" )
+                        .thenCompose( ResultCursor::consumeAsync ) )
+                      .handle( ( summary, error ) ->
+                               {
+                                   session.closeAsync();
+
+                                   if ( error != null )
+                                   {
+                                       handleError( Futures.completionExceptionCause( error ), context );
+                                   }
+                                   else
+                                   {
+                                       context.setBookmark( session.lastBookmark() );
+                                       assertEquals( 1, summary.counters().nodesCreated() );
+                                       context.nodeCreated();
+                                   }
+
+                                   return null;
+                               } );
+    }
+
+    private void handleError( Throwable error, C context )
+    {
+        if ( !stressTest.handleWriteFailure( error, context ) )
+        {
+            throw new RuntimeException( error );
+        }
+    }
+}