Change state behaviour in PersistentDownloader

Moved Stopped state to separate volatile variable keepRunning.

Make state variable volatile.

Remove isRunning method since it is only used in tests.

Synchronize method that move state from initial to running. This
is to make sure only one run is possible so that hasCompleted can be
trusted.

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/core/state/snapshot/PersistentSnapshotDownloader.java

@@ -38,7 +38,8 @@ class PersistentSnapshotDownloader implements Runnable
     private final CoreStateDownloader downloader;
     private final Log log;
     private final TimeoutStrategy.Timeout timeout;
-    private State state;
+    private volatile State state;
+    private volatile boolean keepRunning;
 
     PersistentSnapshotDownloader( LeaderLocator leaderLocator,
             CommandApplicationProcess applicationProcess, CoreStateDownloader downloader, Log log,
@@ -50,6 +51,7 @@ class PersistentSnapshotDownloader implements Runnable
         this.log = log;
         this.timeout = timeout;
         this.state = State.INITIATED;
+        this.keepRunning = true;
     }
 
     PersistentSnapshotDownloader( LeaderLocator leaderLocator,
@@ -63,22 +65,21 @@ private enum State
     {
         INITIATED,
         RUNNING,
-        STOPPED,
         COMPLETED
     }
 
     @Override
     public void run()
     {
-        if ( !initialStateOk() )
+        if ( !moveToRunningState() )
         {
             return;
         }
+
         try
         {
-            state = State.RUNNING;
             applicationProcess.pauseApplier( OPERATION_NAME );
-            while ( state == State.RUNNING )
+            while ( keepRunning )
             {
                 try
                 {
@@ -108,50 +109,26 @@ public void run()
         }
     }
 
-    private boolean initialStateOk()
+    private synchronized boolean moveToRunningState()
     {
-        switch ( state )
+        if ( state != State.INITIATED )
         {
-        case INITIATED:
-            return true;
-        case RUNNING:
-            log.error( "Persistent snapshot downloader is already running. " +
-                       "Illegal state '{}'. Expected '{}'", state, State.INITIATED );
-            return false;
-        case STOPPED:
-            log.info( "Persistent snapshot downloader was stopped before starting" );
-            return false;
-        case COMPLETED:
-            log.error( "Persistent snapshot downloader has already completed. " +
-                       "Illegal state '{}'. Expected '{}'", state, State.INITIATED );
-            return false;
-        default:
-            log.error( "Not a recognised state. " +
-                       "Illegal state '{}'. Expected '{}'", state, State.INITIATED );
             return false;
         }
-    }
-
-    void stop()
-    {
-        if ( state == State.RUNNING )
-        {
-            state = State.STOPPED;
-        }
-        else if ( state == State.INITIATED )
+        else
         {
-            state = State.COMPLETED;
+            state = State.RUNNING;
+            return true;
         }
     }
 
-    boolean isRunning()
+    void stop()
     {
-        return state == State.RUNNING;
+        this.keepRunning = false;
     }
 
     boolean hasCompleted()
     {
         return state == State.COMPLETED;
     }
-
 }

enterprise/causal-clustering/src/test/java/org/neo4j/causalclustering/core/state/snapshot/PersistentSnapshotDownloaderTest.java

@@ -71,11 +71,11 @@ public void shouldPauseAndResumeApplicationProcessIfDownloadIsSuccessful() throw
         verify( applicationProcess, times( 1 ) ).pauseApplier( OPERATION_NAME );
         verify( applicationProcess, times( 1 ) ).resumeApplier( OPERATION_NAME );
         verify( coreStateDownloader, times( 1 ) ).downloadSnapshot( any() );
-        assertFalse( persistentSnapshotDownloader.isRunning() );
+        assertTrue( persistentSnapshotDownloader.hasCompleted() );
     }
 
     @Test
-    public void shouldResumeCommandApplicationProcessIsInterruptedDownloadIsFailing() throws Exception
+    public void shouldResumeCommandApplicationProcessIfInterrupted() throws Exception
     {
         // given
         CoreStateDownloader coreStateDownloader = mock( CoreStateDownloader.class );
@@ -94,20 +94,16 @@ public void shouldResumeCommandApplicationProcessIsInterruptedDownloadIsFailing(
         Thread thread = new Thread( persistentSnapshotDownloader );
         thread.start();
         awaitOneIteration( timeout );
-
-        // then
-        assertTrue( persistentSnapshotDownloader.isRunning() );
-
-        // when
         thread.stop();
 
         // then
         verify( applicationProcess, times( 1 ) ).pauseApplier( OPERATION_NAME );
         verify( applicationProcess, times( 1 ) ).resumeApplier( OPERATION_NAME );
+        assertTrue( persistentSnapshotDownloader.hasCompleted() );
     }
 
     @Test
-    public void shouldResumeCommandApplicationProcessIfDownloadIsStopped() throws Exception
+    public void shouldResumeCommandApplicationProcessIfDownloaderIsStopped() throws Exception
     {
         // given
         CoreStateDownloader coreStateDownloader = mock( CoreStateDownloader.class );
@@ -126,17 +122,13 @@ public void shouldResumeCommandApplicationProcessIfDownloadIsStopped() throws Ex
         Thread thread = new Thread( persistentSnapshotDownloader );
         thread.start();
         awaitOneIteration( timeout );
-
-        // then
-        assertTrue( persistentSnapshotDownloader.isRunning() );
-
-        // when
         persistentSnapshotDownloader.stop();
         thread.join();
 
         // then
         verify( applicationProcess, times( 1 ) ).pauseApplier( OPERATION_NAME );
         verify( applicationProcess, times( 1 ) ).resumeApplier( OPERATION_NAME );
+        assertTrue( persistentSnapshotDownloader.hasCompleted() );
     }
 
     @Test
@@ -161,11 +153,11 @@ public void shouldEventuallySucceed() throws Exception
         verify( applicationProcess, times( 1 ) ).pauseApplier( OPERATION_NAME );
         verify( applicationProcess, times( 1 ) ).resumeApplier( OPERATION_NAME );
         assertEquals( 3, timeout.increments );
-        assertFalse( persistentSnapshotDownloader.isRunning() );
+        assertTrue( persistentSnapshotDownloader.hasCompleted() );
     }
 
     @Test
-    public void shouldNotStartIfAlreadyCompleted() throws Exception
+    public void shouldNotStartDownloadIfAlreadyCompleted() throws Exception
     {
         // given
         CoreStateDownloader coreStateDownloader = mock( CoreStateDownloader.class );
@@ -177,11 +169,12 @@ public void shouldNotStartIfAlreadyCompleted() throws Exception
         PersistentSnapshotDownloader persistentSnapshotDownloader =
                 new PersistentSnapshotDownloader( leaderLocator, applicationProcess, coreStateDownloader, log );
 
+        // when
         persistentSnapshotDownloader.run();
         persistentSnapshotDownloader.run();
 
-        verify( log, times( 1 ) )
-                .error( startsWith( "Persistent snapshot downloader has already completed." ), any(), any() );
+        // then
+        verify( coreStateDownloader, times(1) ).downloadSnapshot( someMember );
         verify( applicationProcess, times( 1 ) ).pauseApplier( OPERATION_NAME );
         verify( applicationProcess, times( 1 ) ).resumeApplier( OPERATION_NAME );
     }
@@ -210,35 +203,11 @@ public void shouldNotStartIfCurrentlyRunning() throws Exception
         persistentSnapshotDownloader.stop();
         thread.join();
 
-        verify( log, times( 1 ) )
-                .error( startsWith( "Persistent snapshot downloader is already running." ), any(), any() );
+        // then
         verify( applicationProcess, times( 1 ) ).pauseApplier( OPERATION_NAME );
         verify( applicationProcess, times( 1 ) ).resumeApplier( OPERATION_NAME );
     }
 
