/**
 * Ensure Armeria's dependencies do not cause a trouble with hbase-shaded-client.
 *
 * @see <a href="https://issues.apache.org/jira/browse/HBASE-14963">HBASE-14963</a>
 */
@Test(expected = NotAllMetaRegionsOnlineException.class)
public void testGuavaConflict() throws Exception {
    // Make sure Armeria is available in the class path.
    assertThat(Version.getAll(Server.class.getClassLoader())).isNotNull();
    // Make sure newer Guava is available in the class path.
    assertThat(Stopwatch.class.getDeclaredConstructor().getModifiers()).is(new Condition<>(
            value -> !Modifier.isPublic(value),
            "Recent Guava Stopwatch should have non-public default constructor."));

    final MetaTableLocator locator = new MetaTableLocator();
    final ZooKeeperWatcher zkw = mock(ZooKeeperWatcher.class);
    final RecoverableZooKeeper zk = mock(RecoverableZooKeeper.class);
    when(zkw.getRecoverableZooKeeper()).thenReturn(zk);
    when(zk.exists(any(), any())).thenReturn(new Stat(0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0));

    locator.waitMetaRegionLocation(zkw, 100);
}
 

private void testZNodeACLs() throws IOException, KeeperException, InterruptedException {

    ZKWatcher watcher = new ZKWatcher(conf, "IntegrationTestZnodeACLs", null);
    RecoverableZooKeeper zk = ZKUtil.connect(this.conf, watcher);

    String baseZNode = watcher.getZNodePaths().baseZNode;

    LOG.info("");
    LOG.info("***********************************************************************************");
    LOG.info("Checking ZK permissions, root znode: " + baseZNode);
    LOG.info("***********************************************************************************");
    LOG.info("");

    checkZnodePermsRecursive(watcher, zk, baseZNode);

    LOG.info("Checking ZK permissions: SUCCESS");
  }
 

private void checkZnodePermsRecursive(ZKWatcher watcher,
    RecoverableZooKeeper zk, String znode) throws KeeperException, InterruptedException {

  boolean expectedWorldReadable = watcher.getZNodePaths().isClientReadable(znode);

  assertZnodePerms(zk, znode, expectedWorldReadable);

  try {
    List<String> children = zk.getChildren(znode, false);

    for (String child : children) {
      checkZnodePermsRecursive(watcher, zk, ZNodePaths.joinZNode(znode, child));
    }
  } catch (KeeperException ke) {
    // if we are not authenticated for listChildren, it is fine.
    if (ke.code() != Code.NOAUTH && ke.code() != Code.NONODE) {
      throw ke;
    }
  }
}
 

/**
 * Deletes just the splice-specific paths in zookeeper.  Does not delete hbase paths.
 */
public static void cleanZookeeper() throws InterruptedException, KeeperException{
    RecoverableZooKeeper rzk=getRecoverableZooKeeper();
    String rootPath=HConfiguration.getConfiguration().getSpliceRootPath();
    for(String path : HConfiguration.zookeeperPaths){
        path=rootPath+path;
        if(rzk.exists(path,false)!=null){
            for(String child : rzk.getChildren(path,false)){
                for(String grandChild : rzk.getChildren(path+"/"+child,false)){
                    rzk.delete(path+"/"+child+"/"+grandChild,-1);
                }
                rzk.delete(path+"/"+child,-1);
            }
            rzk.delete(path,-1);
        }
    }
}
 

/**
 * Bump up timestamp if the provided timestamp value is larger than current timetamp
 * @param timestamp
 * @throws IOException
 * @throws KeeperException
 * @throws InterruptedException
 */
