Cleans up ByteBuffer creation and use in some core-edge specific classes

This commit removes buffers that were allocated but not used, and makes
 sure to reuse buffers in NaiveDurableRaftLog instead of having them
 instantiated in every method call. This class is not thread safe anyway
 because of the way channels are written to.

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

@@ -25,17 +25,12 @@
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 
-import org.neo4j.coreedge.raft.log.monitoring.RaftLogAppendIndexMonitor;
-import org.neo4j.coreedge.raft.log.monitoring.RaftLogCommitIndexMonitor;
 import org.neo4j.coreedge.raft.replication.MarshallingException;
 import org.neo4j.coreedge.raft.replication.ReplicatedContent;
 import org.neo4j.coreedge.raft.replication.Serializer;
 import org.neo4j.io.fs.FileSystemAbstraction;
 import org.neo4j.io.fs.StoreChannel;
 import org.neo4j.kernel.lifecycle.LifecycleAdapter;
-import org.neo4j.kernel.monitoring.Monitors;
-
-import static org.neo4j.coreedge.raft.log.RaftLog.*;
 
 
 /**
@@ -73,6 +68,11 @@ public class NaiveDurableRaftLog extends LifecycleAdapter implements RaftLog
     public static final int COMMIT_INDEX_BYTES = 8;
     public static final String DIRECTORY_NAME = "raft-log";
 
+    private final ByteBuffer entryRecordBuffer = ByteBuffer.allocate( ENTRY_RECORD_LENGTH );
+    private final ByteBuffer contentLengthBuffer = ByteBuffer.allocate( CONTENT_LENGTH_BYTES );
+    private final ByteBuffer commitIndexBuffer = ByteBuffer.allocate( COMMIT_INDEX_BYTES );
+
+
     private final Set<Listener> listeners = new CopyOnWriteArraySet<>();
 
     private final StoreChannel entriesChannel;
@@ -111,7 +111,7 @@ public void shutdown() throws Throwable
     {
         Exception container = new Exception( "Exception happened during shutdown of RaftLog. See suppressed " +
                 "exceptions for details" );
-        boolean shouldThrow = false;
+        boolean shouldThrow;
         shouldThrow = forceAndCloseChannel( entriesChannel, container );
         shouldThrow = forceAndCloseChannel( contentChannel, container ) || shouldThrow;
         shouldThrow = forceAndCloseChannel( commitChannel, container ) || shouldThrow;
@@ -340,12 +340,12 @@ public Entry( long term, long contentPointer )
 
     private void writeEntry( Entry entry ) throws IOException
     {
-        ByteBuffer buffer = ByteBuffer.allocate( ENTRY_RECORD_LENGTH );
-        buffer.putLong( entry.term );
-        buffer.putLong( entry.contentPointer );
-        buffer.flip();
+        entryRecordBuffer.clear();
+        entryRecordBuffer.putLong( entry.term );
+        entryRecordBuffer.putLong( entry.contentPointer );
+        entryRecordBuffer.flip();
 
-        entriesChannel.writeAll( buffer, (appendIndex + 1) * ENTRY_RECORD_LENGTH );
+        entriesChannel.writeAll( entryRecordBuffer, (appendIndex + 1) * ENTRY_RECORD_LENGTH );
         entriesChannel.force( false );
     }
 
@@ -356,11 +356,11 @@ private Entry readEntry( long logIndex ) throws IOException
             return new Entry( -1, -1 );
         }
 
-        ByteBuffer buffer = ByteBuffer.allocate( ENTRY_RECORD_LENGTH );
-        entriesChannel.read( buffer, logIndex * ENTRY_RECORD_LENGTH );
-        buffer.flip();
-        long term = buffer.getLong();
-        long contentPointer = buffer.getLong();
+        entryRecordBuffer.clear();
+        entriesChannel.read( entryRecordBuffer, logIndex * ENTRY_RECORD_LENGTH );
+        entryRecordBuffer.flip();
+        long term = entryRecordBuffer.getLong();
+        long contentPointer = entryRecordBuffer.getLong();
         return new Entry( term, contentPointer );
     }
 
@@ -369,7 +369,7 @@ private int writeContent( RaftLogEntry logEntry ) throws MarshallingException, I
         ByteBuffer contentBuffer = serializer.serialize( logEntry.content() );
         int length = CONTENT_LENGTH_BYTES + contentBuffer.remaining();
 
-        ByteBuffer contentLengthBuffer = ByteBuffer.allocate( CONTENT_LENGTH_BYTES );
+        contentLengthBuffer.clear();
         contentLengthBuffer.putInt( length );
         contentLengthBuffer.flip();
         contentChannel.writeAll( contentLengthBuffer, contentOffset );
@@ -381,10 +381,10 @@ private int writeContent( RaftLogEntry logEntry ) throws MarshallingException, I
 
     private ReplicatedContent readContentFrom( long contentPointer ) throws IOException, MarshallingException
     {
-        ByteBuffer lengthBuffer = ByteBuffer.allocate( CONTENT_LENGTH_BYTES );
-        contentChannel.read( lengthBuffer, contentPointer );
-        lengthBuffer.flip();
-        int contentLength = lengthBuffer.getInt();
+        contentLengthBuffer.clear();
+        contentChannel.read( contentLengthBuffer, contentPointer );
+        contentLengthBuffer.flip();
+        int contentLength = contentLengthBuffer.getInt();
 
         ByteBuffer contentBuffer = ByteBuffer.allocate( contentLength - CONTENT_LENGTH_BYTES );
         contentChannel.read( contentBuffer, contentPointer + CONTENT_LENGTH_BYTES );
@@ -394,10 +394,10 @@ private ReplicatedContent readContentFrom( long contentPointer ) throws IOExcept
 
     private void storeCommitIndex( long commitIndex ) throws IOException
     {
-        ByteBuffer buffer = ByteBuffer.allocate( COMMIT_INDEX_BYTES );
-        buffer.putLong( commitIndex );
-        buffer.flip();
-        commitChannel.writeAll( buffer, 0 );
+        commitIndexBuffer.clear();
+        commitIndexBuffer.putLong( commitIndex );
+        commitIndexBuffer.flip();
+        commitChannel.writeAll( commitIndexBuffer, 0 );
         commitChannel.force( false );
     }
 
@@ -407,9 +407,9 @@ private long readCommitIndex() throws IOException
         {
             return -1;
         }
-        ByteBuffer buffer = ByteBuffer.allocate( COMMIT_INDEX_BYTES );
-        commitChannel.read( buffer, 0 );
-        buffer.flip();
-        return buffer.getLong();
+        commitIndexBuffer.clear();
+        commitChannel.read( commitIndexBuffer, 0 );
+        commitIndexBuffer.flip();
+        return commitIndexBuffer.getLong();
     }
 }

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/state/id_allocation/OnDiskIdAllocationState.java

@@ -49,7 +49,6 @@ public class OnDiskIdAllocationState extends LifecycleAdapter implements IdAlloc
     private final InMemoryIdAllocationState inMemoryIdAllocationState;
 
     private final StatePersister<InMemoryIdAllocationState> statePersister;
-    private final ByteBuffer workingBuffer;
     private final InMemoryIdAllocationState.InMemoryIdAllocationStateChannelMarshal marshal;
 
     public OnDiskIdAllocationState( FileSystemAbstraction fileSystemAbstraction, File stateDir,
@@ -59,8 +58,6 @@ public OnDiskIdAllocationState( FileSystemAbstraction fileSystemAbstraction, Fil
         File fileA = new File( stateDir, FILENAME + "a" );
         File fileB = new File( stateDir, FILENAME + "b" );
 
-        this.workingBuffer = ByteBuffer.allocate( NUMBER_OF_BYTES_PER_WRITE );
-
         IdAllocationStateRecoveryManager recoveryManager =
                 new IdAllocationStateRecoveryManager( fileSystemAbstraction,
                         new InMemoryIdAllocationState.InMemoryIdAllocationStateChannelMarshal() );

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/state/term/OnDiskTermState.java

@@ -38,7 +38,6 @@ public class OnDiskTermState extends LifecycleAdapter implements TermState
     public static final String FILENAME = "term.";
     public static final String DIRECTORY_NAME = "term-state";
 
-    private final ByteBuffer workingBuffer;
     private InMemoryTermState inMemoryTermState;
 
     private final StatePersister<InMemoryTermState> statePersister;
@@ -50,20 +49,14 @@ public OnDiskTermState( FileSystemAbstraction fileSystemAbstraction, File stateD
         File fileA = new File( stateDir, FILENAME + "a" );
         File fileB = new File( stateDir, FILENAME + "b" );
 
-        workingBuffer = ByteBuffer.allocate( InMemoryVoteState.InMemoryVoteStateChannelMarshal
-                .NUMBER_OF_BYTES_PER_VOTE );
-
-
         TermStateRecoveryManager recoveryManager =
                 new TermStateRecoveryManager( fileSystemAbstraction, new InMemoryTermStateChannelMarshal() );
 
         final StateRecoveryManager.RecoveryStatus recoveryStatus = recoveryManager.recover( fileA, fileB );
 
-
         this.inMemoryTermState = recoveryManager.readLastEntryFrom( fileSystemAbstraction, recoveryStatus
                 .previouslyActive() );
 
-
         this.statePersister = new StatePersister<>( fileA, fileB, fileSystemAbstraction, numberOfEntriesBeforeRotation,
                 new InMemoryTermStateChannelMarshal(), recoveryStatus.previouslyInactive(),
                 databaseHealthSupplier );

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/state/vote/OnDiskVoteState.java

@@ -48,9 +48,6 @@ public OnDiskVoteState( FileSystemAbstraction fileSystemAbstraction, File stateD
         File fileA = new File( stateDir, FILENAME + "a" );
         File fileB = new File( stateDir, FILENAME + "b" );
 
-        ByteBuffer workingBuffer = ByteBuffer.allocate( InMemoryVoteState.InMemoryVoteStateChannelMarshal
-                .NUMBER_OF_BYTES_PER_VOTE );
-
         InMemoryVoteState.InMemoryVoteStateChannelMarshal<MEMBER> marshal =
                 new InMemoryVoteState.InMemoryVoteStateChannelMarshal<>( memberByteBufferMarshal );