TODO tidyup - mostly removing unnecessary catch(Throwable)

community/kernel/src/main/java/org/neo4j/kernel/impl/util/Neo4jJobScheduler.java

@@ -51,27 +51,13 @@ public void init()
     @Override
     public Executor executor( final Group group )
     {
-        return new Executor()
-        {
-            @Override
-            public void execute( Runnable command )
-            {
-                schedule( group, command );
-            }
-        };
+        return job -> schedule( group, job );
     }
 
     @Override
     public ThreadFactory threadFactory( final Group group )
     {
-        return new ThreadFactory()
-        {
-            @Override
-            public Thread newThread( Runnable r )
-            {
-                return createNewThread( group, r, NO_METADATA );
-            }
-        };
+        return job -> createNewThread( group, job, NO_METADATA );
     }
 
     @Override
@@ -130,6 +116,11 @@ public JobHandle schedule( Group group, final Runnable runnable, long initialDel
         }
     }
 
+    @Override
+    public void stop()
+    {
+    }
+
     @Override
     public void shutdown()
     {

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/DelayedRenewableTimeoutService.java

@@ -61,7 +61,7 @@ public class DelayedRenewableTimeoutService extends LifecycleAdapter implements
     private final Clock clock;
     private final Log log;
     private final Random random;
-    private final JobScheduler scheduler;
+    private final Neo4jJobScheduler scheduler;
     private JobScheduler.JobHandle jobHandle;
 
     public DelayedRenewableTimeoutService( Clock clock, LogProvider logProvider )
@@ -118,69 +118,60 @@ private long calcTimeoutTimestamp( long milliseconds, long randomRange )
     @Override
     public synchronized void run()
     {
-        try
+
+        long now = clock.millis();
+        Collection<ScheduledRenewableTimeout> triggered = new LinkedList<>();
+
+        synchronized ( timeouts )
         {
-            long now = clock.millis();
-            Collection<ScheduledRenewableTimeout> triggered = new LinkedList<>();
+            // Handle renewals
+            ScheduledRenewableTimeout renew;
+            while ( (renew = pendingRenewals.poll()) != null )
+            {
+                timeouts.remove( renew );
+                renew.setTimeoutTimestamp( calcTimeoutTimestamp( renew.timeoutLength, renew.randomRange ) );
+                timeouts.add( renew );
+            }
 
-            synchronized ( timeouts )
+            // Trigger timeouts
+            for ( ScheduledRenewableTimeout timeout : timeouts )
             {
-                // Handle renewals
-                ScheduledRenewableTimeout renew;
-                while ( (renew = pendingRenewals.poll()) != null )
+                if ( timeout.shouldTrigger( now ) )
                 {
-                    timeouts.remove( renew );
-                    renew.setTimeoutTimestamp( calcTimeoutTimestamp( renew.timeoutLength, renew.randomRange ) );
-                    timeouts.add( renew );
+                    triggered.add( timeout );
                 }
-
-                // Trigger timeouts
-                for ( ScheduledRenewableTimeout timeout : timeouts )
+                else
                 {
-                    if ( timeout.shouldTrigger( now ) )
-                    {
-                        triggered.add( timeout );
-                    }
-                    else
-                    {
-                        // Since the timeouts are sorted, the first timeout we hit that should not be triggered means
-                        // there are no others that should either, so we bail.
-                        break;
-                    }
+                    // Since the timeouts are sorted, the first timeout we hit that should not be triggered means
+                    // there are no others that should either, so we bail.
+                    break;
                 }
             }
+        }
 
-            for ( ScheduledRenewableTimeout timeout : triggered )
-            {
-                timeout.trigger();
-            }
+        triggered.forEach( ScheduledRenewableTimeout::trigger );
 
-            synchronized ( timeouts )
-            {
-                timeouts.removeAll( triggered );
-            }
-        }
-        catch ( Throwable e )
+        synchronized ( timeouts )
         {
-            log.error( "Error handling timeouts", e );
+            timeouts.removeAll( triggered );
         }
     }
 
     @Override
-    public void init() throws Throwable
+    public void init()
     {
         scheduler.init();
     }
 
     @Override
-    public void start() throws Throwable
+    public void start()
     {
         jobHandle = scheduler.scheduleRecurring( new JobScheduler.Group( "Scheduler", POOLED ), this, TIMER_RESOLUTION,
                 TIMER_RESOLUTION_UNIT );
     }
 
     @Override
