Address review comments

community/bolt/src/main/java/org/neo4j/bolt/runtime/BoltConnection.java

@@ -22,6 +22,7 @@
 import io.netty.channel.Channel;
 
 import java.net.SocketAddress;
+import java.util.concurrent.RejectedExecutionException;
 
 import org.neo4j.bolt.v1.runtime.Job;
 
@@ -56,13 +57,6 @@ public interface BoltConnection
      */
     Channel channel();
 
-    /**
-     * Returns the principal the client used to connect
-     *
-     * @return user name
-     */
-    String principal();
-
     /**
      * Returns whether there's any pending Job waiting to be processed
      *
@@ -89,6 +83,13 @@ public interface BoltConnection
      */
     boolean processNextBatch();
 
+    /**
+     * Invoked when an exception is caught during the scheduling of the pending jobs. The caught exception would mostly
+     * be {@link RejectedExecutionException} which is thrown by the thread pool executor when it fails to accept
+     * submitted jobs
+     *
+     * @param t the exception occurred during scheduling
+     */
     void handleSchedulingError( Throwable t );
 
     /**

community/bolt/src/main/java/org/neo4j/bolt/runtime/BoltConnectionQueueMonitorAggregate.java

@@ -33,7 +33,7 @@ public class BoltConnectionQueueMonitorAggregate implements BoltConnectionQueueM
 
     public BoltConnectionQueueMonitorAggregate( BoltConnectionQueueMonitor... monitors )
     {
-        this.monitors = new ArrayList<>( Arrays.asList( monitors ) );
+        this.monitors = Arrays.asList( monitors );
     }
 
     @Override

community/bolt/src/main/java/org/neo4j/bolt/runtime/BoltConnectionReadLimiter.java

@@ -75,13 +75,13 @@ protected int getHighWatermark()
     @Override
     public void enqueued( BoltConnection to, Job job )
     {
-        checkLimitsOnEnqueue( to, counters.compute( to.id(), ( k, v ) -> v == null ? new AtomicInteger( 0 ) : v ).incrementAndGet() );
+        checkLimitsOnEnqueue( to, counters.computeIfAbsent( to.id(), k -> new AtomicInteger( 0 ) ).incrementAndGet() );
     }
 
     @Override
     public void drained( BoltConnection from, Collection<Job> batch )
     {
-        checkLimitsOnDequeue( from, counters.compute( from.id(), ( k, v ) -> v == null ? new AtomicInteger( 0 ) : v ).addAndGet( -batch.size() ) );
+        checkLimitsOnDequeue( from, counters.computeIfAbsent( from.id(), k -> new AtomicInteger( 0 ) ).addAndGet( -batch.size() ) );
     }
 
     private void checkLimitsOnEnqueue( BoltConnection connection, int currentSize )

community/bolt/src/main/java/org/neo4j/bolt/runtime/BoltScheduler.java

@@ -22,6 +22,8 @@
 public interface BoltScheduler extends BoltConnectionLifetimeListener, BoltConnectionQueueMonitor
 {
 
+    String connector();
+
     void start();
 
     void stop();

community/bolt/src/main/java/org/neo4j/bolt/runtime/CachedThreadPoolExecutorFactory.java

@@ -61,48 +61,19 @@ public ExecutorService create( int corePoolSize, int maxPoolSize, Duration keepA
         return result;
     }
 
-    @Override
-    public void destroy( ExecutorService executor )
-    {
-        if ( !( executor instanceof ThreadPool ) )
-        {
-            throw new IllegalArgumentException(
-                    String.format( "The passed executor should already be created by '%s#create()'.", CachedThreadPoolExecutorFactory.class.getName() ) );
-        }
-
-        executor.shutdown();
-        try
-        {
-            if ( !executor.awaitTermination( 10, TimeUnit.SECONDS ) )
-            {
-                executor.shutdownNow();
-
-                if ( !executor.awaitTermination( 10, TimeUnit.SECONDS ) )
-                {
-                    log.error( "Thread pool did not terminate gracefully despite all efforts" );
-                }
-            }
-        }
-        catch ( InterruptedException ex )
-        {
-            executor.shutdownNow();
-            Thread.currentThread().interrupt();
-        }
-    }
-
-    private static BlockingQueue createTaskQueue( int queueSize )
+    private static BlockingQueue<Runnable> createTaskQueue( int queueSize )
     {
         if ( queueSize == UNBOUNDED_QUEUE )
         {
-            return new LinkedBlockingQueue();
+            return new LinkedBlockingQueue<>();
         }
         else if ( queueSize == SYNCHRONOUS_QUEUE )
         {
-            return new SynchronousQueue();
+            return new SynchronousQueue<>();
         }
         else if ( queueSize > 0 )
         {
-            return new ArrayBlockingQueue( queueSize );
+            return new ArrayBlockingQueue<>( queueSize );
         }
 
         throw new IllegalArgumentException( String.format( "Unsupported queue size %d for thread pool creation.", queueSize ) );

community/bolt/src/main/java/org/neo4j/bolt/runtime/DefaultBoltConnection.java

@@ -108,12 +108,6 @@ public Channel channel()
         return channel.rawChannel();
     }
 
-    @Override
-    public String principal()
-    {
-        return machine.owner();
-    }
-
     @Override
     public boolean hasPendingJobs()
     {
@@ -129,8 +123,7 @@ public void start()
     @Override
     public void enqueue( Job job )
     {
-        queue.offer( job );
-        notifyEnqueued( job );
+        enqueueInternal( job );
     }
 
     @Override
@@ -254,10 +247,11 @@ public void stop()
         {
             machine.terminate();
 
-            if ( !hasPendingJobs() )
+            // Enqueue an empty job for close to be handled linearly
+            enqueueInternal( ignore ->
             {
-                processNextBatch( 0 );
-            }
+
+            } );
         }
     }
 
@@ -280,6 +274,12 @@ private void close()
         }
     }
 
+    private void enqueueInternal( Job job )
+    {
+        queue.offer( job );
+        notifyEnqueued( job );
+    }
+
     private void notifyCreated()
     {
         if ( listener != null )

community/bolt/src/main/java/org/neo4j/bolt/runtime/ExecutorBoltScheduler.java

@@ -51,8 +51,7 @@ public class ExecutorBoltScheduler implements BoltScheduler, BoltConnectionLifet
     private ExecutorService threadPool;
 
     public ExecutorBoltScheduler( String connector, ExecutorFactory executorFactory, JobScheduler scheduler, LogService logService, int corePoolSize,
-            int maxPoolSize,
-            Duration keepAlive, int queueSize, ExecutorService forkJoinPool )
+            int maxPoolSize, Duration keepAlive, int queueSize, ExecutorService forkJoinPool )
     {
         this.connector = connector;
         this.executorFactory = executorFactory;
@@ -75,24 +74,36 @@ boolean isActive( BoltConnection connection )
         return activeWorkItems.containsKey( connection.id() );
     }
 
+    @Override
+    public String connector()
+    {
+        return connector;
+    }
+
+    @Override
     public void start()
     {
         threadPool = executorFactory.create( corePoolSize, maxPoolSize, keepAlive, queueSize,
                 new NameAppendingThreadFactory( connector, scheduler.threadFactory( JobScheduler.Groups.boltWorker ) ) );
     }
 
+    @Override
     public void stop()
     {
         if ( threadPool != null )
         {
-            executorFactory.destroy( threadPool );
+            activeConnections.values().forEach( this::stopConnection );
+
+            threadPool.shutdown();
         }
     }
 
     @Override
     public void created( BoltConnection connection )
     {
-        activeConnections.put( connection.id(), connection );
+        BoltConnection previous = activeConnections.put( connection.id(), connection );
+        // We do not expect the same (keyed) connection twice
+        assert previous == null;
     }
 
     @Override
@@ -102,7 +113,7 @@ public void closed( BoltConnection connection )
 
         try
         {
-            CompletableFuture currentFuture = activeWorkItems.remove( id );
+            CompletableFuture<Boolean> currentFuture = activeWorkItems.remove( id );
             if ( currentFuture != null )
             {
                 currentFuture.cancel( true );
@@ -144,7 +155,7 @@ private boolean executeBatch( BoltConnection connection )
     {
         Thread currentThread = Thread.currentThread();
         String originalName = currentThread.getName();
-        String newName = String.format( "%s [%s] ", originalName, connection.remoteAddress(), connector );
+        String newName = String.format( "%s [%s] ", originalName, connection.remoteAddress() );
 
         currentThread.setName( newName );
         try
@@ -157,7 +168,7 @@ private boolean executeBatch( BoltConnection connection )
         }
     }
 
-    private void handleCompletion( BoltConnection connection, Object shouldContinueScheduling, Throwable error )
+    private void handleCompletion( BoltConnection connection, Boolean shouldContinueScheduling, Throwable error )
     {
         CompletableFuture<Boolean> previousFuture = activeWorkItems.remove( connection.id() );
 
@@ -175,13 +186,25 @@ private void handleCompletion( BoltConnection connection, Object shouldContinueS
         }
         else
         {
-            if ( (Boolean)shouldContinueScheduling && connection.hasPendingJobs() )
+            if ( shouldContinueScheduling && connection.hasPendingJobs() )
             {
-                previousFuture.thenAcceptAsync( ignore -> handleSubmission( connection ), forkJoinPool );
+                handleSubmission( connection );
             }
         }
     }
 
+    private void stopConnection( BoltConnection connection )
+    {
+        try
+        {
+            connection.stop();
+        }
+        catch ( Throwable t )
+        {
+            log.warn( String.format( "An unexpected error occurred while stopping BoltConnection [%s]", connection.id() ), t );
+        }
+    }
+
     private static class NameAppendingThreadFactory implements ThreadFactory
     {
         private final String nameToAppend;

community/bolt/src/main/java/org/neo4j/bolt/runtime/ExecutorBoltSchedulerProvider.java

@@ -27,6 +27,7 @@
 import org.neo4j.kernel.configuration.Config;
 import org.neo4j.kernel.impl.logging.LogService;
 import org.neo4j.kernel.lifecycle.LifecycleAdapter;
+import org.neo4j.logging.Log;
 import org.neo4j.scheduler.JobScheduler;
 
 public class ExecutorBoltSchedulerProvider extends LifecycleAdapter implements BoltSchedulerProvider
@@ -35,6 +36,7 @@ public class ExecutorBoltSchedulerProvider extends LifecycleAdapter implements B
     private final ExecutorFactory executorFactory;
     private final JobScheduler scheduler;
     private final LogService logService;
+    private final Log internalLog;
     private final ConcurrentHashMap<String, BoltScheduler> boltSchedulers;
 
     private ExecutorService forkJoinThreadPool;
@@ -45,11 +47,12 @@ public ExecutorBoltSchedulerProvider( Config config, ExecutorFactory executorFac
         this.executorFactory = executorFactory;
         this.scheduler = scheduler;
         this.logService = logService;
+        this.internalLog = logService.getInternalLog( getClass() );
         this.boltSchedulers = new ConcurrentHashMap<>();
     }
 
     @Override
-    public void start() throws Throwable
+    public void start()
     {
         forkJoinThreadPool = new ForkJoinPool();
         config.enabledBoltConnectors().forEach( connector ->
@@ -64,15 +67,27 @@ public void start() throws Throwable
     }
 
     @Override
-    public void stop() throws Throwable
+    public void stop()
     {
-        boltSchedulers.values().forEach( s -> s.stop() );
+        boltSchedulers.values().forEach( this::stopScheduler );
         boltSchedulers.clear();
 
         forkJoinThreadPool.shutdown();
         forkJoinThreadPool = null;
     }
 
+    private void stopScheduler( BoltScheduler scheduler )
+    {
+        try
+        {
+            scheduler.stop();
+        }
+        catch ( Throwable t )
+        {
+            internalLog.warn( String.format( "An unexpected error occurred while stopping BoltScheduler [%s]", scheduler.connector() ), t );
+        }
+    }
+
     @Override
     public BoltScheduler get( BoltChannel channel )
     {

community/bolt/src/main/java/org/neo4j/bolt/runtime/ExecutorFactory.java

@@ -28,6 +28,4 @@ public interface ExecutorFactory
 
     ExecutorService create( int corePoolSize, int maxPoolSize, Duration keepAlive, int queueSize, ThreadFactory threadFactory );
 
-    void destroy( ExecutorService executor );
-
 }

community/bolt/src/main/java/org/neo4j/bolt/v1/runtime/TransactionStateMachineSPI.java

@@ -150,13 +150,13 @@ public BoltResult start() throws KernelException
                 }
                 catch ( KernelException e )
                 {
-                    transactionalContext.close( false );
+                    close( false );
                     onFail.apply();
                     throw new QueryExecutionKernelException( e );
                 }
                 catch ( Throwable e )
                 {
-                    transactionalContext.close( false );
+                    close( false );
                     onFail.apply();
                     throw e;
                 }