本文整理了Java中org.apache.hadoop.hbase.coordination.ZKSplitLogManagerCoordination类的一些代码示例，展示了ZKSplitLogManagerCoordination类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ZKSplitLogManagerCoordination类的具体详情如下：
包路径：org.apache.hadoop.hbase.coordination.ZKSplitLogManagerCoordination
类名称：ZKSplitLogManagerCoordination

ZKSplitLogManagerCoordination介绍

[英]ZooKeeper based implementation of SplitLogManagerCoordination
[中]

代码示例

代码示例来源：origin: apache/hbase

@Override
public void submitTask(String path) {
 createNode(path, zkretries);
}

代码示例来源：origin: apache/hbase

@Override
public void deleteTask(String path) {
 deleteNode(path, zkretries);
}

代码示例来源：origin: apache/hbase

private void resubmitOrFail(String path, ResubmitDirective directive) {
 if (resubmitTask(path, findOrCreateOrphanTask(path), directive) == false) {
  setDone(path, FAILURE);
 }
}

代码示例来源：origin: apache/hbase

/**
 * It is possible for a task to stay in UNASSIGNED state indefinitely - say SplitLogManager wants
 * to resubmit a task. It forces the task to UNASSIGNED state but it dies before it could create
 * the RESCAN task node to signal the SplitLogWorkers to pick up the task. To prevent this
 * scenario the SplitLogManager resubmits all orphan and UNASSIGNED tasks at startup.
 * @param path
 */
private void handleUnassignedTask(String path) {
 if (ZKSplitLog.isRescanNode(watcher, path)) {
  return;
 }
 Task task = findOrCreateOrphanTask(path);
 if (task.isOrphan() && (task.incarnation.get() == 0)) {
  LOG.info("Resubmitting unassigned orphan task " + path);
  // ignore failure to resubmit. The timeout-monitor will handle it later
  // albeit in a more crude fashion
  resubmitTask(path, task, FORCE);
 }
}

代码示例来源：origin: apache/hbase

setDone(path, SUCCESS);
  return;
 setDone(path, FAILURE);
 return;