public static void setTimestamp(long timestamp) throws IOException, KeeperException, InterruptedException {
    TimestampSource timestampSource = SIDriver.driver().getTimestampSource();
    long currentTimestamp = timestampSource.currentTimestamp();
    if (currentTimestamp < timestamp) {
        RecoverableZooKeeper rzk = ZkUtils.getRecoverableZooKeeper();
        HBaseSIEnvironment env = HBaseSIEnvironment.loadEnvironment(new SystemClock(), rzk);
        ConfigurationSource configurationSource = env.configuration().getConfigSource();
        String rootNode = configurationSource.getString(HConfiguration.SPLICE_ROOT_PATH, HConfiguration.DEFAULT_ROOT_PATH);
        String node = rootNode + HConfiguration.MAX_RESERVED_TIMESTAMP_PATH;
        //if (LOG.isDebugEnabled()) {
        SpliceLogUtils.info(LOG, "bump up timestamp to %d", timestamp);
        //}
        byte[] data = Bytes.toBytes(timestamp);
        rzk.setData(node, data, -1 /* version */);
        timestampSource.refresh();
    }
    else {
        //if (LOG.isDebugEnabled()) {
        SpliceLogUtils.info(LOG, "current timestamp = %d >  %d",
                currentTimestamp, timestamp);
        //}
    }
}
 

public static void disableMaster(String masterClusterKey) throws InterruptedException, KeeperException, IOException {

        // Delete all peers from master cluster
        Configuration conf = ReplicationUtils.createConfiguration(masterClusterKey);
        ZKWatcher masterZkw = new ZKWatcher(conf, "replication monitor", null, false);
        RecoverableZooKeeper masterRzk = masterZkw.getRecoverableZooKeeper();
        String[] s = masterClusterKey.split(":");
        String hbaseRootDir = s[2];
        String peerPath = hbaseRootDir+"/replication/peers";
        List<String> peers = masterRzk.getChildren(peerPath, false);
        for (String peer : peers) {
            String p = peerPath + "/" + peer;
            List<String> children = masterRzk.getChildren(p, false);
            String peerStatePath = p + "/" + children.get(0);
            masterRzk.setData(peerStatePath, toByteArray(ReplicationProtos.ReplicationState.State.DISABLED), -1);
            System.out.println("Disabled peer " + peer);
        }
    }
 

