@Override
  public boolean commit() {
    // TODO: add retry attempts
    String name = Thread.currentThread().getName();
    try {
      for (ReplicationStateSync stateSync : replicationStateSyncList) {
        Thread.currentThread().setName(stateSync.getClusterId());
        LOG.info("starting sync for state " + stateSync);
        stateSync.sync();
        LOG.info("synced state " + stateSync);
      }
    } catch (Exception e) {
      Thread.currentThread().setName(name);
      LOG.error(String.format("Error while trying to commit replication state %s", e.getMessage()), e);
      return false;
    } finally {
      Thread.currentThread().setName(name);
    }

    LOG.info("done syncing to all tables, verifying the timestamps...");
    ReplicationStateSync base = replicationStateSyncList.get(0);
    boolean success = true;
    LOG.info("expecting all timestamps to be similar to: " + base);
    for (int idx = 1; idx < replicationStateSyncList.size(); ++idx) {
      ReplicationStateSync other = replicationStateSyncList.get(idx);
      if (!base.replicationStateIsInSync(other)) {
        LOG.error("the timestamp of other : " + other + " is not matching with base: " + base);
        success = false;
      }
    }
    return success;
  }
@Test
  public void testBasicGlobalCommit() throws Exception {
    String commitTime = "100";
    localCluster.createCOWTable(commitTime, 5, DB_NAME, TBL_NAME);
    // simulate drs
    remoteCluster.createCOWTable(commitTime, 5, DB_NAME, TBL_NAME);
    HiveSyncGlobalCommitParams params = getGlobalCommitConfig(commitTime);
    HiveSyncGlobalCommitTool tool = new HiveSyncGlobalCommitTool(params);
    assertTrue(tool.commit());
    compareEqualLastReplicatedTimeStamp(params);
  }