Make store-copying (HA, Backup) work with custom record format config…

…urations

We require that all members of a cluster, and backup clients, are configured with the same record format.
However, there was a problem with store copying, which starts up a temporary embedded database to do recovery of the copied transaction logs.
This embedded database was not told what record format the database as a whole had ben configured with, and thus would fail to start (and fail to recover the copied database) if a non-default format had been configured.

enterprise/com/src/main/java/org/neo4j/com/storecopy/StoreCopyClient.java

@@ -57,6 +57,7 @@
 
 import static java.lang.Math.max;
 import static org.neo4j.helpers.Format.bytes;
+import static org.neo4j.kernel.impl.factory.GraphDatabaseFacadeFactory.Configuration.record_format;
 import static org.neo4j.kernel.impl.transaction.log.TransactionIdStore.BASE_TX_ID;
 import static org.neo4j.kernel.impl.transaction.log.entry.LogHeader.LOG_HEADER_SIZE;
 import static org.neo4j.kernel.impl.transaction.log.entry.LogHeaderWriter.writeLogHeader;
@@ -318,6 +319,7 @@ private GraphDatabaseService newTempDatabase( File tempStore )
                 .setUserLogProvider( NullLogProvider.getInstance() )
                 .newEmbeddedDatabaseBuilder( tempStore.getAbsoluteFile() )
                 .setConfig( "dbms.backup.enabled", Settings.FALSE )
+                .setConfig( record_format, config.get( record_format ) )
                 .setConfig( GraphDatabaseSettings.logs_directory, tempStore.getAbsolutePath() )
                 .setConfig( GraphDatabaseSettings.keep_logical_logs, Settings.TRUE )
                 .setConfig( GraphDatabaseSettings.allow_store_upgrade,

enterprise/com/src/test/java/org/neo4j/com/storecopy/StoreCopyClientTest.java

@@ -40,11 +40,11 @@
 import org.neo4j.kernel.NeoStoreDataSource;
 import org.neo4j.kernel.configuration.Config;
 import org.neo4j.kernel.extension.KernelExtensionFactory;
-import org.neo4j.kernel.impl.factory.GraphDatabaseFacadeFactory;
 import org.neo4j.kernel.impl.storageengine.impl.recordstorage.RecordStorageEngine;
 import org.neo4j.kernel.impl.store.MetaDataStore;
 import org.neo4j.kernel.impl.store.StoreId;
 import org.neo4j.kernel.impl.store.format.highlimit.HighLimit;
+import org.neo4j.kernel.impl.store.format.lowlimit.LowLimitV3_0;
 import org.neo4j.kernel.impl.transaction.log.LogicalTransactionStore;
 import org.neo4j.kernel.impl.transaction.log.TransactionIdStore;
 import org.neo4j.kernel.impl.transaction.log.checkpoint.CheckPointer;
@@ -56,13 +56,17 @@
 import org.neo4j.test.TargetDirectory;
 import org.neo4j.test.TestGraphDatabaseFactory;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.neo4j.graphdb.DynamicLabel.label;
+import static org.neo4j.helpers.collection.MapUtil.stringMap;
+import static org.neo4j.kernel.impl.factory.GraphDatabaseFacadeFactory.Configuration.record_format;
 
 public class StoreCopyClientTest
 {
@@ -136,6 +140,55 @@ public void finishRecoveringStore()
         verify( storeCopyRequest, times( 1 ) ).done();
     }
 
+    @Test
+    public void storeCopyClientMustRespectConfiguredRecordFormat() throws Exception
+    {
+        final File copyDir = new File( testDir.directory(), "copy" );
+        final File originalDir = new File( testDir.directory(), "original" );
+        PageCache pageCache = pageCacheRule.getPageCache( fs );
+        String unkownFormat = "unkown_format";
+        Config config = Config.empty().augment( stringMap( record_format.name(), unkownFormat ) );
+        StoreCopyClient copier = new StoreCopyClient(
+                copyDir, config, loadKernelExtensions(), NullLogProvider.getInstance(), fs, pageCache,
+                new StoreCopyClient.Monitor.Adapter(), false );
+
+        final GraphDatabaseAPI original = (GraphDatabaseAPI) startDatabase( originalDir );
+        StoreCopyClient.StoreCopyRequester storeCopyRequest = storeCopyRequest( originalDir, original );
+
+        // This should complain about the unkown format
+        try
+        {
+            copier.copyStore( storeCopyRequest, CancellationRequest.NEVER_CANCELLED );
+            fail( "copyStore should have failed with this format configuration" );
+        }
+        catch ( Exception e )
+        {
+            assertThat( e.getMessage(), containsString( unkownFormat ) );
+        }
+    }
+
+    @Test
+    public void storeCopyClientMustWorkWithLowLimitRecordFormat() throws Exception
+    {
+        final File copyDir = new File( testDir.directory(), "copy" );
+        final File originalDir = new File( testDir.directory(), "original" );
+        PageCache pageCache = pageCacheRule.getPageCache( fs );
+        Config config = Config.empty().augment( stringMap( record_format.name(), LowLimitV3_0.NAME ) );
+        StoreCopyClient copier = new StoreCopyClient(
+                copyDir, config, loadKernelExtensions(), NullLogProvider.getInstance(), fs, pageCache,
+                new StoreCopyClient.Monitor.Adapter(), false );
+
+        final GraphDatabaseAPI original = (GraphDatabaseAPI) startDatabase( originalDir, LowLimitV3_0.NAME );
+        StoreCopyClient.StoreCopyRequester storeCopyRequest = storeCopyRequest( originalDir, original );
+
+        copier.copyStore( storeCopyRequest, CancellationRequest.NEVER_CANCELLED );
+
+        GraphDatabaseService copy = startDatabase( copyDir, LowLimitV3_0.NAME );
+
+        copy.shutdown();
+        original.shutdown();
+    }
+
     @Test
     public void shouldEndUpWithAnEmptyStoreIfCancellationRequestIssuedJustBeforeRecoveryTakesPlace()
             throws Exception
@@ -247,10 +300,15 @@ public void shouldResetNeoStoreLastTransactionOffsetForNonForensicCopy() throws
     }
 
     private GraphDatabaseService startDatabase( File storeDir )
+    {
+        return startDatabase( storeDir, HighLimit.NAME );
+    }
+
+    private GraphDatabaseService startDatabase( File storeDir, String recordFormatName )
     {
         return new TestGraphDatabaseFactory()
                 .newEmbeddedDatabaseBuilder( storeDir )
-                .setConfig( GraphDatabaseFacadeFactory.Configuration.record_format, HighLimit.NAME )
+                .setConfig( record_format, recordFormatName )
                 .newGraphDatabase();
     }