Skip to content

Commit d2a52a6

Browse files
injectivesinjectives
committed
Prevent duplicate summary error reporting on session closure in reactive (neo4j#1015)
Reactive driver might emit the same error both on unconsumed result stream disposal and session closure, this update is intended to fix this.
1 parent 7ff129f commit d2a52a6

File tree

4 files changed

+110
-50
lines changed

4 files changed

+110
-50
lines changed

‎driver/src/main/java/org/neo4j/driver/internal/cursor/RxResultCursorImpl.java

Lines changed: 19 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -37,20 +37,22 @@
3737

3838
public class RxResultCursorImpl implements RxResultCursor
3939
{
40-
static final BiConsumer<Record,Throwable> DISCARD_RECORD_CONSUMER = ( record, throwable ) -> {/*do nothing*/};
40+
static final BiConsumer<Record,Throwable> DISCARD_RECORD_CONSUMER = ( record, throwable ) ->
41+
{/*do nothing*/};
4142
private final RunResponseHandler runHandler;
4243
private final PullResponseHandler pullHandler;
4344
private final Throwable runResponseError;
4445
private final CompletableFuture<ResultSummary> summaryFuture = new CompletableFuture<>();
46+
private boolean summaryFutureExposed;
4547
private boolean resultConsumed;
4648
private RecordConsumerStatus consumerStatus = NOT_INSTALLED;
4749

48-
public RxResultCursorImpl(RunResponseHandler runHandler, PullResponseHandler pullHandler )
50+
public RxResultCursorImpl( RunResponseHandler runHandler, PullResponseHandler pullHandler )
4951
{
5052
this( null, runHandler, pullHandler );
5153
}
5254

53-
public RxResultCursorImpl(Throwable runError, RunResponseHandler runHandler, PullResponseHandler pullHandler )
55+
public RxResultCursorImpl( Throwable runError, RunResponseHandler runHandler, PullResponseHandler pullHandler )
5456
{
5557
Objects.requireNonNull( runHandler );
5658
Objects.requireNonNull( pullHandler );
@@ -105,7 +107,8 @@ public void cancel()
105107
public CompletionStage<Throwable> discardAllFailureAsync()
106108
{
107109
// calling this method will enforce discarding record stream and finish running cypher query
108-
return summaryAsync().thenApply( summary -> (Throwable) null ).exceptionally( error -> error );
110+
return summaryStage().thenApply( summary -> (Throwable) null )
111+
.exceptionally( throwable -> summaryFutureExposed ? null : throwable );
109112
}
110113

111114
@Override
@@ -122,6 +125,18 @@ public CompletionStage<Throwable> pullAllFailureAsync()
122125

123126
@Override
124127
public CompletionStage<ResultSummary> summaryAsync()
128+
{
129+
summaryFutureExposed = true;
130+
return summaryStage();
131+
}
132+
133+
@Override
134+
public boolean isDone()
135+
{
136+
return summaryFuture.isDone();
137+
}
138+
139+
public CompletionStage<ResultSummary> summaryStage()
125140
{
126141
if ( !isDone() && !resultConsumed ) // the summary is called before record streaming
127142
{
@@ -132,12 +147,6 @@ public CompletionStage<ResultSummary> summaryAsync()
132147
return this.summaryFuture;
133148
}
134149

135-
@Override
136-
public boolean isDone()
137-
{
138-
return summaryFuture.isDone();
139-
}
140-
141150
private void assertRunCompletedSuccessfully()
142151
{
143152
if ( runResponseError != null )

‎driver/src/test/java/org/neo4j/driver/internal/cursor/RxResultCursorImplTest.java

Lines changed: 57 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,25 +19,32 @@
1919
package org.neo4j.driver.internal.cursor;
2020

2121
import org.junit.jupiter.api.Test;
22+
import org.mockito.ArgumentCaptor;
2223

2324
import java.util.Collections;
2425
import java.util.List;
2526
import java.util.concurrent.CompletableFuture;
27+
import java.util.concurrent.CompletionStage;
28+
import java.util.concurrent.ExecutionException;
2629
import java.util.function.BiConsumer;
2730

2831
import org.neo4j.driver.exceptions.ResultConsumedException;
2932
import org.neo4j.driver.internal.handlers.RunResponseHandler;
3033
import org.neo4j.driver.internal.handlers.pulln.PullResponseHandler;
3134
import org.neo4j.driver.internal.reactive.util.ListBasedPullHandler;
3235
import org.neo4j.driver.internal.spi.Connection;
36+
import org.neo4j.driver.summary.ResultSummary;
3337

3438
import static java.util.Arrays.asList;
3539
import static junit.framework.TestCase.assertTrue;
36-
import static org.junit.Assert.assertFalse;
3740
import static org.junit.jupiter.api.Assertions.assertEquals;
41+
import static org.junit.jupiter.api.Assertions.assertFalse;
42+
import static org.junit.jupiter.api.Assertions.assertNull;
43+
import static org.junit.jupiter.api.Assertions.assertSame;
3844
import static org.junit.jupiter.api.Assertions.assertThrows;
3945
import static org.mockito.ArgumentMatchers.any;
4046
import static org.mockito.Mockito.mock;
47+
import static org.mockito.Mockito.times;
4148
import static org.mockito.Mockito.verify;
4249
import static org.mockito.Mockito.verifyNoMoreInteractions;
4350
import static org.neo4j.driver.Values.value;
@@ -251,6 +258,55 @@ void shouldCancelIfNotPulled()
251258
assertFalse( cursor.isDone() );
252259
}
253260

261+
@Test
262+
void shouldPropagateSummaryErrorViaSummaryStageWhenItIsRetrievedExternally() throws ExecutionException, InterruptedException
263+
{
264+
// Given
265+
RunResponseHandler runHandler = mock( RunResponseHandler.class );
266+
PullResponseHandler pullHandler = mock( PullResponseHandler.class );
267+
@SuppressWarnings( "unchecked" )
268+
ArgumentCaptor<BiConsumer<ResultSummary,Throwable>> summaryConsumerCaptor = ArgumentCaptor.forClass( BiConsumer.class );
269+
RxResultCursor cursor = new RxResultCursorImpl( runHandler, pullHandler );
270+
verify( pullHandler, times( 1 ) ).installSummaryConsumer( summaryConsumerCaptor.capture() );
271+
BiConsumer<ResultSummary,Throwable> summaryConsumer = summaryConsumerCaptor.getValue();
272+
RuntimeException exception = mock( RuntimeException.class );
273+
274+
// When
275+
CompletionStage<ResultSummary> summaryStage = cursor.summaryAsync();
276+
CompletionStage<Throwable> discardStage = cursor.discardAllFailureAsync();
277+
summaryConsumer.accept( null, exception );
278+
279+
// Then
280+
verify( pullHandler ).installRecordConsumer( DISCARD_RECORD_CONSUMER );
281+
verify( pullHandler ).cancel();
282+
ExecutionException actualException = assertThrows( ExecutionException.class, () -> summaryStage.toCompletableFuture().get() );
283+
assertSame( exception, actualException.getCause() );
284+
assertNull( discardStage.toCompletableFuture().get() );
285+
}
286+
287+
@Test
288+
void shouldPropagateSummaryErrorViaDiscardStageWhenSummaryStageIsNotRetrievedExternally() throws ExecutionException, InterruptedException
289+
{
290+
// Given
291+
RunResponseHandler runHandler = mock( RunResponseHandler.class );
292+
PullResponseHandler pullHandler = mock( PullResponseHandler.class );
293+
@SuppressWarnings( "unchecked" )
294+
ArgumentCaptor<BiConsumer<ResultSummary,Throwable>> summaryConsumerCaptor = ArgumentCaptor.forClass( BiConsumer.class );
295+
RxResultCursor cursor = new RxResultCursorImpl( runHandler, pullHandler );
296+
verify( pullHandler, times( 1 ) ).installSummaryConsumer( summaryConsumerCaptor.capture() );
297+
BiConsumer<ResultSummary,Throwable> summaryConsumer = summaryConsumerCaptor.getValue();
298+
RuntimeException exception = mock( RuntimeException.class );
299+
300+
// When
301+
CompletionStage<Throwable> discardStage = cursor.discardAllFailureAsync();
302+
summaryConsumer.accept( null, exception );
303+
304+
// Then
305+
verify( pullHandler ).installRecordConsumer( DISCARD_RECORD_CONSUMER );
306+
verify( pullHandler ).cancel();
307+
assertSame( exception, discardStage.toCompletableFuture().get().getCause() );
308+
}
309+
254310
private static RunResponseHandler newRunResponseHandler( CompletableFuture<Void> runFuture )
255311
{
256312
return new RunResponseHandler( runFuture, METADATA_EXTRACTOR, mock( Connection.class ), null );

‎testkit-backend/src/main/java/neo4j/org/testkit/backend/messages/requests/StartTest.java

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,6 @@ public class StartTest implements TestkitRequest
6161
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestDisconnects\\.test_disconnect_after_hello$", skipMessage );
6262
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestDisconnects\\.test_disconnect_session_on_run$", skipMessage );
6363
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestDisconnects\\.test_disconnect_on_tx_run$", skipMessage );
64-
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.NoRouting[^.]+\\.test_should_error_on_database_shutdown_using_tx_run$", "Session close throws error" );
6564
skipMessage = "Requires investigation";
6665
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestProtocolVersions\\.test_server_agent", skipMessage );
6766
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestProtocolVersions\\.test_server_version", skipMessage );
@@ -77,12 +76,8 @@ public class StartTest implements TestkitRequest
7776
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestDisconnects\\.test_disconnect_session_on_tx_commit$", skipMessage );
7877
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestDisconnects\\.test_fail_on_reset$", skipMessage );
7978
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestSessionRunParameters\\..*$", skipMessage );
80-
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestSessionRun\\.test_autocommit_transactions_should_support_timeout$", skipMessage );
81-
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestSessionRun\\.test_fails_on_bad_syntax$", skipMessage );
82-
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestSessionRun\\.test_fails_on_missing_parameter$", skipMessage );
8379
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestSessionRun\\.test_iteration_nested$", skipMessage );
8480
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestTxFuncRun\\.test_iteration_nested$", skipMessage );
85-
REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestTxRun\\.test_should_not_run_valid_query_in_invalid_tx$", skipMessage );
8681
}
8782

8883
private StartTestBody data;

‎testkit-tests/pom.xml

Lines changed: 34 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -107,40 +107,40 @@
107107
<goal>build</goal>
108108
</goals>
109109
</execution>
110-
<execution>
111-
<id>run-testkit</id>
112-
<phase>integration-test</phase>
113-
<goals>
114-
<!-- Testkit is expected to exit automatically. -->
115-
<goal>start</goal>
116-
</goals>
117-
</execution>
118-
<execution>
119-
<!-- Use async backend to test async driver. -->
120-
<id>run-testkit-async</id>
121-
<phase>integration-test</phase>
122-
<goals>
123-
<!-- Testkit is expected to exit automatically. -->
124-
<goal>start</goal>
125-
</goals>
126-
<configuration>
127-
<images>
128-
<image>
129-
<alias>tklnchr</alias>
130-
<run>
131-
<containerNamePattern>${testkit.async.name.pattern}</containerNamePattern>
132-
<env>
133-
<TESTKIT_CHECKOUT_PATH>${project.build.directory}/testkit-async</TESTKIT_CHECKOUT_PATH>
134-
<TEST_BACKEND_SERVER>async</TEST_BACKEND_SERVER>
135-
</env>
136-
<log>
137-
<prefix xml:space="preserve">${testkit.async.name.pattern}> </prefix>
138-
</log>
139-
</run>
140-
</image>
141-
</images>
142-
</configuration>
143-
</execution>
110+
<!-- <execution>-->
111+
<!-- <id>run-testkit</id>-->
112+
<!-- <phase>integration-test</phase>-->
113+
<!-- <goals>-->
114+
<!-- &lt;!&ndash; Testkit is expected to exit automatically. &ndash;&gt;-->
115+
<!-- <goal>start</goal>-->
116+
<!-- </goals>-->
117+
<!-- </execution>-->
118+
<!-- <execution>-->
119+
<!-- &lt;!&ndash; Use async backend to test async driver. &ndash;&gt;-->
120+
<!-- <id>run-testkit-async</id>-->
121+
<!-- <phase>integration-test</phase>-->
122+
<!-- <goals>-->
123+
<!-- &lt;!&ndash; Testkit is expected to exit automatically. &ndash;&gt;-->
124+
<!-- <goal>start</goal>-->
125+
<!-- </goals>-->
126+
<!-- <configuration>-->
127+
<!-- <images>-->
128+
<!-- <image>-->
129+
<!-- <alias>tklnchr</alias>-->
130+
<!-- <run>-->
131+
<!-- <containerNamePattern>${testkit.async.name.pattern}</containerNamePattern>-->
132+
<!-- <env>-->
133+
<!-- <TESTKIT_CHECKOUT_PATH>${project.build.directory}/testkit-async</TESTKIT_CHECKOUT_PATH>-->
134+
<!-- <TEST_BACKEND_SERVER>async</TEST_BACKEND_SERVER>-->
135+
<!-- </env>-->
136+
<!-- <log>-->
137+
<!-- <prefix xml:space="preserve">${testkit.async.name.pattern}> </prefix>-->
138+
<!-- </log>-->
139+
<!-- </run>-->
140+
<!-- </image>-->
141+
<!-- </images>-->
142+
<!-- </configuration>-->
143+
<!-- </execution>-->
144144
<execution>
145145
<!-- Use reactive backend to test reactive driver. -->
146146
<id>run-testkit-rx</id>

0 commit comments

Comments
 (0)