-    @Test
-    public void shouldNotStartIfStoppedBeforeRunning() throws Exception
-    {
-        // given
-        CoreStateDownloader coreStateDownloader = mock( CoreStateDownloader.class );
-        final CommandApplicationProcess applicationProcess = mock( CommandApplicationProcess.class );
-        LeaderLocator leaderLocator = mock( LeaderLocator.class );
-        doThrow( NoLeaderFoundException.class ).when( leaderLocator ).getLeader();
-
-        final Log log = mock( Log.class );
-        PersistentSnapshotDownloader persistentSnapshotDownloader =
-                new PersistentSnapshotDownloader( leaderLocator, applicationProcess, coreStateDownloader, log );
-
-        // when
-        persistentSnapshotDownloader.stop();
-        persistentSnapshotDownloader.run();
-
-        verify( log, times( 1 ) )
-                .error( startsWith( "Persistent snapshot downloader has already completed.") , any(), any() );
-        verify( applicationProcess, never() ).pauseApplier( OPERATION_NAME );
-        verify( applicationProcess, never() ).resumeApplier( OPERATION_NAME );
-    }
-
     private void awaitOneIteration( NoTimeout timeout ) throws TimeoutException
     {
         Predicates.await( () -> timeout.increments > 0, 1, TimeUnit.SECONDS );