if (slt.isUnassigned()) {
 LOG.debug("Task not yet acquired " + path + ", ver=" + version);
 handleUnassignedTask(path);
} else if (slt.isOwned()) {
 heartbeat(path, version, slt.getServerName());
} else if (slt.isResigned()) {
 LOG.info("Task " + path + " entered state=" + slt.toString());
 resubmitOrFail(path, FORCE);
} else if (slt.isDone()) {
 LOG.info("Task " + path + " entered state=" + slt.toString());
 if (taskFinisher != null && !ZKSplitLog.isRescanNode(watcher, path)) {
  if (taskFinisher.finish(slt.getServerName(), ZKSplitLog.getFileName(path)) == Status.DONE) {
   setDone(path, SUCCESS);
  } else {
   resubmitOrFail(path, CHECK);
  setDone(path, SUCCESS);
 resubmitOrFail(path, CHECK);
} else {
 LOG.error(HBaseMarkers.FATAL, "logic error - unexpected zk state for path = "
   + path + " data = " + slt.toString());
 setDone(path, FAILURE);

代码示例来源：origin: apache/hbase

private void createRescanSuccess(String path) {
 SplitLogCounters.tot_mgr_rescan.increment();
 getDataSetWatch(path, zkretries);
}

代码示例来源：origin: apache/hbase

boolean result = resubmit(path, version);
if (!result) {
 task.heartbeatNoDetails(EnvironmentEdgeManager.currentTime());
rescan(Long.MAX_VALUE);
SplitLogCounters.tot_mgr_resubmit.increment();
return true;

代码示例来源：origin: harbby/presto-connectors

@Override
public void init() throws IOException {
 this.zkretries = conf.getLong("hbase.splitlog.zk.retries", DEFAULT_ZK_RETRIES);
 this.resubmitThreshold = conf.getLong("hbase.splitlog.max.resubmit", DEFAULT_MAX_RESUBMIT);
 this.timeout = conf.getInt(HConstants.HBASE_SPLITLOG_MANAGER_TIMEOUT, DEFAULT_TIMEOUT);
 setRecoveryMode(true);
 if (this.watcher != null) {
  this.watcher.registerListener(this);
  lookForOrphans();
 }
}

代码示例来源：origin: harbby/presto-connectors

private boolean resubmit(ServerName serverName, String path, int version) {
 try {
  // blocking zk call but this is done from the timeout thread
  SplitLogTask slt =
    new SplitLogTask.Unassigned(this.details.getServerName(), getRecoveryMode());
  if (ZKUtil.setData(this.watcher, path, slt.toByteArray(), version) == false) {
   LOG.debug("failed to resubmit task " + path + " version changed");
   return false;
  }
 } catch (NoNodeException e) {
  LOG.warn("failed to resubmit because znode doesn't exist " + path
    + " task done (or forced done by removing the znode)");
  try {
   getDataSetWatchSuccess(path, null, Integer.MIN_VALUE);
  } catch (DeserializationException e1) {
   LOG.debug("Failed to re-resubmit task " + path + " because of deserialization issue", e1);
   return false;
  }
  return false;
 } catch (KeeperException.BadVersionException e) {
  LOG.debug("failed to resubmit task " + path + " version changed");
  return false;
 } catch (KeeperException e) {
  SplitLogCounters.tot_mgr_resubmit_failed.incrementAndGet();
  LOG.warn("failed to resubmit " + path, e);
  return false;
 }
 return true;
}

代码示例来源：origin: apache/hbase

public ZkCoordinatedStateManager(Server server) {
 this.watcher = server.getZooKeeper();
 splitLogWorkerCoordination = new ZkSplitLogWorkerCoordination(server.getServerName(), watcher);
 splitLogManagerCoordination = new ZKSplitLogManagerCoordination(server.getConfiguration(),
   watcher);
}

代码示例来源：origin: apache/hbase

@Override
public void checkTasks() {
 rescan(Long.MAX_VALUE);
}

代码示例来源：origin: apache/hbase

private void heartbeat(String path, int new_version, ServerName workerName) {
 Task task = findOrCreateOrphanTask(path);
 if (new_version != task.last_version) {
  if (task.isUnassigned()) {
   LOG.info("Task " + path + " acquired by " + workerName);
  }
  task.heartbeat(EnvironmentEdgeManager.currentTime(), new_version, workerName);
  SplitLogCounters.tot_mgr_heartbeat.increment();
 } else {
  // duplicate heartbeats - heartbeats w/o zk node version
  // changing - are possible. The timeout thread does
  // getDataSetWatch() just to check whether a node still
  // exists or not
 }
 return;
}

代码示例来源：origin: apache/hbase

@Override
public void init() throws IOException {
 this.zkretries = conf.getLong("hbase.splitlog.zk.retries", DEFAULT_ZK_RETRIES);
 this.resubmitThreshold = conf.getLong("hbase.splitlog.max.resubmit", DEFAULT_MAX_RESUBMIT);
 this.timeout = conf.getInt(HConstants.HBASE_SPLITLOG_MANAGER_TIMEOUT, DEFAULT_TIMEOUT);
 if (this.watcher != null) {
  this.watcher.registerListener(this);
  lookForOrphans();
 }
}

代码示例来源：origin: apache/hbase

private boolean resubmit(String path, int version) {
 try {
  // blocking zk call but this is done from the timeout thread
  SplitLogTask slt =
    new SplitLogTask.Unassigned(this.details.getServerName());
  if (ZKUtil.setData(this.watcher, path, slt.toByteArray(), version) == false) {
   LOG.debug("Failed to resubmit task " + path + " version changed");
   return false;
  }
 } catch (NoNodeException e) {
  LOG.warn("Failed to resubmit because znode doesn't exist " + path
    + " task done (or forced done by removing the znode)");
  try {
   getDataSetWatchSuccess(path, null, Integer.MIN_VALUE);
  } catch (DeserializationException e1) {
   LOG.debug("Failed to re-resubmit task " + path + " because of deserialization issue", e1);
   return false;
  }
  return false;
 } catch (KeeperException.BadVersionException e) {
  LOG.debug("Failed to resubmit task " + path + " version changed");
  return false;
 } catch (KeeperException e) {
  SplitLogCounters.tot_mgr_resubmit_failed.increment();
  LOG.warn("Failed to resubmit " + path, e);
  return false;
 }
 return true;
}

代码示例来源：origin: harbby/presto-connectors

setDone(path, SUCCESS);
  return;
 setDone(path, FAILURE);
 return;
if (slt.isUnassigned()) {
 LOG.debug("task not yet acquired " + path + " ver = " + version);
 handleUnassignedTask(path);
} else if (slt.isOwned()) {
 heartbeat(path, version, slt.getServerName());
} else if (slt.isResigned()) {
 LOG.info("task " + path + " entered state: " + slt.toString());
 resubmitOrFail(path, FORCE);
} else if (slt.isDone()) {
 LOG.info("task " + path + " entered state: " + slt.toString());
 if (taskFinisher != null && !ZKSplitLog.isRescanNode(watcher, path)) {
  if (taskFinisher.finish(slt.getServerName(), ZKSplitLog.getFileName(path)) == Status.DONE) {
   setDone(path, SUCCESS);
  } else {
   resubmitOrFail(path, CHECK);
  setDone(path, SUCCESS);
 resubmitOrFail(path, CHECK);
} else {
 LOG.fatal("logic error - unexpected zk state for path = " + path + " data = "
   + slt.toString());
 setDone(path, FAILURE);

代码示例来源：origin: apache/hbase

private void createNodeSuccess(String path) {
 LOG.debug("Put up splitlog task at znode " + path);
 getDataSetWatch(path, zkretries);
}

代码示例来源：origin: harbby/presto-connectors

/**
 * It is possible for a task to stay in UNASSIGNED state indefinitely - say SplitLogManager wants
 * to resubmit a task. It forces the task to UNASSIGNED state but it dies before it could create
 * the RESCAN task node to signal the SplitLogWorkers to pick up the task. To prevent this
 * scenario the SplitLogManager resubmits all orphan and UNASSIGNED tasks at startup.
 * @param path
 */
private void handleUnassignedTask(String path) {
 if (ZKSplitLog.isRescanNode(watcher, path)) {
  return;
 }
 Task task = findOrCreateOrphanTask(path);
 if (task.isOrphan() && (task.incarnation.get() == 0)) {
  LOG.info("resubmitting unassigned orphan task " + path);
  // ignore failure to resubmit. The timeout-monitor will handle it later
  // albeit in a more crude fashion
  resubmitTask(path, task, FORCE);
 }
}

代码示例来源：origin: harbby/presto-connectors

boolean result = resubmit(this.details.getServerName(), path, version);
if (!result) {
 task.heartbeatNoDetails(EnvironmentEdgeManager.currentTime());
rescan(Long.MAX_VALUE);
SplitLogCounters.tot_mgr_resubmit.incrementAndGet();
return true;

代码示例来源：origin: harbby/presto-connectors

@Override
public void initialize(Server server) {
 this.server = server;
 this.watcher = server.getZooKeeper();
 splitLogWorkerCoordination = new ZkSplitLogWorkerCoordination(this, watcher);
 splitLogManagerCoordination = new ZKSplitLogManagerCoordination(this, watcher);
 splitTransactionCoordination = new ZKSplitTransactionCoordination(this, watcher);
 closeRegionCoordination = new ZkCloseRegionCoordination(this, watcher);
 openRegionCoordination = new ZkOpenRegionCoordination(this, watcher);
 regionMergeCoordination = new ZkRegionMergeCoordination(this, watcher);
}

代码示例来源：origin: harbby/presto-connectors

@Override
public void checkTasks() {
 rescan(Long.MAX_VALUE);
};