private void reportDiagnostics(String diagnostics) {
    try {
        RecoverableZooKeeper rzk = ZkUtils.getRecoverableZooKeeper();
        String root = HConfiguration.getConfiguration().getSpliceRootPath();

        String diagnosticsPath = root + HBaseConfiguration.OLAP_SERVER_PATH + HBaseConfiguration.OLAP_SERVER_DIAGNOSTICS_PATH + "/" + queueName;

        if (rzk.exists(diagnosticsPath, false) != null) {
            rzk.setData(diagnosticsPath, Bytes.toBytes(diagnostics), -1);
        } else {
            rzk.create(diagnosticsPath, Bytes.toBytes(diagnostics), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    } catch (Exception e) {
        LOG.error("Exception while trying to report diagnostics", e);
        // ignore this exception during error reporting
    }
}
 

private void reportDiagnostics(String diagnostics) {
    try {
        RecoverableZooKeeper rzk = ZkUtils.getRecoverableZooKeeper();
        String root = HConfiguration.getConfiguration().getSpliceRootPath();

        String diagnosticsRoot = root + HBaseConfiguration.OLAP_SERVER_PATH + HBaseConfiguration.OLAP_SERVER_DIAGNOSTICS_PATH;
        zkSafeCreate(diagnosticsRoot);
        String diagnosticsPath = diagnosticsRoot + "/spark-" + queueName;

        if (rzk.exists(diagnosticsPath, false) != null) {
            rzk.setData(diagnosticsPath, com.splicemachine.primitives.Bytes.toBytes(diagnostics), -1);
        } else {
            rzk.create(diagnosticsPath, com.splicemachine.primitives.Bytes.toBytes(diagnostics), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    } catch (Exception e) {
        LOG.error("Exception while trying to report diagnostics", e);
        // ignore this exception during error reporting
    }
}
 

private void assertZnodePerms(RecoverableZooKeeper zk, String znode,
    boolean expectedWorldReadable) throws KeeperException, InterruptedException {
  Stat stat = new Stat();
  List<ACL> acls;
  try {
    acls = zk.getZooKeeper().getACL(znode, stat);
  } catch (NoNodeException ex) {
    LOG.debug("Caught exception for missing znode", ex);
    // the znode is deleted. Probably it was a temporary znode (like RIT).
    return;
  }
  String[] superUsers = superUser == null ? null : superUser.split(",");

  LOG.info("Checking ACLs for znode znode:" + znode + " acls:" + acls);

  for (ACL acl : acls) {
    int perms = acl.getPerms();
    Id id = acl.getId();
    // We should only set at most 3 possible ACL for 3 Ids. One for everyone, one for superuser
    // and one for the hbase user
    if (Ids.ANYONE_ID_UNSAFE.equals(id)) {
      // everyone should be set only if we are expecting this znode to be world readable
      assertTrue(expectedWorldReadable);
      // assert that anyone can only read
      assertEquals(perms, Perms.READ);
    } else if (superUsers != null && ZKWatcher.isSuperUserId(superUsers, id)) {
      // assert that super user has all the permissions
      assertEquals(perms, Perms.ALL);
    } else if (new Id("sasl", masterPrincipal).equals(id)) {
      // hbase.master.kerberos.principal?
      assertEquals(perms, Perms.ALL);
    } else {
      fail("An ACL is found which is not expected for the znode:" + znode + " , ACL:" + acl);
    }
  }
}
 

/**
 * Gets a direct interface to a ZooKeeper instance.
 *
 * @return a direct interface to ZooKeeper.
 */
public static RecoverableZooKeeper getRecoverableZooKeeper(){
    try{
        return zkManager.getRecoverableZooKeeper();
    }catch(ZooKeeperConnectionException e){
        LOG.error("Unable to connect to zookeeper, aborting",e);
        throw new RuntimeException(e);
    }
}
 

public static boolean safeDelete(String path,int version, RecoverableZooKeeper rzk) throws KeeperException, InterruptedException{
    try{
        rzk.delete(path,version);
        return true;
    }catch(KeeperException e){
        if(e.code()!=KeeperException.Code.NONODE)
            throw e;
        else
            return false;
    }
}
 

public static boolean recursiveSafeCreate(String path,byte[] bytes,List<ACL> acls,CreateMode createMode) throws InterruptedException, KeeperException{
    if(path==null || path.length()<=0) return true; //nothing to do, we've gone all the way to the root
    RecoverableZooKeeper rzk=getRecoverableZooKeeper();
    try{
        return safeCreate(path,bytes,acls,createMode,rzk);
    }catch(KeeperException e){
        if(e.code()==KeeperException.Code.NONODE){
            //parent node doesn't exist, so recursively create it, and then try and create your node again
            String parent=path.substring(0,path.lastIndexOf('/'));
            recursiveSafeCreate(parent,new byte[]{},acls,CreateMode.PERSISTENT);
            return safeCreate(path,bytes,acls,createMode);
        }else
            throw e;
    }
}
 

public static boolean recursiveSafeCreate(String path,byte[] bytes,List<ACL> acls,CreateMode createMode, RecoverableZooKeeper rzk) throws InterruptedException, KeeperException{
    if(path==null || path.length()<=0) return true; //nothing to do, we've gone all the way to the root
    try{
        return safeCreate(path,bytes,acls,createMode,rzk);
    }catch(KeeperException e){
        if(e.code()==KeeperException.Code.NONODE){
            //parent node doesn't exist, so recursively create it, and then try and create your node again
            String parent=path.substring(0,path.lastIndexOf('/'));
            recursiveSafeCreate(parent,new byte[]{},acls,CreateMode.PERSISTENT, rzk);
            return safeCreate(path,bytes,acls,createMode, rzk);
        }else
            throw e;
    }
}
 

public static boolean safeCreate(String path,byte[] bytes,List<ACL> acls,CreateMode createMode,RecoverableZooKeeper zooKeeper) throws KeeperException, InterruptedException{
    try{
        zooKeeper.create(path,bytes,acls,createMode);
        return true;
    }catch(KeeperException ke){
        if(ke.code()!=KeeperException.Code.NODEEXISTS)
            throw ke;
        else
            return true;
    }
}
 

public static void recursiveDelete(String path, RecoverableZooKeeper rzk) throws InterruptedException, KeeperException, IOException{
    List<String> children=getChildren(path,false, rzk);
    for(String child : children){
        recursiveDelete(path+"/"+child, rzk);
    }
    delete(path, rzk);
}
 

public static List<String> getChildren(String path,boolean watch, RecoverableZooKeeper rzk) throws IOException{
    try{
        return rzk.getChildren(path,watch);
    }catch(InterruptedException | KeeperException e){
        throw new IOException(e);
    }
}
 

public static void setData(String path,byte[] data,int version, RecoverableZooKeeper rzk) throws IOException{
    try{
        rzk.setData(path,data,version);
    }catch(KeeperException | InterruptedException e){
        throw new IOException(e);
    }
}
 

public static boolean validZookeeper() throws InterruptedException, KeeperException{
    RecoverableZooKeeper rzk=getRecoverableZooKeeper();

    String rootPath=HConfiguration.getConfiguration().getSpliceRootPath();
    for(String path : HConfiguration.zookeeperPaths){
        if(rzk.exists(rootPath+path,false)==null)
            return false;
    }
    return true;
}
 

public static HBaseSIEnvironment loadEnvironment(Clock clock,RecoverableZooKeeper rzk) throws IOException{
    HBaseSIEnvironment env = INSTANCE;
    if(env==null){
        synchronized(HBaseSIEnvironment.class){
            env = INSTANCE;
            if(env==null){
                env = INSTANCE = new HBaseSIEnvironment(rzk,clock);
            }
        }
    }
    return env;
}
 

@SuppressWarnings("unchecked")
public HBaseSIEnvironment(RecoverableZooKeeper rzk,Clock clock) throws IOException{
    ByteComparisons.setComparator(HBaseComparator.INSTANCE);
    this.config=HConfiguration.getConfiguration();

    this.timestampSource = new ZkTimestampSource(config,rzk);
    this.partitionCache = PartitionCacheService.loadPartitionCache(config);
    this.partitionFactory = TableFactoryService.loadTableFactory(clock, this.config,partitionCache);
    this.oldestActiveTransactionTaskFactory = new HOldestActiveTransactionTaskFactory();
    TxnNetworkLayerFactory txnNetworkLayerFactory= TableFactoryService.loadTxnNetworkLayer(this.config);
    this.txnStore = new CoprocessorTxnStore(txnNetworkLayerFactory,timestampSource,null);
    int completedTxnCacheSize = config.getCompletedTxnCacheSize();
    int completedTxnConcurrency = config.getCompletedTxnConcurrency();
    this.txnSupplier = new CompletedTxnCacheSupplier(txnStore,completedTxnCacheSize,completedTxnConcurrency);
    this.txnStore.setCache(txnSupplier);
    this.opFactory = HOperationFactory.INSTANCE;
    this.txnOpFactory = new SimpleTxnOperationFactory(exceptionFactory(),opFactory);
    this.clock = clock;
    this.fileSystem = new HNIOFileSystem(FileSystem.get((Configuration) config.getConfigSource().unwrapDelegate()), exceptionFactory());
    this.snowflakeFactory = new HSnowflakeFactory();
    this.clusterHealthFactory = new HClusterHealthFactory(rzk);
    this.ignoreTxnSupplier = new IgnoreTxnSupplier(partitionFactory, txnOpFactory);
    this.filesystemAdmin = new HFilesystemAdmin(HBaseConnectionFactory.getInstance(config).getConnection().getAdmin());

    this.keepAlive = new QueuedKeepAliveScheduler(config.getTransactionKeepAliveInterval(),
            config.getTransactionTimeout(),
            config.getTransactionKeepAliveThreads(),
            txnStore);
    this.rollForward = new HBaseRollForward(txnSupplier, config);
    this.rollForward.start();
    siDriver = SIDriver.loadDriver(this);
}
 

public static AdapterSIEnvironment loadEnvironment(Clock clock, RecoverableZooKeeper rzk, DataSource connectionPool) throws IOException{
    AdapterSIEnvironment env = INSTANCE;
    if(env==null){
        synchronized(AdapterSIEnvironment.class){
            env = INSTANCE;
            if(env==null){
                env = INSTANCE = new AdapterSIEnvironment(rzk,clock,connectionPool);
            }
        }
    }
    return env;
}
 

@SuppressWarnings("unchecked")
public AdapterSIEnvironment(RecoverableZooKeeper rzk, Clock clock, DataSource connectionPool) throws IOException{
    ByteComparisons.setComparator(HBaseComparator.INSTANCE);
    this.config=HConfiguration.getConfiguration();

    this.timestampSource =new ZkTimestampSource(config,rzk);
    this.partitionCache = PartitionCacheService.loadPartitionCache(config);
    this.partitionFactory = new AdapterTableFactory(connectionPool);
    this.partitionFactory.initialize(clock, this.config, partitionCache);
    TxnNetworkLayerFactory txnNetworkLayerFactory= TableFactoryService.loadTxnNetworkLayer(this.config);
    this.oldestActiveTransactionTaskFactory = new HOldestActiveTransactionTaskFactory();
    this.txnStore = new CoprocessorTxnStore(txnNetworkLayerFactory,timestampSource,null);
    int completedTxnCacheSize = config.getCompletedTxnCacheSize();
    int completedTxnConcurrency = config.getCompletedTxnConcurrency();
    this.txnSupplier = new CompletedTxnCacheSupplier(txnStore,completedTxnCacheSize,completedTxnConcurrency);
    this.txnStore.setCache(txnSupplier);
    this.opFactory =HOperationFactory.INSTANCE;
    this.txnOpFactory = new SimpleTxnOperationFactory(exceptionFactory(),opFactory);
    this.clock = clock;
    this.fileSystem =new HNIOFileSystem(FileSystem.get((Configuration) config.getConfigSource().unwrapDelegate()), exceptionFactory());
    this.snowflakeFactory = new HSnowflakeFactory();
    this.clusterHealthFactory = new HClusterHealthFactory(rzk);
    this.ignoreTxnSupplier = new IgnoreTxnSupplier(partitionFactory, txnOpFactory);
    this.filesystemAdmin = new HFilesystemAdmin(HBaseConnectionFactory.getInstance(config).getConnection().getAdmin());

    this.keepAlive = new QueuedKeepAliveScheduler(config.getTransactionKeepAliveInterval(),
            config.getTransactionTimeout(),
            config.getTransactionKeepAliveThreads(),
            txnStore);
    siDriver = SIDriver.loadDriver(this);
}
 

public static boolean shouldCaptureIncrementalChanges(FileSystem fs,Path rootDir) throws StandardException{
    boolean shouldRegister = false;
    try {
        boolean enabled = incrementalBackupEnabled();
        if (enabled) {
            RecoverableZooKeeper zooKeeper = ZkUtils.getRecoverableZooKeeper();
            String spliceBackupPath = BackupUtils.getBackupPath();
            if (zooKeeper.exists(spliceBackupPath, false)==null){
                return false;
            }
            boolean isRestoreMode = SIDriver.driver().lifecycleManager().isRestoreMode();
            if (!isRestoreMode) {
                if (BackupUtils.existsDatabaseBackup(fs, rootDir)) {
                    if (LOG.isDebugEnabled()) {
                        SpliceLogUtils.debug(LOG, "There exists a successful full or incremental backup in the system");
                    }
                    shouldRegister = true;
                } else {
                    List<String> backupJobs = zooKeeper.getChildren(spliceBackupPath, false);
                    for (String backupId : backupJobs) {
                        String path = spliceBackupPath + "/" + backupId;
                        byte[] data = zooKeeper.getData(path, false, null);
                        BackupJobStatus status = BackupJobStatus.parseFrom(data);
                        if (status.getScope() == BackupJobStatus.BackupScope.DATABASE) {
                            if (LOG.isDebugEnabled()) {
                                SpliceLogUtils.debug(LOG, "A database backup is running");
                            }
                            shouldRegister = true;
                        }
                    }
                }
            }
        }
        return shouldRegister;
    }
    catch (Exception e) {
        e.printStackTrace();
        throw StandardException.plainWrapException(e);
    }
}
 

private String getMasterCluster() throws InterruptedException, KeeperException{
    RecoverableZooKeeper rzk = zkWatcher.getRecoverableZooKeeper();
    String path = replicationMonitorPath + "/master";
    if (rzk.exists(path, false) != null) {
        List<String> children = rzk.getChildren(path, false);
        return children.size() > 0 ? children.get(0) : null;
    }
    return null;
}
 

private boolean involvedInReplication() throws InterruptedException, KeeperException{
    RecoverableZooKeeper rzk = zkWatcher.getRecoverableZooKeeper();
    String masterPath = replicationMonitorPath + "/master/" + thisCluster;
    String peerPath = replicationMonitorPath + "/peers/" + thisCluster;
    return rzk.exists(masterPath, false) != null || rzk.exists(peerPath, false) != null;

}
 

public ReplicationMonitorWatcher(ReplicationMonitorChore replicationMonitor,
                                 RecoverableZooKeeper rzk,
                                 String monitorPath,
                                 String clusterKey) {
    this.replicationMonitor = replicationMonitor;
    this.rzk = rzk;
    this.monitorPath = monitorPath;
    this.clusterKey = clusterKey;
}
 

private long getMasterTimestamp() throws IOException, KeeperException, InterruptedException {
    String masterClusterKey = getClusterKey(masterCluster);
    Configuration conf = ReplicationUtils.createConfiguration(masterClusterKey);

    try (ZKWatcher masterZkw = new ZKWatcher(conf, "replication monitor", null, false)) {
        RecoverableZooKeeper rzk = masterZkw.getRecoverableZooKeeper();
        String rootNode = HConfiguration.getConfiguration().getSpliceRootPath();
        String node = rootNode + HConfiguration.MAX_RESERVED_TIMESTAMP_PATH;
        byte[] data = rzk.getData(node, false, null);
        long ts = Bytes.toLong(data);
        return ts;
    }
}
 

private boolean isMasterReachable() {
    String masterClusterKey = getClusterKey(masterCluster);
    Configuration conf = ReplicationUtils.createConfiguration(masterClusterKey);

    try (ZKWatcher masterZkw = new ZKWatcher(conf, "replication monitor", null, false)) {
        String[] s = masterClusterKey.split(":");
        RecoverableZooKeeper rzk = masterZkw.getRecoverableZooKeeper();
        List<String> children = rzk.getChildren(s[2], false);
        return true;
    }
    catch (Exception e) {
        return false;
    }
}
 

private void setNewMasterTimestamp(String newMasterClusterKey,
                                   long ts) throws IOException, KeeperException, InterruptedException{
    Configuration conf = ReplicationUtils.createConfiguration(newMasterClusterKey);

    try (ZKWatcher masterZkw = new ZKWatcher(conf, "replication monitor", null, false)) {
        RecoverableZooKeeper rzk = masterZkw.getRecoverableZooKeeper();
        String rootNode = HConfiguration.getConfiguration().getSpliceRootPath();
        String node = rootNode + HConfiguration.MAX_RESERVED_TIMESTAMP_PATH;
        rzk.setData(node, Bytes.toBytes(ts), -1);
    }
}
 

public static List<String> getRegionServers(Configuration conf) throws IOException, InterruptedException, KeeperException{
    try(ZKWatcher zkWatcher = new ZKWatcher(conf, "replication monitor", null, false)) {
        RecoverableZooKeeper zk = zkWatcher.getRecoverableZooKeeper();
        List<String> servers = zk.getChildren("/splice/servers", false);
        return servers;
    }
}