-    public void stop() throws Throwable
+    public void stop()
     {
         jobHandle.cancel( false );
         scheduler.stop();

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/replication/shipping/RaftLogShipper.java

@@ -41,6 +41,7 @@
 import static java.lang.String.format;
 
 import static org.neo4j.coreedge.raft.RenewableTimeoutService.RenewableTimeout;
+import static org.neo4j.coreedge.raft.replication.shipping.RaftLogShipper.Mode.PIPELINE;
 import static org.neo4j.coreedge.raft.replication.shipping.RaftLogShipper.Timeouts.RESEND;
 
 /// Optimizations
@@ -155,25 +156,11 @@ public Object identity()
     public synchronized void start()
     {
         log.info( "Starting log shipper: %s", statusAsString() );
-
-        try
-        {
-            timeoutService = new DelayedRenewableTimeoutService( clock, logProvider );
-            timeoutService.init();
-            timeoutService.start();
-        }
-        catch ( Throwable e )
-        {
-            // TODO: Think about how to handle this. We cannot be allowed to throw when
-            // TODO: starting the log shippers from the main RAFT handling. The timeout
-            // TODO: service is a LifeCycle.
-            // TODO: Should we have and use one system level timeout service instead?
-
-            log.error( "Failed to start log shipper " + statusAsString(), e );
-        }
-
+        timeoutService = new DelayedRenewableTimeoutService( clock, logProvider );
+        timeoutService.init();
+        timeoutService.start();
         sendSingle( raftLog.appendIndex(), lastLeaderContext );
-    }
+   }
 
     public synchronized void stop()
     {
@@ -222,7 +209,7 @@ public synchronized void onMatch( long newMatchIndex, LeaderContext leaderContex
                 if ( sendNextBatchAfterMatch( leaderContext ) )
                 {
                     log.info( "%s: caught up after mismatch, moving to PIPELINE mode", statusAsString() );
-                    mode = Mode.PIPELINE;
+                    mode = PIPELINE;
                 }
                 else
                 {
@@ -236,7 +223,7 @@ public synchronized void onMatch( long newMatchIndex, LeaderContext leaderContex
                     if ( sendNextBatchAfterMatch( leaderContext ) )
                     {
                         log.info( "%s: caught up, moving to PIPELINE mode", statusAsString() );
-                        mode = Mode.PIPELINE;
+                        mode = PIPELINE;
                     }
                 }
                 break;
@@ -300,49 +287,40 @@ public synchronized void onCommitUpdate( LeaderContext leaderContext )
 
     private synchronized void onScheduledTimeoutExpiry()
     {
-        try
+        if ( timedOut() )
         {
-            if ( timedOut() )
-            {
-                onTimeout();
-            }
-            else if ( timeoutAbsoluteMillis != 0 )
-            {
-                long timeLeft = timeoutAbsoluteMillis - clock.millis();
+            onTimeout();
+            return;
+        }
 
-                if ( timeLeft > 0 )
-                {
-                    scheduleTimeout( timeLeft );
-                }
-                else
-                {
-                    onTimeout();
-                }
-            }
+        if ( timeoutAbsoluteMillis <= 0 )
+        {
+            return;
         }
-        catch ( Throwable e )
+
+        long timeLeft = timeoutAbsoluteMillis - clock.millis();
+        if ( timeLeft > 0 )
+        {
+            scheduleTimeout( timeLeft );
+        }
+        else
         {
-            log.error( "Exception during timeout handling: " + statusAsString(), e );
+            onTimeout();
         }
     }
 
-    private void onTimeout() throws IOException
+    private void onTimeout()
     {
-        switch ( mode )
+        if ( mode == PIPELINE )
         {
-            case PIPELINE:
-            /* we leave pipelined mode here, because the follower seems
-             * unresponsive and we do not want to spam it with new entries */
-                log.info( "%s: timed out, moving to CATCHUP mode", statusAsString() );
-                mode = Mode.CATCHUP;
-            /* fallthrough */
-            case CATCHUP:
-            case MISMATCH:
-                if ( lastLeaderContext != null )
-                {
-                    sendSingle( lastSentIndex, lastLeaderContext );
-                }
-                break;
+            /* The follower seems unresponsive and we do not want to spam it with new entries */
+            log.info( "%s: timed out, moving to CATCHUP mode", statusAsString() );
+            mode = Mode.CATCHUP;
+        }
+
+        if ( lastLeaderContext != null )
+        {
+            sendSingle( lastSentIndex, lastLeaderContext );
         }
     }
 
@@ -417,8 +395,8 @@ private void sendSingle( long logIndex, LeaderContext leaderContext )
         sendRange( logIndex, logIndex, leaderContext );
     }
 
-    private void sendNewEntries( long prevLogIndex, long prevLogTerm, RaftLogEntry[] newEntries, LeaderContext
-            leaderContext )
+    private void sendNewEntries( long prevLogIndex, long prevLogTerm, RaftLogEntry[] newEntries,
+                                 LeaderContext leaderContext )
     {
         scheduleTimeout( retryTimeMillis );