Do not keep bookmark in transaction

Pass it directly to the session, when the transaction is committed.
This simplifies `ExplicitTransaction` a bit by removing all bookmark
handling from it.

Author: lutovich

driver/src/main/java/org/neo4j/driver/internal/ExplicitTransaction.java

@@ -68,7 +68,6 @@ private enum State
     private final NetworkSession session;
     private final ResultCursorsHolder resultCursors;
 
-    private volatile Bookmarks bookmarks = Bookmarks.empty();
     private volatile State state = State.ACTIVE;
 
     public ExplicitTransaction( Connection connection, NetworkSession session )
@@ -229,19 +228,6 @@ public void markTerminated()
         state = State.TERMINATED;
     }
 
-    public Bookmarks bookmark()
-    {
-        return bookmarks;
-    }
-
-    public void setBookmarks( Bookmarks bookmarks )
-    {
-        if ( bookmarks != null && !bookmarks.isEmpty() )
-        {
-            this.bookmarks = bookmarks;
-        }
-    }
-
     private CompletionStage<Void> doCommitAsync()
     {
         if ( state == State.TERMINATED )
@@ -252,7 +238,7 @@ private CompletionStage<Void> doCommitAsync()
         return protocol.commitTransaction( connection )
                 .thenApply( newBookmarks ->
                 {
-                    setBookmarks( newBookmarks );
+                    session.setBookmarks( newBookmarks );
                     return null;
                 } );
     }
@@ -283,7 +269,6 @@ private void transactionClosed( State newState )
     {
         state = newState;
         connection.release(); // release in background
-        session.setBookmarks( bookmarks );
     }
 
     private void terminateConnectionOnThreadInterrupt( String reason )

driver/src/test/java/org/neo4j/driver/internal/ExplicitTransactionTest.java

@@ -201,40 +201,6 @@ void shouldBeClosedWhenMarkedTerminatedAndClosed()
         assertFalse( tx.isOpen() );
     }
 
-    @Test
-    void shouldHaveEmptyBookmarkInitially()
-    {
-        ExplicitTransaction tx = beginTx( connectionMock() );
-        assertTrue( tx.bookmark().isEmpty() );
-    }
-
-    @Test
-    void shouldNotKeepInitialBookmark()
-    {
-        ExplicitTransaction tx = beginTx( connectionMock(), Bookmarks.from( "Dog" ) );
-        assertTrue( tx.bookmark().isEmpty() );
-    }
-
-    @Test
-    void shouldNotOverwriteBookmarkWithNull()
-    {
-        ExplicitTransaction tx = beginTx( connectionMock() );
-        tx.setBookmarks( Bookmarks.from( "Cat" ) );
-        assertEquals( "Cat", tx.bookmark().maxBookmarkAsString() );
-        tx.setBookmarks( null );
-        assertEquals( "Cat", tx.bookmark().maxBookmarkAsString() );
-    }
-
-    @Test
-    void shouldNotOverwriteBookmarkWithEmptyBookmark()
-    {
-        ExplicitTransaction tx = beginTx( connectionMock() );
-        tx.setBookmarks( Bookmarks.from( "Cat" ) );
-        assertEquals( "Cat", tx.bookmark().maxBookmarkAsString() );
-        tx.setBookmarks( Bookmarks.empty() );
-        assertEquals( "Cat", tx.bookmark().maxBookmarkAsString() );
-    }
-
     @Test
     void shouldReleaseConnectionWhenBeginFails()
     {

driver/src/test/java/org/neo4j/driver/internal/NetworkSessionTest.java

@@ -27,6 +27,7 @@
 
 import java.util.Map;
 
+import org.neo4j.driver.internal.messaging.BoltProtocol;
 import org.neo4j.driver.internal.messaging.request.PullAllMessage;
 import org.neo4j.driver.internal.messaging.request.RunMessage;
 import org.neo4j.driver.internal.retry.FixedRetryLogic;
@@ -62,6 +63,7 @@
 import static org.mockito.Mockito.RETURNS_MOCKS;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -243,11 +245,17 @@ void acquiresNewConnectionForBeginTx()
     @Test
     void updatesBookmarkWhenTxIsClosed()
     {
-        Transaction tx = session.beginTransaction();
-        setBookmarks( tx, Bookmarks.from( "TheBookmark" ) );
+        Bookmarks bookmarkAfterCommit = Bookmarks.from( "TheBookmark" );
+
+        BoltProtocol protocol = spy( DEFAULT_TEST_PROTOCOL );
+        doReturn( completedFuture( bookmarkAfterCommit ) ).when( protocol ).commitTransaction( any( Connection.class ) );
 
+        when( connection.protocol() ).thenReturn( protocol );
+
+        Transaction tx = session.beginTransaction();
         assertNull( session.lastBookmark() );
 
+        tx.success();
         tx.close();
         assertEquals( "TheBookmark", session.lastBookmark() );
     }
@@ -309,18 +317,21 @@ void bookmarkIsPropagatedBetweenTransactions()
 
         NetworkSession session = newSession( connectionProvider, READ );
 
+        BoltProtocol protocol = spy( DEFAULT_TEST_PROTOCOL );
+        doReturn( completedFuture( bookmarks1 ), completedFuture( bookmarks2 ) ).when( protocol ).commitTransaction( any( Connection.class ) );
+
+        when( connection.protocol() ).thenReturn( protocol );
+
         try ( Transaction tx = session.beginTransaction() )
         {
-            setBookmarks( tx, bookmarks1 );
+            tx.success();
         }
-
         assertEquals( bookmarks1, Bookmarks.from( session.lastBookmark() ) );
 
         try ( Transaction tx = session.beginTransaction() )
         {
             verifyBeginTx( connection, bookmarks1 );
-            assertTrue( getBookmarks( tx ).isEmpty() );
-            setBookmarks( tx, bookmarks2 );
+            tx.success();
         }
         assertEquals( bookmarks2, Bookmarks.from( session.lastBookmark() ) );
     }
@@ -898,16 +909,6 @@ private static void verifyRunAndFlush( Connection connectionMock, String stateme
         verify( connectionMock, mode ).writeAndFlush( eq( new RunMessage( statement ) ), any(), eq( PullAllMessage.PULL_ALL ), any() );
     }
 
-    private static Bookmarks getBookmarks( Transaction tx )
-    {
-        return ((ExplicitTransaction) tx).bookmark();
-    }
-
-    private static void setBookmarks( Transaction tx, Bookmarks bookmarks )
-    {
-        ((ExplicitTransaction) tx).setBookmarks( bookmarks );
-    }
-
     private static void setupFailingCommit( Connection connection, int times )
     {
         doAnswer( new Answer<Void>()