[
https://issues.apache.org/jira/browse/HIVE-22865?focusedWorklogId=393256&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-393256
]
ASF GitHub Bot logged work on HIVE-22865:
-----------------------------------------
Author: ASF GitHub Bot
Created on: 26/Feb/20 08:49
Start Date: 26/Feb/20 08:49
Worklog Time Spent: 10m
Work Description: aasha commented on 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:
us...@infra.apache.org
Issue Time Tracking
-------------------
Worklog Id: (was: 393256)
Time Spent: 1.5h (was: 1h 20m)
> Include data in replication staging directory
> ---------------------------------------------
>
> Key: HIVE-22865
> URL: https://issues.apache.org/jira/browse/HIVE-22865
> Project: Hive
> Issue Type: Task
> Reporter: PRAVIN KUMAR SINHA
> Assignee: PRAVIN KUMAR SINHA
> Priority: Major
> Labels: pull-request-available
> Attachments: HIVE-22865.1.patch, HIVE-22865.2.patch,
> HIVE-22865.3.patch
>
> Time Spent: 1.5h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
