Skip to content

Commit

Permalink
Updating stress tests to improve stability when testing clustered setup
Browse files Browse the repository at this point in the history
  • Loading branch information
injectives committed Jan 22, 2021
1 parent d44eadc commit 57eb7ea
Show file tree
Hide file tree
Showing 13 changed files with 611 additions and 130 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 );
Expand Down Expand Up @@ -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 );

Expand All @@ -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() ) );

Expand Down Expand Up @@ -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 )
Expand Down Expand Up @@ -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;
}
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
} ));
}
}
Original file line number Diff line number Diff line change
@@ -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();
}
}
Original file line number Diff line number Diff line change
@@ -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 );
}
}
}
Loading

0 comments on commit 57eb7ea

Please sign in to comment.