More refactoring

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/ReplicationModule.java

@@ -88,7 +88,8 @@ public ReplicationModule( MemberId myself, PlatformModule platformModule, Config
                 progressRetryStrategy,
                 leaderRetryStrategy,
                 availabilityTimeoutMillis,
-                globalAvailabilityGuard, logProvider, platformModule.monitors );
+                globalAvailabilityGuard, logProvider,
+                platformModule.monitors );
     }
 
     public RaftReplicator getReplicator()

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/core/state/machines/tx/ChunkedTransaction.java

@@ -109,32 +109,18 @@ public ByteBuf readChunk( ByteBufAllocator allocator ) throws Exception
             */
             channel.putInt( -1 );
         }
-        try
+
+        // write to chunks if empty and there is more to write
+        while ( txWriter.canWrite() && chunks.isEmpty() )
         {
-            // write to chunks if empty and there is more to write
-            while ( txWriter.canWrite() && chunks.isEmpty() )
-            {
-                txWriter.write( channel );
-            }
-            // nothing more to write, close the channel to get the potential last buffer
-            if ( chunks.isEmpty() )
-            {
-                channel.close();
-            }
-            return chunks.poll();
+            txWriter.write( channel );
         }
-        catch ( Throwable t )
+        // nothing more to write, close the channel to get the potential last buffer
+        if ( chunks.isEmpty() )
         {
-            try
-            {
-                close();
-            }
-            catch ( Exception e )
-            {
-                t.addSuppressed( e );
-            }
-            throw t;
+            channel.close();
         }
+        return chunks.poll();
     }
 
     @Override

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/messaging/ReconnectingChannel.java

@@ -25,8 +25,8 @@
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.EventLoop;
 import io.netty.util.AttributeKey;
-import io.netty.util.concurrent.DefaultPromise;
 import io.netty.util.concurrent.Promise;
 
 import java.time.Clock;
@@ -50,6 +50,7 @@ public class ReconnectingChannel implements Channel
 
     private final Log log;
     private final Bootstrap bootstrap;
+    private final EventLoop eventLoop;
     private final SocketAddress destination;
     private final TimeoutStrategy connectionBackoffStrategy;
 
@@ -61,14 +62,16 @@ public class ReconnectingChannel implements Channel
     private TimeoutStrategy.Timeout connectionBackoff;
     private CappedLogger cappedLogger;
 
-    ReconnectingChannel( Bootstrap bootstrap, SocketAddress destination, final Log log )
+    ReconnectingChannel( Bootstrap bootstrap, EventLoop eventLoop, SocketAddress destination, final Log log )
     {
-        this( bootstrap, destination, log, new ExponentialBackoffStrategy( 100, 1600, MILLISECONDS ) );
+        this( bootstrap, eventLoop, destination, log, new ExponentialBackoffStrategy( 100, 1600, MILLISECONDS ) );
     }
 
-    private ReconnectingChannel( Bootstrap bootstrap, SocketAddress destination, final Log log, TimeoutStrategy connectionBackoffStrategy )
+    private ReconnectingChannel( Bootstrap bootstrap, EventLoop eventLoop, SocketAddress destination, final Log log,
+            TimeoutStrategy connectionBackoffStrategy )
     {
         this.bootstrap = bootstrap;
+        this.eventLoop = eventLoop;
         this.destination = destination;
         this.log = log;
         this.cappedLogger = new CappedLogger( log ).setTimeLimit( 20, TimeUnit.SECONDS, Clock.systemUTC() );
@@ -157,32 +160,34 @@ private Future<Void> write( Object msg, boolean flush )
 
         if ( channel.isActive() )
         {
-            return doWrite( msg, flush );
+            if ( flush )
+            {
+                return channel.writeAndFlush( msg );
+            }
+            else
+            {
+                return channel.write( msg );
+            }
         }
         else
         {
-            Promise<Void> promise = new DefaultPromise<>( bootstrap.config().group().next() );
+            Promise<Void> promise = eventLoop.newPromise();
             BiConsumer<io.netty.channel.Channel,Object> writer;
 
-            writer = ( channel, message ) -> chain( doWrite( msg, flush ), promise );
+            if ( flush )
+            {
+                writer = ( channel, message ) -> chain( channel.writeAndFlush( msg ), promise );
+            }
+            else
+            {
+                writer = ( channel, message ) -> chain( channel.write( msg ), promise );
+            }
 
             deferredWrite( msg, fChannel, promise, true, writer );
             return promise;
         }
     }
 
-    private ChannelFuture doWrite( Object msg, boolean flush )
-    {
-        if ( flush )
-        {
-            return channel.writeAndFlush( msg );
-        }
-        else
-        {
-            return channel.write( msg );
-        }
-    }
-
     /**
      * Chains a channel future to a promise. Used when the returned promise
      * was not allocated through the channel and cannot be used as the

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/messaging/SenderService.java

@@ -107,7 +107,7 @@ private Channel channel( AdvertisedSocketAddress destination )
 
         if ( channel == null )
         {
-            channel = new ReconnectingChannel( bootstrap, destination, log );
+            channel = new ReconnectingChannel( bootstrap, eventLoopGroup.next(), destination, log );
             channel.start();
             ReconnectingChannel existingNonBlockingChannel = channels.putIfAbsent( destination, channel );
 

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/messaging/marshalling/v2/decoding/ReplicatedContentChunkDecoder.java

@@ -41,7 +41,7 @@ public class ReplicatedContentChunkDecoder extends ByteToMessageDecoder
 
     ReplicatedContentChunkDecoder()
     {
-        setCumulator( new ContentChunkAccumulator() );
+        setCumulator( new ContentChunkCumulator() );
     }
 
     @Override
@@ -60,7 +60,7 @@ protected void decode( ChannelHandlerContext ctx, ByteBuf in, List<Object> out )
         }
     }
 
-    private class ContentChunkAccumulator implements Cumulator
+    private class ContentChunkCumulator implements Cumulator
     {
         @Override
         public ByteBuf cumulate( ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in )

enterprise/causal-clustering/src/test/java/org/neo4j/causalclustering/messaging/ReconnectingChannelIT.java

@@ -90,7 +90,7 @@ public void before()
     {
         elg = new NioEventLoopGroup( 0 );
         Bootstrap bootstrap = new Bootstrap().channel( NioSocketChannel.class ).group( elg ).handler( childCounter );
-        channel = new ReconnectingChannel( bootstrap, listenAddress, log );
+        channel = new ReconnectingChannel( bootstrap, elg.next(), listenAddress, log );
     }
 
     @After