aasha commented on a change in pull request #911: HIVE-22865 Include data in replication staging directory URL: https://github.com/apache/hive/pull/911#discussion_r384345535
########## File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosWithCopyData.java ########## @@ -0,0 +1,422 @@ +/* + * 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.commons.io.FileUtils; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.cli.CliSessionState; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.HiveMetaStoreClient; +import org.apache.hadoop.hive.metastore.MetaStoreTestUtils; +import org.apache.hadoop.hive.metastore.PersistenceManagerProvider; +import org.apache.hadoop.hive.metastore.api.Database; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.hadoop.hive.metastore.messaging.json.gzip.GzipJSONMessageEncoder; +import org.apache.hadoop.hive.ql.DriverFactory; +import org.apache.hadoop.hive.ql.IDriver; +import org.apache.hadoop.hive.ql.exec.Task; +import org.apache.hadoop.hive.ql.exec.repl.ReplDumpWork; +import org.apache.hadoop.hive.ql.metadata.Hive; +import org.apache.hadoop.hive.ql.processors.CommandProcessorException; +import org.apache.hadoop.hive.ql.session.SessionState; +import org.apache.hadoop.hive.shims.Utils; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TestName; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.hadoop.hive.metastore.ReplChangeManager.SOURCE_OF_REPLICATION; +import static org.junit.Assert.assertEquals;; + +public class TestReplicationScenariosWithCopyData { + + @Rule + public final TestName testName = new TestName(); + + private final static String DBNOTIF_LISTENER_CLASSNAME = + "org.apache.hive.hcatalog.listener.DbNotificationListener"; + // FIXME : replace with hive copy once that is copied + private final static String tid = + TestReplicationScenariosWithCopyData.class.getCanonicalName().toLowerCase().replace('.','_') + "_" + System.currentTimeMillis(); + private final static String TEST_PATH = + System.getProperty("test.warehouse.dir", "/tmp") + Path.SEPARATOR + tid; + + static HiveConf hconf; + static HiveMetaStoreClient metaStoreClient; + private static IDriver driver; + private static String proxySettingName; + private static HiveConf hconfMirror; + private static IDriver driverMirror; + private static HiveMetaStoreClient metaStoreClientMirror; + private static boolean isMigrationTest; + + // Make sure we skip backward-compat checking for those tests that don't generate events + + protected static final Logger LOG = LoggerFactory.getLogger(TestReplicationScenariosWithCopyData.class); + private ArrayList<String> lastResults; + + private final boolean VERIFY_SETUP_STEPS = false; + // if verifySetup is set to true, all the test setup we do will perform additional + // verifications as well, which is useful to verify that our setup occurred + // correctly when developing and debugging tests. These verifications, however + // do not test any new functionality for replication, and thus, are not relevant + // for testing replication itself. For steady state, we want this to be false. + + @BeforeClass + public static void setUpBeforeClass() throws Exception { + HashMap<String, String> overrideProperties = new HashMap<>(); + overrideProperties.put(MetastoreConf.ConfVars.EVENT_MESSAGE_FACTORY.getHiveName(), + GzipJSONMessageEncoder.class.getCanonicalName()); + internalBeforeClassSetup(overrideProperties, false); + } + + static void internalBeforeClassSetup(Map<String, String> additionalProperties, boolean forMigration) + throws Exception { + hconf = new HiveConf(TestReplicationScenariosWithCopyData.class); + String metastoreUri = System.getProperty("test."+MetastoreConf.ConfVars.THRIFT_URIS.getHiveName()); + if (metastoreUri != null) { + hconf.set(MetastoreConf.ConfVars.THRIFT_URIS.getHiveName(), metastoreUri); + return; + } + isMigrationTest = forMigration; + + hconf.set(MetastoreConf.ConfVars.TRANSACTIONAL_EVENT_LISTENERS.getHiveName(), + DBNOTIF_LISTENER_CLASSNAME); // turn on db notification listener on metastore + hconf.setBoolVar(HiveConf.ConfVars.REPLCMENABLED, true); + hconf.setBoolVar(HiveConf.ConfVars.FIRE_EVENTS_FOR_DML, true); + hconf.setVar(HiveConf.ConfVars.REPLCMDIR, TEST_PATH + "/cmroot/"); + proxySettingName = "hadoop.proxyuser." + Utils.getUGI().getShortUserName() + ".hosts"; + hconf.set(proxySettingName, "*"); + hconf.setVar(HiveConf.ConfVars.REPLDIR,TEST_PATH + "/hrepl/"); + hconf.set(MetastoreConf.ConfVars.THRIFT_CONNECTION_RETRIES.getHiveName(), "3"); + hconf.set(HiveConf.ConfVars.PREEXECHOOKS.varname, ""); + hconf.set(HiveConf.ConfVars.POSTEXECHOOKS.varname, ""); + hconf.set(HiveConf.ConfVars.HIVE_IN_TEST_REPL.varname, "true"); + hconf.setBoolVar(HiveConf.ConfVars.HIVE_IN_TEST, true); + hconf.set(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY.varname, "false"); + hconf.set(HiveConf.ConfVars.HIVE_TXN_MANAGER.varname, + "org.apache.hadoop.hive.ql.lockmgr.DummyTxnManager"); + hconf.set(HiveConf.ConfVars.METASTORE_RAW_STORE_IMPL.varname, + "org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore"); + hconf.setBoolVar(HiveConf.ConfVars.HIVEOPTIMIZEMETADATAQUERIES, true); + hconf.setBoolVar(HiveConf.ConfVars.HIVESTATSAUTOGATHER, true); + hconf.setBoolVar(HiveConf.ConfVars.HIVE_STATS_RELIABLE, true); + System.setProperty(HiveConf.ConfVars.PREEXECHOOKS.varname, " "); + System.setProperty(HiveConf.ConfVars.POSTEXECHOOKS.varname, " "); + + additionalProperties.forEach((key, value) -> { + hconf.set(key, value); + }); + + MetaStoreTestUtils.startMetaStoreWithRetry(hconf); + // re set the WAREHOUSE property to the test dir, as the previous command added a random port to it + hconf.set(MetastoreConf.ConfVars.WAREHOUSE.getVarname(), System.getProperty("test.warehouse.dir", "/tmp")); + + Path testPath = new Path(TEST_PATH); + FileSystem fs = FileSystem.get(testPath.toUri(),hconf); + fs.mkdirs(testPath); + driver = DriverFactory.newDriver(hconf); + SessionState.start(new CliSessionState(hconf)); + metaStoreClient = new HiveMetaStoreClient(hconf); + + FileUtils.deleteDirectory(new File("metastore_db2")); + HiveConf hconfMirrorServer = new HiveConf(); + hconfMirrorServer.set(HiveConf.ConfVars.METASTORECONNECTURLKEY.varname, "jdbc:derby:;databaseName=metastore_db2;create=true"); + MetaStoreTestUtils.startMetaStoreWithRetry(hconfMirrorServer); + hconfMirror = new HiveConf(hconf); + String thriftUri = MetastoreConf.getVar(hconfMirrorServer, MetastoreConf.ConfVars.THRIFT_URIS); + MetastoreConf.setVar(hconfMirror, MetastoreConf.ConfVars.THRIFT_URIS, thriftUri); + + if (forMigration) { + hconfMirror.setBoolVar(HiveConf.ConfVars.HIVE_STRICT_MANAGED_TABLES, true); + hconfMirror.setBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY, true); + hconfMirror.set(HiveConf.ConfVars.HIVE_TXN_MANAGER.varname, + "org.apache.hadoop.hive.ql.lockmgr.DbTxnManager"); + } + driverMirror = DriverFactory.newDriver(hconfMirror); + metaStoreClientMirror = new HiveMetaStoreClient(hconfMirror); + + PersistenceManagerProvider.setTwoMetastoreTesting(true); + } + + @AfterClass + public static void tearDownAfterClass(){ + } + + @Before + public void setUp(){ + // before each test + SessionState.get().setCurrentDatabase("default"); + } + + @After + public void tearDown(){ + // after each test + } + + private static int next = 0; + private synchronized void advanceDumpDir() { + next++; + ReplDumpWork.injectNextDumpDirForTest(String.valueOf(next)); + } + + static class Tuple { + final String dumpLocation; + final String lastReplId; + + Tuple(String dumpLocation, String lastReplId) { + this.dumpLocation = dumpLocation; + this.lastReplId = lastReplId; + } + } + + private Tuple bootstrapLoadAndVerify(String dbName, String replDbName) throws IOException { + return incrementalLoadAndVerify(dbName, null, replDbName); + } + + private Tuple incrementalLoadAndVerify(String dbName, String fromReplId, String replDbName) throws IOException { + Tuple dump = replDumpDb(dbName, fromReplId, null, null); + loadAndVerify(replDbName, dump.dumpLocation, dump.lastReplId); + return dump; + } + + private Tuple replDumpDb(String dbName, String fromReplID, String toReplID, String limit) throws IOException { + advanceDumpDir(); + String dumpCmd = "REPL DUMP " + dbName; + if (null != fromReplID) { + dumpCmd = dumpCmd + " FROM " + fromReplID; + } + if (null != toReplID) { + dumpCmd = dumpCmd + " TO " + toReplID; + } + if (null != limit) { + dumpCmd = dumpCmd + " LIMIT " + limit; + } + run(dumpCmd, driver); + String dumpLocation = getResult(0, 0, driver); + String lastReplId = getResult(0, 1, true, driver); + LOG.info("Dumped to {} with id {} for command: {}", dumpLocation, lastReplId, dumpCmd); + return new Tuple(dumpLocation, lastReplId); + } + + private void loadAndVerify(String replDbName, String dumpLocation, String lastReplId) throws IOException { + run("REPL LOAD " + replDbName + " FROM '" + dumpLocation + "'", driverMirror); + verifyRun("REPL STATUS " + replDbName, lastReplId, driverMirror); + return; + } + + private abstract class checkTaskPresent { + public boolean hasTask(Task rootTask) { + if (rootTask == null) { + return false; + } + if (validate(rootTask)) { + return true; + } + List<Task<?>> childTasks = rootTask.getChildTasks(); + if (childTasks == null) { + return false; + } + for (Task<?> childTask : childTasks) { + if (hasTask(childTask)) { + return true; + } + } + return false; + } + + public abstract boolean validate(Task task); + } + + @Test + public void testIncrementalLoadWithDumpData() throws IOException { Review comment: Are these new test scenarios? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
