itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTablesBootstrap.java - hive - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.hadoop.hive.ql.parse;

 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
 import org.apache.hadoop.hive.metastore.messaging.json.gzip.GzipJSONMessageEncoder;
 import org.apache.hadoop.hive.metastore.txn.TxnStore;
 import org.apache.hadoop.hive.metastore.txn.TxnUtils;
 import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore;
 import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore.CallerArguments;
 import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore.BehaviourInjection;
 import org.apache.hadoop.hive.ql.exec.repl.util.ReplUtils;
 import org.apache.hadoop.hive.ql.ErrorMsg;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;

 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.BeforeClass;

 import javax.annotation.Nullable;
 import java.io.File;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 /**
  * TestReplicationScenariosAcidTables - test bootstrap of ACID tables during an incremental.
  */
 public class TestReplicationScenariosAcidTablesBootstrap
         extends BaseReplicationScenariosAcidTables {

   private static List<String> dumpWithoutAcidClause = Collections.singletonList(
           "'" + ReplUtils.REPL_DUMP_INCLUDE_ACID_TABLES + "'='false'");
   private static List<String> dumpWithAcidBootstrapClause = Arrays.asList(
           "'" + ReplUtils.REPL_DUMP_INCLUDE_ACID_TABLES + "'='true'",
           "'" + HiveConf.ConfVars.REPL_BOOTSTRAP_ACID_TABLES + "'='true'");
   private static List<String> dumpWithAcidClause = Arrays.asList(
           "'" + ReplUtils.REPL_DUMP_INCLUDE_ACID_TABLES + "'='true'");

   @BeforeClass
   public static void classLevelSetup() throws Exception {
     Map<String, String> overrides = new HashMap<>();
     overrides.put(MetastoreConf.ConfVars.EVENT_MESSAGE_FACTORY.getHiveName(),
         GzipJSONMessageEncoder.class.getCanonicalName());

     internalBeforeClassSetup(overrides, TestReplicationScenariosAcidTablesBootstrap.class);
   }

   @Test
   public void testAcidTablesBootstrapDuringIncremental() throws Throwable {
     // Take a bootstrap dump without acid tables
     WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName,
             dumpWithoutAcidClause);
     LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
     replica.load(replicatedDbName, primaryDbName);
     verifyLoadExecution(replicatedDbName, bootstrapDump.lastReplicationId, false);

     // Take a incremental dump with acid table bootstrap
     prepareIncAcidData(primaryDbName);
     prepareIncNonAcidData(primaryDbName);
     LOG.info(testName.getMethodName() + ": incremental dump and load dump with acid table bootstrap.");
     WarehouseInstance.Tuple incrementalDump = primary.run("use " + primaryDbName)
             .dump(primaryDbName, dumpWithAcidBootstrapClause);
     replica.load(replicatedDbName, primaryDbName);
     verifyIncLoad(replicatedDbName, incrementalDump.lastReplicationId);
     // Ckpt should be set on bootstrapped tables.
     String hiveDumpLocation = incrementalDump.dumpLocation + File.separator + ReplUtils.REPL_HIVE_BASE_DIR;
     replica.verifyIfCkptSetForTables(replicatedDbName, acidTableNames, hiveDumpLocation);

     // Take a second normal incremental dump after Acid table boostrap
     prepareInc2AcidData(primaryDbName, primary.hiveConf);
     prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
     LOG.info(testName.getMethodName()
              + ": second incremental dump and load dump after incremental with acid table " +
             "bootstrap.");
     WarehouseInstance.Tuple inc2Dump = primary.run("use " + primaryDbName)
             .dump(primaryDbName);
     replica.load(replicatedDbName, primaryDbName);
     verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
   }

   @Test
   public void testRetryAcidTablesBootstrapFromDifferentDump() throws Throwable {
     WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName,
             dumpWithoutAcidClause);
     LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
     replica.load(replicatedDbName, primaryDbName);
     verifyLoadExecution(replicatedDbName, bootstrapDump.lastReplicationId, false);

     prepareIncAcidData(primaryDbName);
     prepareIncNonAcidData(primaryDbName);
     LOG.info(testName.getMethodName() + ": first incremental dump with acid table bootstrap.");
     WarehouseInstance.Tuple incDump = primary.run("use " + primaryDbName)
             .dump(primaryDbName, dumpWithAcidBootstrapClause);

     // Fail setting ckpt property for table t5 but success for earlier tables
     BehaviourInjection<CallerArguments, Boolean> callerVerifier
             = new BehaviourInjection<CallerArguments, Boolean>() {
       @Nullable
       @Override
       public Boolean apply(@Nullable CallerArguments args) {
         if (args.tblName.equalsIgnoreCase("t5") && args.dbName.equalsIgnoreCase(replicatedDbName)) {
           injectionPathCalled = true;
           LOG.warn("Verifier - DB : " + args.dbName + " TABLE : " + args.tblName);
           return false;
         }
         return true;
       }
     };

     // Fail repl load before the ckpt property is set for t4 and after it is set for t2.
     // In the retry, these half baked tables should be dropped and bootstrap should be successful.
     InjectableBehaviourObjectStore.setAlterTableModifier(callerVerifier);
     try {
       LOG.info(testName.getMethodName()
               + ": loading first incremental dump with acid table bootstrap (will fail)");
       replica.loadFailure(replicatedDbName, primaryDbName);
       callerVerifier.assertInjectionsPerformed(true, false);
     } finally {
       InjectableBehaviourObjectStore.resetAlterTableModifier();
     }
     Path baseDumpDir = new Path(primary.hiveConf.getVar(HiveConf.ConfVars.REPLDIR));
     Path nonRecoverablePath = TestReplicationScenarios.getNonRecoverablePath(baseDumpDir, primaryDbName, primary.hiveConf);
     if(nonRecoverablePath != null){
       baseDumpDir.getFileSystem(primary.hiveConf).delete(nonRecoverablePath, true);
     }
     //Load again should succeed as checkpointing is in place
     replica.load(replicatedDbName, primaryDbName);
     verifyIncLoad(replicatedDbName, incDump.lastReplicationId);

     prepareInc2AcidData(primaryDbName, primary.hiveConf);
     prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
     LOG.info(testName.getMethodName() + ": second incremental dump with acid table bootstrap");
     WarehouseInstance.Tuple inc2Dump = primary.run("use " + primaryDbName)
             .dump(primaryDbName, dumpWithAcidClause);
     LOG.info(testName.getMethodName()
             + ": trying to load second incremental dump (with acid bootstrap) again."
             + " Should succeed.");
     replica.load(replicatedDbName, primaryDbName);
     verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
   }

   @Test
   public void retryIncBootstrapAcidFromDifferentDumpWithoutCleanTablesConfig() throws Throwable {
     prepareDataAndDump(primaryDbName,
             dumpWithoutAcidClause);
     replica.load(replicatedDbName, primaryDbName);

     prepareIncAcidData(primaryDbName);
     prepareIncNonAcidData(primaryDbName);
     primary.run("use " + primaryDbName)
             .dump(primaryDbName, dumpWithAcidBootstrapClause);
     replica.load(replicatedDbName, primaryDbName);
     primary.run("use " + primaryDbName)
             .dump(primaryDbName, dumpWithAcidBootstrapClause);
     // Re-bootstrapping from different bootstrap dump without clean tables config should fail.
     replica.loadFailure(replicatedDbName, primaryDbName, Collections.emptyList(),
             ErrorMsg.REPL_BOOTSTRAP_LOAD_PATH_NOT_VALID.getErrorCode());
   }

   @Test
   public void testAcidTablesBootstrapDuringIncrementalWithOpenTxnsTimeout() throws Throwable {
     // Take a dump without ACID tables
     WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName,
                                                                 dumpWithoutAcidClause);
     LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
     replica.load(replicatedDbName, primaryDbName);

     // Open concurrent transactions, create data for incremental and take an incremental dump
     // with ACID table bootstrap.
     int numTxns = 5;
     HiveConf primaryConf = primary.getConf();
     TxnStore txnHandler = TxnUtils.getTxnStore(primary.getConf());
     // Open 5 txns
     List<Long> txns = openTxns(numTxns, txnHandler, primaryConf);
     prepareIncNonAcidData(primaryDbName);
     prepareIncAcidData(primaryDbName);
     // Allocate write ids for tables t1 and t2 for all txns
     // t1=5+2(insert) and t2=5+6(insert, alter add column), now alter also creates a transaction
     Map<String, Long> tables = new HashMap<>();
     tables.put("t1", numTxns+2L);
     tables.put("t2", numTxns+6L);
     List<Long> lockIds = allocateWriteIdsForTablesAndAcquireLocks(primaryDbName, tables, txnHandler, txns, primaryConf);

     // Bootstrap dump with open txn timeout as 1s.
     List<String> withConfigs = new LinkedList<>(dumpWithAcidBootstrapClause);
             withConfigs.add("'hive.repl.bootstrap.dump.open.txn.timeout'='1s'");
     WarehouseInstance.Tuple incDump = primary
             .run("use " + primaryDbName)
             .dump(primaryDbName, withConfigs);

     // After bootstrap dump, all the opened txns should be aborted. Verify it.
     verifyAllOpenTxnsAborted(txns, primaryConf);
     releaseLocks(txnHandler, lockIds);
     verifyNextId(tables, primaryDbName, primaryConf);

     // Incremental load with ACID bootstrap should also replicate the aborted write ids on
     // tables t1 and t2
     HiveConf replicaConf = replica.getConf();
     LOG.info(testName.getMethodName() + ": loading incremental dump with ACID bootstrap.");
     replica.load(replicatedDbName, primaryDbName);
     verifyIncLoad(replicatedDbName, incDump.lastReplicationId);
     // Verify if HWM is properly set after REPL LOAD
     verifyNextId(tables, replicatedDbName, replicaConf);

     // Verify if all the aborted write ids are replicated to the replicated DB
     for(Map.Entry<String, Long> entry : tables.entrySet()) {
       entry.setValue((long) numTxns);
     }
     verifyWriteIdsForTables(tables, replicaConf, replicatedDbName);

     // Verify if entries added in COMPACTION_QUEUE for each table/partition
     // t1-> 1 entry and t2-> 2 entries (1 per partition)
     tables.clear();
     tables.put("t1", 1L);
     tables.put("t2", 4L);
     verifyCompactionQueue(tables, replicatedDbName, replicaConf);
   }

   @Test
   public void testBootstrapAcidTablesDuringIncrementalWithConcurrentWrites() throws Throwable {
     // Dump and load bootstrap without ACID tables.
     WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName,
                                                                 dumpWithoutAcidClause);
     LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
     replica.load(replicatedDbName, primaryDbName);

     // Create incremental data for incremental load with bootstrap of ACID
     prepareIncNonAcidData(primaryDbName);
     prepareIncAcidData(primaryDbName);
     // Perform concurrent writes. Bootstrap won't see the written data but the subsequent
     // incremental repl should see it. We can not inject callerVerifier since an incremental dump
     // would not cause an ALTER DATABASE event. Instead we piggy back on
     // getCurrentNotificationEventId() which is anyway required for a bootstrap.
     BehaviourInjection<CurrentNotificationEventId, CurrentNotificationEventId> callerInjectedBehavior
             = new BehaviourInjection<CurrentNotificationEventId, CurrentNotificationEventId>() {
       @Nullable
       @Override
       public CurrentNotificationEventId apply(@Nullable CurrentNotificationEventId input) {
         if (injectionPathCalled) {
           nonInjectedPathCalled = true;
         } else {
           // Do some writes through concurrent thread
           injectionPathCalled = true;
           Thread t = new Thread(new Runnable() {
             @Override
             public void run() {
               LOG.info("Entered new thread");
               try {
                 prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
                 prepareInc2AcidData(primaryDbName, primary.hiveConf);
               } catch (Throwable t) {
                 Assert.assertNull(t);
               }
               LOG.info("Exit new thread success");
             }
           });
           t.start();
           LOG.info("Created new thread {}", t.getName());
           try {
             t.join();
           } catch (InterruptedException e) {
             throw new RuntimeException(e);
           }
         }
         return input;
       }
     };

     InjectableBehaviourObjectStore.setGetCurrentNotificationEventIdBehaviour(callerInjectedBehavior);
     WarehouseInstance.Tuple incDump = null;
     try {
       incDump = primary.dump(primaryDbName, dumpWithAcidBootstrapClause);
       callerInjectedBehavior.assertInjectionsPerformed(true, true);
     } finally {
       // reset the behaviour
       InjectableBehaviourObjectStore.resetGetCurrentNotificationEventIdBehaviour();
     }

     // While bootstrapping ACID tables it has taken snapshot before concurrent thread performed
     // write. So concurrent writes won't be dumped.
     LOG.info(testName.getMethodName() +
             ": loading incremental dump containing bootstrapped ACID tables.");
     replica.load(replicatedDbName, primaryDbName);
     verifyIncLoad(replicatedDbName, incDump.lastReplicationId);

     // Next Incremental should include the concurrent writes
     LOG.info(testName.getMethodName() +
             ": dumping second normal incremental dump from event id = " + incDump.lastReplicationId);
     WarehouseInstance.Tuple inc2Dump = primary.dump(primaryDbName);
     LOG.info(testName.getMethodName() +
             ": loading second normal incremental dump from event id = " + incDump.lastReplicationId);
     replica.load(replicatedDbName, primaryDbName);
     verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.hadoop.hive.ql.parse;

	import org.apache.hadoop.hive.conf.HiveConf;
	import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
	import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
	import org.apache.hadoop.hive.metastore.messaging.json.gzip.GzipJSONMessageEncoder;
	import org.apache.hadoop.hive.metastore.txn.TxnStore;
	import org.apache.hadoop.hive.metastore.txn.TxnUtils;
	import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore;
	import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore.CallerArguments;
	import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore.BehaviourInjection;
	import org.apache.hadoop.hive.ql.exec.repl.util.ReplUtils;
	import org.apache.hadoop.hive.ql.ErrorMsg;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.Path;

	import org.junit.Assert;
	import org.junit.Test;
	import org.junit.BeforeClass;

	import javax.annotation.Nullable;
	import java.io.File;
	import java.util.Arrays;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.LinkedList;
	import java.util.List;
	import java.util.Map;
	/**
	* TestReplicationScenariosAcidTables - test bootstrap of ACID tables during an incremental.
	*/
	public class TestReplicationScenariosAcidTablesBootstrap
	extends BaseReplicationScenariosAcidTables {

	private static List<String> dumpWithoutAcidClause = Collections.singletonList(
	"'" + ReplUtils.REPL_DUMP_INCLUDE_ACID_TABLES + "'='false'");
	private static List<String> dumpWithAcidBootstrapClause = Arrays.asList(
	"'" + ReplUtils.REPL_DUMP_INCLUDE_ACID_TABLES + "'='true'",
	"'" + HiveConf.ConfVars.REPL_BOOTSTRAP_ACID_TABLES + "'='true'");
	private static List<String> dumpWithAcidClause = Arrays.asList(
	"'" + ReplUtils.REPL_DUMP_INCLUDE_ACID_TABLES + "'='true'");

	@BeforeClass
	public static void classLevelSetup() throws Exception {
	Map<String, String> overrides = new HashMap<>();
	overrides.put(MetastoreConf.ConfVars.EVENT_MESSAGE_FACTORY.getHiveName(),
	GzipJSONMessageEncoder.class.getCanonicalName());

	internalBeforeClassSetup(overrides, TestReplicationScenariosAcidTablesBootstrap.class);
	}

	@Test
	public void testAcidTablesBootstrapDuringIncremental() throws Throwable {
	// Take a bootstrap dump without acid tables
	WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName,
	dumpWithoutAcidClause);
	LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
	replica.load(replicatedDbName, primaryDbName);
	verifyLoadExecution(replicatedDbName, bootstrapDump.lastReplicationId, false);

	// Take a incremental dump with acid table bootstrap
	prepareIncAcidData(primaryDbName);
	prepareIncNonAcidData(primaryDbName);
	LOG.info(testName.getMethodName() + ": incremental dump and load dump with acid table bootstrap.");
	WarehouseInstance.Tuple incrementalDump = primary.run("use " + primaryDbName)
	.dump(primaryDbName, dumpWithAcidBootstrapClause);
	replica.load(replicatedDbName, primaryDbName);
	verifyIncLoad(replicatedDbName, incrementalDump.lastReplicationId);
	// Ckpt should be set on bootstrapped tables.
	String hiveDumpLocation = incrementalDump.dumpLocation + File.separator + ReplUtils.REPL_HIVE_BASE_DIR;
	replica.verifyIfCkptSetForTables(replicatedDbName, acidTableNames, hiveDumpLocation);

	// Take a second normal incremental dump after Acid table boostrap
	prepareInc2AcidData(primaryDbName, primary.hiveConf);
	prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
	LOG.info(testName.getMethodName()
	+ ": second incremental dump and load dump after incremental with acid table " +
	"bootstrap.");
	WarehouseInstance.Tuple inc2Dump = primary.run("use " + primaryDbName)
	.dump(primaryDbName);
	replica.load(replicatedDbName, primaryDbName);
	verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
	}

	@Test
	public void testRetryAcidTablesBootstrapFromDifferentDump() throws Throwable {
	WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName,
	dumpWithoutAcidClause);
	LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
	replica.load(replicatedDbName, primaryDbName);
	verifyLoadExecution(replicatedDbName, bootstrapDump.lastReplicationId, false);

	prepareIncAcidData(primaryDbName);
	prepareIncNonAcidData(primaryDbName);
	LOG.info(testName.getMethodName() + ": first incremental dump with acid table bootstrap.");
	WarehouseInstance.Tuple incDump = primary.run("use " + primaryDbName)
	.dump(primaryDbName, dumpWithAcidBootstrapClause);

	// Fail setting ckpt property for table t5 but success for earlier tables
	BehaviourInjection<CallerArguments, Boolean> callerVerifier
	= new BehaviourInjection<CallerArguments, Boolean>() {
	@Nullable
	@Override
	public Boolean apply(@Nullable CallerArguments args) {
	if (args.tblName.equalsIgnoreCase("t5") && args.dbName.equalsIgnoreCase(replicatedDbName)) {
	injectionPathCalled = true;
	LOG.warn("Verifier - DB : " + args.dbName + " TABLE : " + args.tblName);
	return false;
	}
	return true;
	}
	};

	// Fail repl load before the ckpt property is set for t4 and after it is set for t2.
	// In the retry, these half baked tables should be dropped and bootstrap should be successful.
	InjectableBehaviourObjectStore.setAlterTableModifier(callerVerifier);
	try {
	LOG.info(testName.getMethodName()
	+ ": loading first incremental dump with acid table bootstrap (will fail)");
	replica.loadFailure(replicatedDbName, primaryDbName);
	callerVerifier.assertInjectionsPerformed(true, false);
	} finally {
	InjectableBehaviourObjectStore.resetAlterTableModifier();
	}
	Path baseDumpDir = new Path(primary.hiveConf.getVar(HiveConf.ConfVars.REPLDIR));
	Path nonRecoverablePath = TestReplicationScenarios.getNonRecoverablePath(baseDumpDir, primaryDbName, primary.hiveConf);
	if(nonRecoverablePath != null){
	baseDumpDir.getFileSystem(primary.hiveConf).delete(nonRecoverablePath, true);
	}
	//Load again should succeed as checkpointing is in place
	replica.load(replicatedDbName, primaryDbName);
	verifyIncLoad(replicatedDbName, incDump.lastReplicationId);

	prepareInc2AcidData(primaryDbName, primary.hiveConf);
	prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
	LOG.info(testName.getMethodName() + ": second incremental dump with acid table bootstrap");
	WarehouseInstance.Tuple inc2Dump = primary.run("use " + primaryDbName)
	.dump(primaryDbName, dumpWithAcidClause);
	LOG.info(testName.getMethodName()
	+ ": trying to load second incremental dump (with acid bootstrap) again."
	+ " Should succeed.");
	replica.load(replicatedDbName, primaryDbName);
	verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
	}

	@Test
	public void retryIncBootstrapAcidFromDifferentDumpWithoutCleanTablesConfig() throws Throwable {
	prepareDataAndDump(primaryDbName,
	dumpWithoutAcidClause);
	replica.load(replicatedDbName, primaryDbName);

	prepareIncAcidData(primaryDbName);
	prepareIncNonAcidData(primaryDbName);
	primary.run("use " + primaryDbName)
	.dump(primaryDbName, dumpWithAcidBootstrapClause);
	replica.load(replicatedDbName, primaryDbName);
	primary.run("use " + primaryDbName)
	.dump(primaryDbName, dumpWithAcidBootstrapClause);
	// Re-bootstrapping from different bootstrap dump without clean tables config should fail.
	replica.loadFailure(replicatedDbName, primaryDbName, Collections.emptyList(),
	ErrorMsg.REPL_BOOTSTRAP_LOAD_PATH_NOT_VALID.getErrorCode());
	}

	@Test
	public void testAcidTablesBootstrapDuringIncrementalWithOpenTxnsTimeout() throws Throwable {
	// Take a dump without ACID tables
	WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName,
	dumpWithoutAcidClause);
	LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
	replica.load(replicatedDbName, primaryDbName);

	// Open concurrent transactions, create data for incremental and take an incremental dump
	// with ACID table bootstrap.
	int numTxns = 5;
	HiveConf primaryConf = primary.getConf();
	TxnStore txnHandler = TxnUtils.getTxnStore(primary.getConf());
	// Open 5 txns
	List<Long> txns = openTxns(numTxns, txnHandler, primaryConf);
	prepareIncNonAcidData(primaryDbName);
	prepareIncAcidData(primaryDbName);
	// Allocate write ids for tables t1 and t2 for all txns
	// t1=5+2(insert) and t2=5+6(insert, alter add column), now alter also creates a transaction
	Map<String, Long> tables = new HashMap<>();
	tables.put("t1", numTxns+2L);
	tables.put("t2", numTxns+6L);
	List<Long> lockIds = allocateWriteIdsForTablesAndAcquireLocks(primaryDbName, tables, txnHandler, txns, primaryConf);

	// Bootstrap dump with open txn timeout as 1s.
	List<String> withConfigs = new LinkedList<>(dumpWithAcidBootstrapClause);
	withConfigs.add("'hive.repl.bootstrap.dump.open.txn.timeout'='1s'");
	WarehouseInstance.Tuple incDump = primary
	.run("use " + primaryDbName)
	.dump(primaryDbName, withConfigs);

	// After bootstrap dump, all the opened txns should be aborted. Verify it.
	verifyAllOpenTxnsAborted(txns, primaryConf);
	releaseLocks(txnHandler, lockIds);
	verifyNextId(tables, primaryDbName, primaryConf);

	// Incremental load with ACID bootstrap should also replicate the aborted write ids on
	// tables t1 and t2
	HiveConf replicaConf = replica.getConf();
	LOG.info(testName.getMethodName() + ": loading incremental dump with ACID bootstrap.");
	replica.load(replicatedDbName, primaryDbName);
	verifyIncLoad(replicatedDbName, incDump.lastReplicationId);
	// Verify if HWM is properly set after REPL LOAD
	verifyNextId(tables, replicatedDbName, replicaConf);

	// Verify if all the aborted write ids are replicated to the replicated DB
	for(Map.Entry<String, Long> entry : tables.entrySet()) {
	entry.setValue((long) numTxns);
	}
	verifyWriteIdsForTables(tables, replicaConf, replicatedDbName);

	// Verify if entries added in COMPACTION_QUEUE for each table/partition
	// t1-> 1 entry and t2-> 2 entries (1 per partition)
	tables.clear();
	tables.put("t1", 1L);
	tables.put("t2", 4L);
	verifyCompactionQueue(tables, replicatedDbName, replicaConf);
	}

	@Test
	public void testBootstrapAcidTablesDuringIncrementalWithConcurrentWrites() throws Throwable {
	// Dump and load bootstrap without ACID tables.
	WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName,
	dumpWithoutAcidClause);
	LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
	replica.load(replicatedDbName, primaryDbName);

	// Create incremental data for incremental load with bootstrap of ACID
	prepareIncNonAcidData(primaryDbName);
	prepareIncAcidData(primaryDbName);
	// Perform concurrent writes. Bootstrap won't see the written data but the subsequent
	// incremental repl should see it. We can not inject callerVerifier since an incremental dump
	// would not cause an ALTER DATABASE event. Instead we piggy back on
	// getCurrentNotificationEventId() which is anyway required for a bootstrap.
	BehaviourInjection<CurrentNotificationEventId, CurrentNotificationEventId> callerInjectedBehavior
	= new BehaviourInjection<CurrentNotificationEventId, CurrentNotificationEventId>() {
	@Nullable
	@Override
	public CurrentNotificationEventId apply(@Nullable CurrentNotificationEventId input) {
	if (injectionPathCalled) {
	nonInjectedPathCalled = true;
	} else {
	// Do some writes through concurrent thread
	injectionPathCalled = true;
	Thread t = new Thread(new Runnable() {
	@Override
	public void run() {
	LOG.info("Entered new thread");
	try {
	prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
	prepareInc2AcidData(primaryDbName, primary.hiveConf);
	} catch (Throwable t) {
	Assert.assertNull(t);
	}
	LOG.info("Exit new thread success");
	}
	});
	t.start();
	LOG.info("Created new thread {}", t.getName());
	try {
	t.join();
	} catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	}
	return input;
	}
	};

	InjectableBehaviourObjectStore.setGetCurrentNotificationEventIdBehaviour(callerInjectedBehavior);
	WarehouseInstance.Tuple incDump = null;
	try {
	incDump = primary.dump(primaryDbName, dumpWithAcidBootstrapClause);
	callerInjectedBehavior.assertInjectionsPerformed(true, true);
	} finally {
	// reset the behaviour
	InjectableBehaviourObjectStore.resetGetCurrentNotificationEventIdBehaviour();
	}

	// While bootstrapping ACID tables it has taken snapshot before concurrent thread performed
	// write. So concurrent writes won't be dumped.
	LOG.info(testName.getMethodName() +
	": loading incremental dump containing bootstrapped ACID tables.");
	replica.load(replicatedDbName, primaryDbName);
	verifyIncLoad(replicatedDbName, incDump.lastReplicationId);

	// Next Incremental should include the concurrent writes
	LOG.info(testName.getMethodName() +
	": dumping second normal incremental dump from event id = " + incDump.lastReplicationId);
	WarehouseInstance.Tuple inc2Dump = primary.dump(primaryDbName);
	LOG.info(testName.getMethodName() +
	": loading second normal incremental dump from event id = " + incDump.lastReplicationId);
	replica.load(replicatedDbName, primaryDbName);
	verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
	}
	}