Improve error and close handling in decoders

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/core/consensus/protocol/v2/RaftProtocolServerInstallerV2.java

@@ -81,7 +81,6 @@ public void install( Channel channel ) throws Exception
                 .server( channel, log )
                 .modify( modifiers )
                 .addFraming()
-                .onClose( decodingDispatcher::close )
                 .add( "raft_content_type_dispatcher", new ContentTypeDispatcher( contentTypeProtocol ) )
                 .add( "raft_component_decoder", decodingDispatcher )
                 .add( "raft_content_decoder", new ReplicatedContentDecoder( contentTypeProtocol ) )

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

@@ -33,13 +33,15 @@
 import org.neo4j.causalclustering.messaging.marshalling.v2.ContentType;
 import org.neo4j.logging.LogProvider;
 
-public class DecodingDispatcher extends RequestDecoderDispatcher<ContentType> implements AutoCloseable
+public class DecodingDispatcher extends RequestDecoderDispatcher<ContentType>
 {
     private final ReplicatedContentChunkDecoder decoder;
 
     public DecodingDispatcher( Protocol<ContentType> protocol, LogProvider logProvider )
     {
         super( protocol, logProvider );
+        decoder = new ReplicatedContentChunkDecoder();
+
         register( ContentType.ContentType, new ByteToMessageDecoder()
         {
             @Override
@@ -52,13 +54,12 @@ protected void decode( ChannelHandlerContext ctx, ByteBuf in, List<Object> out )
             }
         } );
         register( ContentType.RaftLogEntryTerms, new RaftLogEntryTermsDecoder( protocol ) );
-        decoder = new ReplicatedContentChunkDecoder();
         register( ContentType.ReplicatedContent, decoder );
         register( ContentType.Message, new RaftMessageDecoder( protocol ) );
     }
 
     @Override
-    public void close()
+    public void channelInactive( ChannelHandlerContext ctx )
     {
         decoder.close();
     }

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

@@ -34,28 +34,24 @@
 
 public class ReplicatedContentChunkDecoder extends ByteToMessageDecoder implements AutoCloseable
 {
+    private final CoreReplicatedContentMarshal contentMarshal = new CoreReplicatedContentMarshal();
+
     private UnfinishedChunk unfinishedChunk;
-    private final CoreReplicatedContentMarshal coreReplicatedContentMarshal = new CoreReplicatedContentMarshal();
-    private boolean closed;
 
     @Override
     protected void decode( ChannelHandlerContext ctx, ByteBuf in, List<Object> out ) throws Exception
     {
-        if ( closed )
-        {
-            return;
-        }
         try
         {
-
             boolean isLast = in.readBoolean();
             if ( unfinishedChunk == null )
             {
                 byte contentType = in.readByte();
                 int allocationSize = in.readInt();
                 if ( isLast )
                 {
-                    out.add( coreReplicatedContentMarshal.read( contentType,
+
+                    out.add( contentMarshal.read( contentType,
                             new NetworkReadableClosableChannelNetty4( in.readSlice( in.readableBytes() ) ) ) );
                 }
                 else
@@ -79,14 +75,14 @@ protected void decode( ChannelHandlerContext ctx, ByteBuf in, List<Object> out )
 
                 if ( isLast )
                 {
-                    out.add( coreReplicatedContentMarshal.read( unfinishedChunk.contentType,
+                    out.add( contentMarshal.read( unfinishedChunk.contentType,
                             new NetworkReadableClosableChannelNetty4( unfinishedChunk.content() ) ) );
                     unfinishedChunk.release();
                     unfinishedChunk = null;
                 }
             }
         }
-        catch ( Exception e )
+        catch ( Throwable e )
         {
             release();
             throw e;
@@ -98,12 +94,13 @@ private void release()
         if ( unfinishedChunk != null )
         {
             unfinishedChunk.release();
+            unfinishedChunk = null;
         }
     }
 
+    @Override
     public void close()
     {
-        closed = true;
         release();
     }