Improve kernel panic when applying transactions to the log

Before the BatchingTransactionAppender was raising a kernel panic when
not being able to append transaction to the log, but not when failing
to force the log to disk. Such a case was dealt with by the
TransactionRepresentationCommitProcess. Now all the responsibility for
kernel panicing is moved down into BatchingTransactionAppander.

community/kernel/src/main/java/org/neo4j/kernel/impl/api/TransactionRepresentationCommitProcess.java

@@ -33,6 +33,7 @@
 import org.neo4j.kernel.impl.transaction.tracing.StoreApplyEvent;
 
 import static org.neo4j.kernel.api.exceptions.Status.Transaction.CouldNotCommit;
+import static org.neo4j.kernel.api.exceptions.Status.Transaction.CouldNotWriteToLog;
 import static org.neo4j.kernel.api.exceptions.Status.Transaction.ValidationFailed;
 
 public class TransactionRepresentationCommitProcess implements TransactionCommitProcess
@@ -88,9 +89,9 @@ private long appendToLog(
         {
             transactionId = appender.append( transaction, logAppendEvent );
         }
-        catch ( Throwable e )
+        catch ( Throwable cause )
         {
-            throw exception( Status.Transaction.CouldNotWriteToLog, e,
+            throw new TransactionFailureException( CouldNotWriteToLog, cause,
                     "Could not append transaction representation to log" );
         }
         commitEvent.setTransactionId( transactionId );

community/kernel/src/main/java/org/neo4j/kernel/impl/transaction/log/BatchingTransactionAppender.java

@@ -189,10 +189,10 @@ public void checkPoint( LogPosition logPosition, LogCheckPointEvent logCheckPoin
                 }
             }
         }
-        catch ( Throwable t )
+        catch ( Throwable cause )
         {
-            kernelHealth.panic( t );
-            throw t;
+            kernelHealth.panic( cause );
+            throw cause;
         }
 
         forceAfterAppend( logCheckPointEvent );
@@ -226,11 +226,10 @@ public void publishAsCommitted()
 
     /**
      * @return A TransactionCommitment instance with metadata about the committed transaction, such as whether or not
-     * this transaction
-     * contains any legacy index changes.
+     * this transaction contains any legacy index changes.
      */
-    private TransactionCommitment appendToLog( TransactionRepresentation transaction, long transactionId ) throws
-            IOException
+    private TransactionCommitment appendToLog( TransactionRepresentation transaction, long transactionId )
+            throws IOException
     {
         // Reset command writer so that we, after we've written the transaction, can ask it whether or
         // not any legacy index command was written. If so then there's additional ordering to care about below.
@@ -338,7 +337,15 @@ private void forceLog( LogForceEvents logForceEvents ) throws IOException
         {
             force();
         }
-        unparkAll( links );
+        catch ( final Throwable panic )
+        {
+            kernelHealth.panic( panic );
+            throw panic;
+        }
+        finally
+        {
+            unparkAll( links );
+        }
     }
 
     private void unparkAll( ThreadLink links )

community/kernel/src/test/java/org/neo4j/kernel/impl/transaction/log/BatchingTransactionAppenderTest.java

@@ -60,6 +60,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -241,7 +242,45 @@ public void shouldNotCallTransactionCommittedOnFailedAppendedTransaction() throw
         catch ( IOException e )
         {
             // THEN
-            assertTrue( contains( e, failureMessage, IOException.class ) );
+            assertSame( failure, e );
+            verify( transactionIdStore, times( 1 ) ).nextCommittingTransactionId();
+            verify( transactionIdStore, times( 1 ) ).transactionClosed( txId );
+            verify( kernelHealth ).panic( failure );
+        }
+    }
+
+    @Test
+    public void shouldNotCallTransactionCommittedOnFailedForceLogToDisk() throws Exception
+    {
+        // GIVEN
+        long txId = 3;
+        String failureMessage = "Forces a failure";
+        WritableLogChannel channel = spy( new InMemoryLogChannel() );
+        IOException failure = new IOException( failureMessage );
+        doThrow( failure ).when( channel ).force();
+        LogFile logFile = mock( LogFile.class );
+        when( logFile.getWriter() ).thenReturn( channel );
+        TransactionMetadataCache metadataCache = new TransactionMetadataCache( 10, 10 );
+        TransactionIdStore transactionIdStore = mock( TransactionIdStore.class );
+        when( transactionIdStore.nextCommittingTransactionId() ).thenReturn( txId );
+        Mockito.reset( kernelHealth );
+        TransactionAppender appender = life.add( new BatchingTransactionAppender( logFile, NO_ROTATION,
+                metadataCache, transactionIdStore, BYPASS, kernelHealth ) );
+
+        life.start();
+
+        // WHEN
+        TransactionRepresentation transaction = mock( TransactionRepresentation.class );
+        when( transaction.additionalHeader() ).thenReturn( new byte[0] );
+        try
+        {
+            appender.append( transaction, logAppendEvent );
+            fail( "Expected append to fail. Something is wrong with the test itself" );
+        }
+        catch ( IOException e )
+        {
+            // THEN
+            assertSame( failure, e );
             verify( transactionIdStore, times( 1 ) ).nextCommittingTransactionId();
             verify( transactionIdStore, times( 1 ) ).transactionClosed( txId );
             verify( kernelHealth ).panic( failure );