/*
 * 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.metastore;

import java.io.IOException;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.nio.file.Files;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.ReplChangeManager.RecycleType;

import static org.apache.hadoop.hive.common.repl.ReplConst.SOURCE_OF_REPLICATION;

import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.SerDeInfo;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.serde2.columnar.LazyBinaryColumnarSerDe;
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.DefaultImpersonationProvider;
import org.apache.hadoop.security.authorize.ProxyUsers;
import org.apache.hadoop.util.StringUtils;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;

import com.google.common.collect.ImmutableMap;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

public class TestReplChangeManager {
  private static HiveMetaStoreClient client;
  private static HiveConf hiveConf;
  private static Warehouse warehouse;
  private static MiniDFSCluster m_dfs;
  private static String cmroot;
  private static FileSystem fs;

  private static HiveConf permhiveConf;
  private static Warehouse permWarehouse;
  private static MiniDFSCluster permDdfs;
  private static String permCmroot;

  @BeforeClass
  public static void setUp() throws Exception {
    internalSetUp();
    internalSetUpProvidePerm();
    try {
      client = new HiveMetaStoreClient(hiveConf);
    } catch (Throwable e) {
      System.err.println("Unable to open the metastore");
      System.err.println(StringUtils.stringifyException(e));
      throw e;
    }
  }

  private static void internalSetUpProvidePerm() throws Exception {
    Configuration configuration = new Configuration();
    configuration.set("dfs.permissions.enabled", "false");
    String noPermBaseDir = Files.createTempDirectory("noPerm").toFile().getAbsolutePath();
    configuration.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, noPermBaseDir);
    configuration.set("dfs.client.use.datanode.hostname", "true");
    permDdfs = new MiniDFSCluster.Builder(configuration).numDataNodes(2).format(true).build();
    permhiveConf = new HiveConf(TestReplChangeManager.class);
    permhiveConf.set(HiveConf.ConfVars.METASTOREWAREHOUSE.varname,
      "hdfs://" + permDdfs.getNameNode().getHostAndPort() + HiveConf.ConfVars.METASTOREWAREHOUSE.defaultStrVal);
    permhiveConf.setBoolean(HiveConf.ConfVars.REPLCMENABLED.varname, true);
    permCmroot = "hdfs://" + permDdfs.getNameNode().getHostAndPort() + "/cmroot";
    permhiveConf.set(HiveConf.ConfVars.REPLCMDIR.varname, permCmroot);
    permhiveConf.setInt(CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY, 60);
    permWarehouse = new Warehouse(permhiveConf);
  }

  private static void internalSetUp() throws Exception {
    m_dfs = new MiniDFSCluster.Builder(new Configuration()).numDataNodes(2).format(true).build();
    hiveConf = new HiveConf(TestReplChangeManager.class);
    hiveConf.set(HiveConf.ConfVars.METASTOREWAREHOUSE.varname,
      "hdfs://" + m_dfs.getNameNode().getHostAndPort() + HiveConf.ConfVars.METASTOREWAREHOUSE.defaultStrVal);
    hiveConf.setBoolean(HiveConf.ConfVars.REPLCMENABLED.varname, true);
    cmroot = "hdfs://" + m_dfs.getNameNode().getHostAndPort() + "/cmroot";
    hiveConf.set(HiveConf.ConfVars.REPLCMDIR.varname, cmroot);
    hiveConf.setInt(CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY, 60);
    warehouse = new Warehouse(hiveConf);
    fs = new Path(cmroot).getFileSystem(hiveConf);
  }

  @AfterClass
  public static void tearDown() {
    try {
      m_dfs.shutdown();
      permDdfs.shutdown();
      client.close();
    } catch (Throwable e) {
      System.err.println("Unable to close metastore");
      System.err.println(StringUtils.stringifyException(e));
      throw e;
    }
  }

  private Partition createPartition(String dbName, String tblName,
      List<FieldSchema> columns, List<String> partVals, SerDeInfo serdeInfo) {
    StorageDescriptor sd = new StorageDescriptor(columns, null,
        "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat",
        "org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat",
        false, 0, serdeInfo, null, null, null);
    return new Partition(partVals, dbName, tblName, 0, 0, sd, null);
  }

  private void createFile(Path path, String content) throws IOException {
    FSDataOutputStream output = path.getFileSystem(hiveConf).create(path);
    output.writeChars(content);
    output.close();
  }

  @Test
  public void testRecyclePartTable() throws Exception {
    // Create db1/t1/dt=20160101/part
    //              /dt=20160102/part
    //              /dt=20160103/part
    // Test: recycle single file (dt=20160101/part)
    //       recycle single partition (dt=20160102)
    //       recycle table t1
    String dbName = "db1";
    client.dropDatabase(dbName, true, true);

    Database db = new Database();
    db.putToParameters(SOURCE_OF_REPLICATION, "1,2,3");
    db.setName(dbName);
    client.createDatabase(db);

    String tblName = "t1";
    List<FieldSchema> columns = new ArrayList<FieldSchema>();
    columns.add(new FieldSchema("foo", "string", ""));
    columns.add(new FieldSchema("bar", "string", ""));

    List<FieldSchema> partColumns = new ArrayList<FieldSchema>();
    partColumns.add(new FieldSchema("dt", "string", ""));

    SerDeInfo serdeInfo = new SerDeInfo("LBCSerDe", LazyBinaryColumnarSerDe.class.getCanonicalName(), new HashMap<String, String>());

    StorageDescriptor sd
      = new StorageDescriptor(columns, null,
          "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat",
          "org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat",
    false, 0, serdeInfo, null, null, null);
    Map<String, String> tableParameters = new HashMap<String, String>();

    Table tbl = new Table(tblName, dbName, "", 0, 0, 0, sd, partColumns, tableParameters, "", "", "");

    client.createTable(tbl);

    List<String> values = Arrays.asList("20160101");
    Partition part1 = createPartition(dbName, tblName, columns, values, serdeInfo);
    client.add_partition(part1);

    values = Arrays.asList("20160102");
    Partition part2 = createPartition(dbName, tblName, columns, values, serdeInfo);
    client.add_partition(part2);

    values = Arrays.asList("20160103");
    Partition part3 = createPartition(dbName, tblName, columns, values, serdeInfo);
    client.add_partition(part3);

    Path part1Path = new Path(warehouse.getDefaultPartitionPath(db, tbl, ImmutableMap.of("dt", "20160101")), "part");
    createFile(part1Path, "p1");
    String path1Chksum = ReplChangeManager.checksumFor(part1Path, fs);

    Path part2Path = new Path(warehouse.getDefaultPartitionPath(db, tbl, ImmutableMap.of("dt", "20160102")), "part");
    createFile(part2Path, "p2");
    String path2Chksum = ReplChangeManager.checksumFor(part2Path, fs);

    Path part3Path = new Path(warehouse.getDefaultPartitionPath(db, tbl, ImmutableMap.of("dt", "20160103")), "part");
    createFile(part3Path, "p3");
    String path3Chksum = ReplChangeManager.checksumFor(part3Path, fs);

    assertTrue(part1Path.getFileSystem(hiveConf).exists(part1Path));
    assertTrue(part2Path.getFileSystem(hiveConf).exists(part2Path));
    assertTrue(part3Path.getFileSystem(hiveConf).exists(part3Path));

    ReplChangeManager cm = ReplChangeManager.getInstance(hiveConf);
    // verify cm.recycle(db, table, part) api moves file to cmroot dir
    int ret = cm.recycle(part1Path, RecycleType.MOVE, false);
    Assert.assertEquals(ret, 1);
    Path cmPart1Path = ReplChangeManager.getCMPath(hiveConf, part1Path.getName(), path1Chksum, cmroot.toString());
    assertTrue(cmPart1Path.getFileSystem(hiveConf).exists(cmPart1Path));

    // Verify dropPartition recycle part files
    client.dropPartition(dbName, tblName, Arrays.asList("20160102"));
    assertFalse(part2Path.getFileSystem(hiveConf).exists(part2Path));
    Path cmPart2Path = ReplChangeManager.getCMPath(hiveConf, part2Path.getName(), path2Chksum, cmroot.toString());
    assertTrue(cmPart2Path.getFileSystem(hiveConf).exists(cmPart2Path));

    // Verify dropTable recycle partition files
    client.dropTable(dbName, tblName);
    assertFalse(part3Path.getFileSystem(hiveConf).exists(part3Path));
    Path cmPart3Path = ReplChangeManager.getCMPath(hiveConf, part3Path.getName(), path3Chksum, cmroot.toString());
    assertTrue(cmPart3Path.getFileSystem(hiveConf).exists(cmPart3Path));

    client.dropDatabase(dbName, true, true);
  }

  @Test
  public void testRecycleNonPartTable() throws Exception {
    // Create db2/t1/part1
    //              /part2
    //              /part3
    // Test: recycle single file (part1)
    //       recycle table t1
    String dbName = "db2";
    client.dropDatabase(dbName, true, true);

    Database db = new Database();
    db.putToParameters(SOURCE_OF_REPLICATION, "1, 2, 3");
    db.setName(dbName);
    client.createDatabase(db);

    String tblName = "t1";
    List<FieldSchema> columns = new ArrayList<FieldSchema>();
    columns.add(new FieldSchema("foo", "string", ""));
    columns.add(new FieldSchema("bar", "string", ""));

    SerDeInfo serdeInfo = new SerDeInfo("LBCSerDe", LazyBinaryColumnarSerDe.class.getCanonicalName(), new HashMap<String, String>());

    StorageDescriptor sd
      = new StorageDescriptor(columns, null,
          "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat",
          "org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat",
    false, 0, serdeInfo, null, null, null);
    Map<String, String> tableParameters = new HashMap<String, String>();

    Table tbl = new Table(tblName, dbName, "", 0, 0, 0, sd, null, tableParameters, "", "", "");

    client.createTable(tbl);

    Path filePath1 = new Path(warehouse.getDefaultTablePath(db, tblName), "part1");
    createFile(filePath1, "f1");
    String fileChksum1 = ReplChangeManager.checksumFor(filePath1, fs);

    Path filePath2 = new Path(warehouse.getDefaultTablePath(db, tblName), "part2");
    createFile(filePath2, "f2");
    String fileChksum2 = ReplChangeManager.checksumFor(filePath2, fs);

    Path filePath3 = new Path(warehouse.getDefaultTablePath(db, tblName), "part3");
    createFile(filePath3, "f3");
    String fileChksum3 = ReplChangeManager.checksumFor(filePath3, fs);

    assertTrue(filePath1.getFileSystem(hiveConf).exists(filePath1));
    assertTrue(filePath2.getFileSystem(hiveConf).exists(filePath2));
    assertTrue(filePath3.getFileSystem(hiveConf).exists(filePath3));

    ReplChangeManager cm = ReplChangeManager.getInstance(hiveConf);
    // verify cm.recycle(Path) api moves file to cmroot dir
    cm.recycle(filePath1, RecycleType.MOVE, false);
    assertFalse(filePath1.getFileSystem(hiveConf).exists(filePath1));

    Path cmPath1 = ReplChangeManager.getCMPath(hiveConf, filePath1.getName(), fileChksum1, cmroot.toString());
    assertTrue(cmPath1.getFileSystem(hiveConf).exists(cmPath1));

    // Verify dropTable recycle table files
    client.dropTable(dbName, tblName);

    Path cmPath2 = ReplChangeManager.getCMPath(hiveConf, filePath2.getName(), fileChksum2,cmroot.toString());
    assertFalse(filePath2.getFileSystem(hiveConf).exists(filePath2));
    assertTrue(cmPath2.getFileSystem(hiveConf).exists(cmPath2));

    Path cmPath3 = ReplChangeManager.getCMPath(hiveConf, filePath3.getName(), fileChksum3, cmroot.toString());
    assertFalse(filePath3.getFileSystem(hiveConf).exists(filePath3));
    assertTrue(cmPath3.getFileSystem(hiveConf).exists(cmPath3));

    client.dropDatabase(dbName, true, true);
  }

  @Test
  public void testClearer() throws Exception {
    FileSystem fs = warehouse.getWhRoot().getFileSystem(hiveConf);
    long now = System.currentTimeMillis();
    Path dirDb = new Path(warehouse.getWhRoot(), "db3");
    fs.delete(dirDb, true);
    fs.mkdirs(dirDb);
    Path dirTbl1 = new Path(dirDb, "tbl1");
    fs.mkdirs(dirTbl1);
    Path part11 = new Path(dirTbl1, "part1");
    createFile(part11, "testClearer11");
    String fileChksum11 = ReplChangeManager.checksumFor(part11, fs);
    Path part12 = new Path(dirTbl1, "part2");
    createFile(part12, "testClearer12");
    String fileChksum12 = ReplChangeManager.checksumFor(part12, fs);
    Path dirTbl2 = new Path(dirDb, "tbl2");
    fs.mkdirs(dirTbl2);
    Path part21 = new Path(dirTbl2, "part1");
    createFile(part21, "testClearer21");
    String fileChksum21 = ReplChangeManager.checksumFor(part21, fs);
    Path part22 = new Path(dirTbl2, "part2");
    createFile(part22, "testClearer22");
    String fileChksum22 = ReplChangeManager.checksumFor(part22, fs);
    Path dirTbl3 = new Path(dirDb, "tbl3");
    fs.mkdirs(dirTbl3);
    Path part31 = new Path(dirTbl3, "part1");
    createFile(part31, "testClearer31");
    String fileChksum31 = ReplChangeManager.checksumFor(part31, fs);
    Path part32 = new Path(dirTbl3, "part2");
    createFile(part32, "testClearer32");
    String fileChksum32 = ReplChangeManager.checksumFor(part32, fs);
    ReplChangeManager.getInstance(hiveConf).recycle(dirTbl1, RecycleType.MOVE, false);
    ReplChangeManager.getInstance(hiveConf).recycle(dirTbl2, RecycleType.MOVE, false);
    ReplChangeManager.getInstance(hiveConf).recycle(dirTbl3, RecycleType.MOVE, true);

    assertTrue(fs.exists(ReplChangeManager.getCMPath(hiveConf, part11.getName(), fileChksum11, cmroot.toString())));
    assertTrue(fs.exists(ReplChangeManager.getCMPath(hiveConf, part12.getName(), fileChksum12, cmroot.toString())));
    assertTrue(fs.exists(ReplChangeManager.getCMPath(hiveConf, part21.getName(), fileChksum21, cmroot.toString())));
    assertTrue(fs.exists(ReplChangeManager.getCMPath(hiveConf, part22.getName(), fileChksum22, cmroot.toString())));
    assertTrue(fs.exists(ReplChangeManager.getCMPath(hiveConf, part31.getName(), fileChksum31, cmroot.toString())));
    assertTrue(fs.exists(ReplChangeManager.getCMPath(hiveConf, part32.getName(), fileChksum32, cmroot.toString())));

    fs.setTimes(ReplChangeManager.getCMPath(hiveConf, part11.getName(), fileChksum11, cmroot.toString()),
            now - 7 * 86400 * 1000 * 2, now - 7 * 86400 * 1000 * 2);
    fs.setTimes(ReplChangeManager.getCMPath(hiveConf, part21.getName(), fileChksum21, cmroot.toString()),
            now - 7 * 86400 * 1000 * 2, now - 7 * 86400 * 1000 * 2);
    fs.setTimes(ReplChangeManager.getCMPath(hiveConf, part31.getName(), fileChksum31, cmroot.toString()),
            now - 7 * 86400 * 1000 * 2, now - 7 * 86400 * 1000 * 2);
    fs.setTimes(ReplChangeManager.getCMPath(hiveConf, part32.getName(), fileChksum32, cmroot.toString()),
            now - 7 * 86400 * 1000 * 2, now - 7 * 86400 * 1000 * 2);

    ReplChangeManager.scheduleCMClearer(hiveConf);

    long start = System.currentTimeMillis();
    long end;
    boolean cleared = false;
    do {
      Thread.sleep(200);
      end = System.currentTimeMillis();
      if (end - start > 5000) {
        Assert.fail("timeout, cmroot has not been cleared");
      }
      if (!fs.exists(ReplChangeManager.getCMPath(hiveConf, part11.getName(), fileChksum11, cmroot.toString())) &&
          fs.exists(ReplChangeManager.getCMPath(hiveConf, part12.getName(), fileChksum12, cmroot.toString())) &&
          !fs.exists(ReplChangeManager.getCMPath(hiveConf, part21.getName(), fileChksum21, cmroot.toString())) &&
          fs.exists(ReplChangeManager.getCMPath(hiveConf, part22.getName(), fileChksum22, cmroot.toString())) &&
          !fs.exists(ReplChangeManager.getCMPath(hiveConf, part31.getName(), fileChksum31, cmroot.toString())) &&
          !fs.exists(ReplChangeManager.getCMPath(hiveConf, part32.getName(), fileChksum32, cmroot.toString()))) {
        cleared = true;
      }
    } while (!cleared);
  }

  @Test
  public void testRecycleUsingImpersonation() throws Exception {
    FileSystem fs = warehouse.getWhRoot().getFileSystem(hiveConf);
    Path dirDb = new Path(warehouse.getWhRoot(), "db3");
    long now = System.currentTimeMillis();
    fs.delete(dirDb, true);
    fs.mkdirs(dirDb);
    Path dirTbl1 = new Path(dirDb, "tbl1");
    fs.mkdirs(dirTbl1);
    Path part11 = new Path(dirTbl1, "part1");
    createFile(part11, "testClearer11");
    String fileChksum11 = ReplChangeManager.checksumFor(part11, fs);
    Path part12 = new Path(dirTbl1, "part2");
    createFile(part12, "testClearer12");
    String fileChksum12 = ReplChangeManager.checksumFor(part12, fs);
    final UserGroupInformation proxyUserUgi =
      UserGroupInformation.createRemoteUser("impala");
    setGroupsInConf(UserGroupInformation.getCurrentUser().getGroupNames(),
      "impala", hiveConf);
    //set owner of data path to impala
    fs.setOwner(dirTbl1, "impala", "default");
    fs.setOwner(part11, "impala", "default");
    fs.setOwner(part12, "impala", "default");
    proxyUserUgi.doAs((PrivilegedExceptionAction<Void>) () -> {
      try {
        //impala doesn't have access. Should provide access control exception
        ReplChangeManager.getInstance(hiveConf).recycle(dirTbl1, RecycleType.MOVE, false);
        Assert.fail();
      } catch (AccessControlException e) {
        assertTrue(e.getMessage().contains("Permission denied: user=impala, access=EXECUTE"));
        assertTrue(e.getMessage().contains("/cmroot"));
      }
      return null;
    });
    ReplChangeManager.getInstance().recycle(dirTbl1, RecycleType.MOVE, false);
    Assert.assertFalse(fs.exists(part11));
    Assert.assertFalse(fs.exists(part12));
    assertTrue(fs.exists(ReplChangeManager.getCMPath(hiveConf, part11.getName(), fileChksum11, cmroot)));
    assertTrue(fs.exists(ReplChangeManager.getCMPath(hiveConf, part12.getName(), fileChksum12, cmroot)));
    fs.setTimes(ReplChangeManager.getCMPath(hiveConf, part11.getName(), fileChksum11, cmroot),
      now - 7 * 86400 * 1000 * 2, now - 7 * 86400 * 1000 * 2);
    ReplChangeManager.scheduleCMClearer(hiveConf);

    long start = System.currentTimeMillis();
    long end;
    boolean cleared = false;
    do {
      Thread.sleep(200);
      end = System.currentTimeMillis();
      if (end - start > 5000) {
        Assert.fail("timeout, cmroot has not been cleared");
      }
      if (!fs.exists(ReplChangeManager.getCMPath(hiveConf, part11.getName(), fileChksum11, cmroot)) &&
        fs.exists(ReplChangeManager.getCMPath(hiveConf, part12.getName(), fileChksum12, cmroot))) {
        cleared = true;
      }
    } while (!cleared);
  }

  @Test
  public void tesRecyleImpersionationWithGroupPermissions() throws Exception {
    FileSystem fs = warehouse.getWhRoot().getFileSystem(hiveConf);
    Path dirDb = new Path(warehouse.getWhRoot(), "db3");
    long now = System.currentTimeMillis();
    fs.delete(dirDb, true);
    fs.mkdirs(dirDb);
    Path dirTbl2 = new Path(dirDb, "tbl2");
    fs.mkdirs(dirTbl2);
    Path part21 = new Path(dirTbl2, "part1");
    createFile(part21, "testClearer21");
    String fileChksum21 = ReplChangeManager.checksumFor(part21, fs);
    Path part22 = new Path(dirTbl2, "part2");
    createFile(part22, "testClearer22");
    String fileChksum22 = ReplChangeManager.checksumFor(part22, fs);
    final UserGroupInformation proxyUserUgi =
            UserGroupInformation.createUserForTesting("impala2", new String[] {"supergroup"});
    //set owner of data path to impala2
    fs.setOwner(dirTbl2, "impala2", "default");
    fs.setOwner(part21, "impala2", "default");
    fs.setOwner(part22, "impala2", "default");
    proxyUserUgi.doAs((PrivilegedExceptionAction<Void>) () -> {
      try {
        //impala2 doesn't have access but it belongs to a group which does.
        ReplChangeManager.getInstance(hiveConf).recycle(dirTbl2, RecycleType.MOVE, false);
      } catch (Exception e) {
        Assert.fail();
      }
      return null;
    });
    Assert.assertFalse(fs.exists(part21));
    Assert.assertFalse(fs.exists(part22));
    assertTrue(fs.exists(ReplChangeManager.getCMPath(hiveConf, part21.getName(), fileChksum21, cmroot)));
    assertTrue(fs.exists(ReplChangeManager.getCMPath(hiveConf, part22.getName(), fileChksum22, cmroot)));
    fs.setTimes(ReplChangeManager.getCMPath(hiveConf, part21.getName(), fileChksum21, cmroot),
            now - 7 * 86400 * 1000 * 2, now - 7 * 86400 * 1000 * 2);
    ReplChangeManager.scheduleCMClearer(hiveConf);

    long start = System.currentTimeMillis();
    long end;
    boolean cleared = false;
    do {
      Thread.sleep(200);
      end = System.currentTimeMillis();
      if (end - start > 5000) {
        Assert.fail("timeout, cmroot has not been cleared");
      }
      if (!fs.exists(ReplChangeManager.getCMPath(hiveConf, part21.getName(), fileChksum21, cmroot)) &&
              fs.exists(ReplChangeManager.getCMPath(hiveConf, part22.getName(), fileChksum22, cmroot))) {
        cleared = true;
      }
    } while (!cleared);
  }

  @Test
  public void testRecycleUsingImpersonationWithAccess() throws Exception {
    try {
      ReplChangeManager.resetReplChangeManagerInstance();
      ReplChangeManager.getInstance(permhiveConf);
      FileSystem fs = permWarehouse.getWhRoot().getFileSystem(permhiveConf);
      long now = System.currentTimeMillis();
      Path dirDb = new Path(permWarehouse.getWhRoot(), "db3");
      fs.delete(dirDb, true);
      fs.mkdirs(dirDb);
      Path dirTbl1 = new Path(dirDb, "tbl1");
      fs.mkdirs(dirTbl1);
      Path part11 = new Path(dirTbl1, "part1");
      createFile(part11, "testClearer11");
      String fileChksum11 = ReplChangeManager.checksumFor(part11, fs);
      Path part12 = new Path(dirTbl1, "part2");
      createFile(part12, "testClearer12");
      String fileChksum12 = ReplChangeManager.checksumFor(part12, fs);
      final UserGroupInformation proxyUserUgi =
        UserGroupInformation.createRemoteUser("impala");
      setGroupsInConf(UserGroupInformation.getCurrentUser().getGroupNames(),
        "impala", permhiveConf);
      //set owner of data path to impala
      fs.setOwner(dirTbl1, "impala", "default");
      proxyUserUgi.doAs((PrivilegedExceptionAction<Void>) () -> {
        ReplChangeManager.getInstance().recycle(dirTbl1, RecycleType.MOVE, false);
        return null;
      });
      Assert.assertFalse(fs.exists(part11));
      Assert.assertFalse(fs.exists(part12));
      assertTrue(fs.exists(ReplChangeManager.getCMPath(permhiveConf, part11.getName(), fileChksum11, permCmroot)));
      assertTrue(fs.exists(ReplChangeManager.getCMPath(permhiveConf, part12.getName(), fileChksum12, permCmroot)));
      fs.setTimes(ReplChangeManager.getCMPath(permhiveConf, part11.getName(), fileChksum11, permCmroot),
        now - 7 * 86400 * 1000 * 2, now - 7 * 86400 * 1000 * 2);
      ReplChangeManager.scheduleCMClearer(permhiveConf);

      long start = System.currentTimeMillis();
      long end;
      boolean cleared = false;
      do {
        Thread.sleep(200);
        end = System.currentTimeMillis();
        if (end - start > 5000) {
          Assert.fail("timeout, cmroot has not been cleared");
        }
        //Impala owned file is cleared by Hive CMClearer
        if (!fs.exists(ReplChangeManager.getCMPath(permhiveConf, part11.getName(), fileChksum11, permCmroot)) &&
          fs.exists(ReplChangeManager.getCMPath(permhiveConf, part12.getName(), fileChksum12, permCmroot))) {
          cleared = true;
        }
      } while (!cleared);
    } finally {
      ReplChangeManager.resetReplChangeManagerInstance();
      ReplChangeManager.getInstance(hiveConf);
    }
  }

  private void setGroupsInConf(String[] groupNames, String proxyUserName, Configuration conf)
    throws IOException {
    conf.set(
      DefaultImpersonationProvider.getTestProvider().getProxySuperuserGroupConfKey(proxyUserName),
      StringUtils.join(",", Arrays.asList(groupNames)));
    configureSuperUserIPAddresses(conf, proxyUserName);
    ProxyUsers.refreshSuperUserGroupsConfiguration(conf);
  }

  private void configureSuperUserIPAddresses(Configuration conf,
                                             String superUserShortName) throws IOException {
    List<String> ipList = new ArrayList<String>();
    Enumeration<NetworkInterface> netInterfaceList = NetworkInterface
      .getNetworkInterfaces();
    while (netInterfaceList.hasMoreElements()) {
      NetworkInterface inf = netInterfaceList.nextElement();
      Enumeration<InetAddress> addrList = inf.getInetAddresses();
      while (addrList.hasMoreElements()) {
        InetAddress addr = addrList.nextElement();
        ipList.add(addr.getHostAddress());
      }
    }
    StringBuilder builder = new StringBuilder();
    for (String ip : ipList) {
      builder.append(ip);
      builder.append(',');
    }
    builder.append("127.0.1.1,");
    builder.append(InetAddress.getLocalHost().getCanonicalHostName());
    conf.setStrings(DefaultImpersonationProvider.getTestProvider().getProxySuperuserIpConfKey(superUserShortName),
      builder.toString());
  }

  @Test
  public void shouldIdentifyCMURIs() {
    assertTrue(ReplChangeManager
        .isCMFileUri(new Path("hdfs://localhost:90000/somepath/adir/", "ab.jar#e239s2233")));
    assertFalse(ReplChangeManager
        .isCMFileUri(new Path("/somepath/adir/", "ab.jar")));
  }
}