Allow transaction catchuper to append logs ino correct directory

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

@@ -51,6 +51,7 @@ public class TransactionLogFiles extends LifecycleAdapter implements LogFiles
 {
     public static final String DEFAULT_NAME = "neostore.transaction.db";
     public static final FilenameFilter DEFAULT_FILENAME_FILTER = TransactionLogFilesHelper.DEFAULT_FILENAME_FILTER;
+    private static final File[] EMPTY_FILES_ARRAY = {};
 
     private final TransactionLogFilesContext logFilesContext;
     private final TransactionLogFileInformation logFileInformation;
@@ -107,7 +108,12 @@ public long getLogVersion( String historyLogFilename )
     @Override
     public File[] logFiles()
     {
-        return fileSystem.listFiles( fileHelper.getParentDirectory(), fileHelper.getLogFilenameFilter() );
+        File[] files = fileSystem.listFiles( fileHelper.getParentDirectory(), fileHelper.getLogFilenameFilter() );
+        if ( files == null )
+        {
+            return EMPTY_FILES_ARRAY;
+        }
+        return files;
     }
 
     @Override

enterprise/backup/src/test/java/org/neo4j/backup/BackupFlowTest.java

@@ -35,6 +35,7 @@
 import org.neo4j.consistency.checking.full.ConsistencyCheckIncompleteException;
 import org.neo4j.consistency.checking.full.ConsistencyFlags;
 import org.neo4j.helpers.progress.ProgressMonitorFactory;
+import org.neo4j.io.fs.FileSystemAbstraction;
 import org.neo4j.logging.LogProvider;
 
 import static org.hamcrest.CoreMatchers.equalTo;
@@ -53,26 +54,28 @@ public class BackupFlowTest
     public ExpectedException expectedException = ExpectedException.none();
 
     // dependencies
-    final ConsistencyCheckService consistencyCheckService = mock( ConsistencyCheckService.class );
-    final OutsideWorld outsideWorld = mock( OutsideWorld.class );
-    final LogProvider logProvider = mock( LogProvider.class );
-    final BackupStrategyWrapper firstStrategy = mock( BackupStrategyWrapper.class );
-    final BackupStrategyWrapper secondStrategy = mock( BackupStrategyWrapper.class );
+    private final ConsistencyCheckService consistencyCheckService = mock( ConsistencyCheckService.class );
+    private final OutsideWorld outsideWorld = mock( OutsideWorld.class );
+    private final FileSystemAbstraction fileSystem = mock( FileSystemAbstraction.class );
+    private final LogProvider logProvider = mock( LogProvider.class );
+    private final BackupStrategyWrapper firstStrategy = mock( BackupStrategyWrapper.class );
+    private final BackupStrategyWrapper secondStrategy = mock( BackupStrategyWrapper.class );
 
     BackupFlow subject;
 
     // test method parameter mocks
-    final OnlineBackupContext onlineBackupContext = mock( OnlineBackupContext.class );
-    final OnlineBackupRequiredArguments requiredArguments = mock( OnlineBackupRequiredArguments.class );
+    private final OnlineBackupContext onlineBackupContext = mock( OnlineBackupContext.class );
+    private final OnlineBackupRequiredArguments requiredArguments = mock( OnlineBackupRequiredArguments.class );
 
     // mock returns
-    private ProgressMonitorFactory progressMonitorFactory = mock( ProgressMonitorFactory.class );
-    private Path reportDir = mock( Path.class );
-    private ConsistencyCheckService.Result consistencyCheckResult = mock( ConsistencyCheckService.Result.class );
+    private final ProgressMonitorFactory progressMonitorFactory = mock( ProgressMonitorFactory.class );
+    private final Path reportDir = mock( Path.class );
+    private final ConsistencyCheckService.Result consistencyCheckResult = mock( ConsistencyCheckService.Result.class );
 
     @Before
     public void setup()
     {
+        when( outsideWorld.fileSystem() ).thenReturn( fileSystem );
         when( onlineBackupContext.getRequiredArguments() ).thenReturn( requiredArguments );
         when( requiredArguments.getReportDir() ).thenReturn( reportDir );
         subject = new BackupFlow( consistencyCheckService, outsideWorld, logProvider, progressMonitorFactory,

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/catchup/storecopy/RemoteStore.java

@@ -162,8 +162,8 @@ public void copy( AdvertisedSocketAddress from, StoreId expectedStoreId, File de
     private CatchupResult pullTransactions( AdvertisedSocketAddress from, StoreId expectedStoreId, File storeDir, long fromTxId, boolean asPartOfStoreCopy )
             throws IOException, StoreCopyFailedException
     {
-        try ( TransactionLogCatchUpWriter writer = transactionLogFactory.create( storeDir, fs, pageCache, logProvider,
-                fromTxId, asPartOfStoreCopy ) )
+        try ( TransactionLogCatchUpWriter writer = transactionLogFactory.create( storeDir, fs, pageCache, config,
+                logProvider, fromTxId, asPartOfStoreCopy ) )
         {
             log.info( "Pulling transactions from: %d", fromTxId );
 

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/catchup/tx/TransactionLogCatchUpFactory.java

@@ -24,13 +24,15 @@
 
 import org.neo4j.io.fs.FileSystemAbstraction;
 import org.neo4j.io.pagecache.PageCache;
+import org.neo4j.kernel.configuration.Config;
 import org.neo4j.logging.LogProvider;
 
 public class TransactionLogCatchUpFactory
 {
     public TransactionLogCatchUpWriter create( File storeDir, FileSystemAbstraction fs, PageCache pageCache,
-            LogProvider logProvider, long fromTxId, boolean asPartOfStoreCopy ) throws IOException
+            Config config, LogProvider logProvider, long fromTxId, boolean asPartOfStoreCopy ) throws IOException
     {
-        return new TransactionLogCatchUpWriter( storeDir, fs, pageCache, logProvider, fromTxId, asPartOfStoreCopy );
+        return new TransactionLogCatchUpWriter( storeDir, fs, pageCache, config, logProvider, fromTxId,
+                asPartOfStoreCopy );
     }
 }

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/catchup/tx/TransactionLogCatchUpWriter.java

@@ -24,6 +24,7 @@
 
 import org.neo4j.io.fs.FileSystemAbstraction;
 import org.neo4j.io.pagecache.PageCache;
+import org.neo4j.kernel.configuration.Config;
 import org.neo4j.kernel.impl.store.MetaDataStore;
 import org.neo4j.kernel.impl.transaction.CommittedTransactionRepresentation;
 import org.neo4j.kernel.impl.transaction.log.LogPosition;
@@ -52,14 +53,19 @@ public class TransactionLogCatchUpWriter implements TxPullResponseListener, Auto
     private long lastTxId = -1;
     private long expectedTxId;
 
-    TransactionLogCatchUpWriter( File storeDir, FileSystemAbstraction fs, PageCache pageCache,
+    TransactionLogCatchUpWriter( File storeDir, FileSystemAbstraction fs, PageCache pageCache, Config config,
             LogProvider logProvider, long fromTxId, boolean asPartOfStoreCopy ) throws IOException
     {
         this.pageCache = pageCache;
         this.log = logProvider.getLog( getClass() );
         this.asPartOfStoreCopy = asPartOfStoreCopy;
-        this.logFiles = LogFilesBuilder.activeFilesBuilder( storeDir, fs, pageCache )
-                .withLastCommittedTransactionIdSupplier( () -> fromTxId - 1 ).build();
+        LogFilesBuilder logFilesBuilder = LogFilesBuilder.activeFilesBuilder( storeDir, fs, pageCache )
+                .withLastCommittedTransactionIdSupplier( () -> fromTxId - 1 );
+        if ( !asPartOfStoreCopy )
+        {
+            logFilesBuilder.withConfig( config );
+        }
+        this.logFiles = logFilesBuilder.build();
         this.lifespan.add( logFiles );
         this.writer = new TransactionLogWriter( new LogEntryWriter( logFiles.getLogFile().getWriter() ) );
         this.storeDir = storeDir;

enterprise/causal-clustering/src/test/java/org/neo4j/causalclustering/catchup/storecopy/RemoteStoreTest.java

@@ -135,9 +135,8 @@ public void shouldCloseDownTxLogWriterIfTxStreamingFails() throws Exception
     private TransactionLogCatchUpFactory factory( TransactionLogCatchUpWriter writer ) throws IOException
     {
         TransactionLogCatchUpFactory factory = mock( TransactionLogCatchUpFactory.class );
-        when( factory.create( isNull(), any( FileSystemAbstraction.class ),
-                isNull(), any( LogProvider.class ), anyLong(), anyBoolean() ) )
-                .thenReturn( writer );
+        when( factory.create( isNull(), any( FileSystemAbstraction.class ), isNull(), any( Config.class ),
+                any( LogProvider.class ), anyLong(), anyBoolean() ) ).thenReturn( writer );
         return factory;
     }
 }

enterprise/causal-clustering/src/test/java/org/neo4j/causalclustering/catchup/tx/TransactionLogCatchUpWriterTest.java

@@ -22,9 +22,13 @@
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
 
 import org.neo4j.causalclustering.identity.StoreId;
 import org.neo4j.graphdb.factory.GraphDatabaseSettings;
@@ -66,24 +70,31 @@
 import static org.junit.Assert.assertTrue;
 import static org.neo4j.kernel.impl.transaction.command.Commands.createNode;
 
+@RunWith( Parameterized.class )
 public class TransactionLogCatchUpWriterTest
 {
     @Rule
     public final TestDirectory dir = TestDirectory.testDirectory();
-
     @Rule
     public final DefaultFileSystemRule fsRule = new DefaultFileSystemRule();
-
     @Rule
     public final PageCacheRule pageCacheRule = new PageCacheRule();
-
     @Rule
     public NeoStoreDataSourceRule dsRule = new NeoStoreDataSourceRule();
 
+    @Parameterized.Parameter
+    public boolean partOfStoreCopy;
+
     private PageCache pageCache;
     private FileSystemAbstraction fs;
     private File storeDir;
 
+    @Parameterized.Parameters
+    public static List<Boolean> partOfStoreCopy()
+    {
+        return Arrays.asList( Boolean.TRUE, Boolean.FALSE );
+    }
+
     @Before
     public void setup() throws IOException
     {
@@ -101,7 +112,8 @@ public void shouldCreateTransactionLogWithCheckpoint() throws Exception
     @Test
     public void createTransactionLogWithCheckpointInCustomLocation() throws IOException
     {
-        createTransactionLogWithCheckpoint( Config.defaults( GraphDatabaseSettings.logical_logs_location, "custom-tx-logs") );
+        createTransactionLogWithCheckpoint( Config.defaults( GraphDatabaseSettings.logical_logs_location,
+                "custom-tx-logs") );
     }
 
     private void createTransactionLogWithCheckpoint( Config config ) throws IOException
@@ -111,8 +123,8 @@ private void createTransactionLogWithCheckpoint( Config config ) throws IOExcept
         int fromTxId = 37;
         int endTxId = fromTxId + 5;
 
-        TransactionLogCatchUpWriter catchUpWriter = new TransactionLogCatchUpWriter( storeDir, fs, pageCache,
-                NullLogProvider.getInstance(), fromTxId, true );
+        TransactionLogCatchUpWriter catchUpWriter = new TransactionLogCatchUpWriter( storeDir, fs, pageCache, config,
+                NullLogProvider.getInstance(), fromTxId, partOfStoreCopy );
 
         // when
         for ( int i = fromTxId; i <= endTxId; i++ )
@@ -123,28 +135,35 @@ private void createTransactionLogWithCheckpoint( Config config ) throws IOExcept
         catchUpWriter.close();
 
         // then
-        verifyTransactionsInLog( fromTxId, endTxId, config );
-        verifyCheckpointInLog( config ); // necessary for recovery
+        LogFilesBuilder logFilesBuilder = LogFilesBuilder.activeFilesBuilder( storeDir, fs, pageCache );
+        if ( !partOfStoreCopy )
+        {
+            logFilesBuilder.withConfig( config );
+        }
+        LogFiles logFiles = logFilesBuilder.build();
+
+        verifyTransactionsInLog( logFiles, fromTxId, endTxId );
+        if ( partOfStoreCopy )
+        {
+            verifyCheckpointInLog( logFiles );
+        }
     }
 
-    private void verifyCheckpointInLog( Config config ) throws IOException
+    private void verifyCheckpointInLog( LogFiles logFiles )
     {
         LogEntryReader<ReadableClosablePositionAwareChannel> logEntryReader = new VersionAwareLogEntryReader<>(
                 new RecordStorageCommandReaderFactory(), InvalidLogEntryHandler.STRICT );
-        LogFiles logFiles = LogFilesBuilder.activeFilesBuilder( storeDir, fs, pageCache )
-                .withConfig( config )
-                .withLogEntryReader( logEntryReader ).build();
         final LogTailScanner logTailScanner = new LogTailScanner( logFiles, logEntryReader, new Monitors() );
 
         LogTailInformation tailInformation = logTailScanner.getTailInformation();
         assertNotNull( tailInformation.lastCheckPoint );
         assertTrue( tailInformation.commitsAfterLastCheckpoint() );
     }
 
-    private void verifyTransactionsInLog( long fromTxId, long endTxId, Config config ) throws IOException
+    private void verifyTransactionsInLog( LogFiles logFiles, long fromTxId, long endTxId ) throws
+            IOException
     {
         long expectedTxId = fromTxId;
-        LogFiles logFiles = LogFilesBuilder.activeFilesBuilder( storeDir, fs, pageCache ).withConfig( config ).build();
         LogVersionedStoreChannel versionedStoreChannel = logFiles.openForVersion( 0 );
         try ( ReadableLogChannel channel =
                       new ReadAheadLogChannel( versionedStoreChannel, LogVersionBridge.NO_MORE_CHANNELS, 1024 ) )