UnmanagedTransaction { public CompletionStage<UnmanagedTransaction> beginAsync(Bookmark initialBookmark, TransactionConfig config ) { return protocol.beginTransaction( connection, initialBookmark, config ) .handle( ( ignore, beginError ) -> { if ( beginError != null ) { connection.release(); throw Futures.asCompletionException( beginError ); } return this; } ); } UnmanagedTransaction(Connection connection, BookmarkHolder bookmarkHolder, long fetchSize ); CompletionStage<UnmanagedTransaction> beginAsync(Bookmark initialBookmark, TransactionConfig config ); CompletionStage<Void> closeAsync(); CompletionStage<Void> commitAsync(); CompletionStage<Void> rollbackAsync(); CompletionStage<ResultCursor> runAsync(Query query, boolean waitForRunResponse ); CompletionStage<RxResultCursor> runRx(Query query); boolean isOpen(); void markTerminated( Throwable cause ); Connection connection();  }
@Test void shouldReleaseConnectionWhenBeginFails() { RuntimeException error = new RuntimeException( "Wrong bookmark!" ); Connection connection = connectionWithBegin( handler -> handler.onFailure( error ) ); UnmanagedTransaction tx = new UnmanagedTransaction( connection, new DefaultBookmarkHolder(), UNLIMITED_FETCH_SIZE ); Bookmark bookmark = InternalBookmark.parse( "SomeBookmark" ); TransactionConfig txConfig = TransactionConfig.empty(); RuntimeException e = assertThrows( RuntimeException.class, () -> await( tx.beginAsync( bookmark, txConfig ) ) ); assertEquals( error, e ); verify( connection ).release(); }
@Test void shouldNotReleaseConnectionWhenBeginSucceeds() { Connection connection = connectionWithBegin( handler -> handler.onSuccess( emptyMap() ) ); UnmanagedTransaction tx = new UnmanagedTransaction( connection, new DefaultBookmarkHolder(), UNLIMITED_FETCH_SIZE ); Bookmark bookmark = InternalBookmark.parse( "SomeBookmark" ); TransactionConfig txConfig = TransactionConfig.empty(); await( tx.beginAsync( bookmark, txConfig ) ); verify( connection, never() ).release(); }