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

ZooKeeperItf.retryOperation介绍

[英]Perform the given operation, retrying in case of connection loss.

Note that in case of connection loss, you are never sure if the operation succeeded or not, so it might be executed twice. Therefore:

• in case of a delete operation, be prepared to deal with a NoNode exception
• in case of a create operation, be prepared to deal with a NodeExists exception
• in case of creation of a sequential node, two nodes might have been created. If they are ephemeral, you can use Stat.ephemeralOwner to find out the ones that belong to the current session. Otherwise, embed the necessary identification into the name or data.

Do not call this method from within a ZooKeeper watcher callback, as it might block for a longer time and hence block the delivery of other events, including the Disconnected event.

This is a Lily-specific method.
[中]执行给定的操作，在连接丢失时重试。
请注意，在连接丢失的情况下，您永远无法确定操作是否成功，因此可能会执行两次。因此：
*如果是删除操作，请准备好处理非节点异常
*如果是创建操作，请准备好处理NodeExists异常
*在创建顺序节点的情况下，可能已经创建了两个节点。如果它们是短暂的，可以使用Stat.ephemeralOwner查找属于当前会话的。否则，在名称或数据中嵌入必要的标识。
不要在ZooKeeper watcher回调中调用此方法，因为它可能会阻塞更长的时间，从而阻塞其他事件的传递，包括断开连接的事件。
这是一种百合特有的方法。

代码示例

代码示例来源：origin: NGDATA/hbase-indexer

/**
 * Updates data on a zookeeper node.
 * 
 * <p>
 * The supplied data is used for the last node in the path. The path must
 * already exist. It is not checked if the data is changed or not. This will
 * cause the version of the node to be increased.
 * <p>
 * This operation is retried until it succeeds.
 */
public static void update(final ZooKeeperItf zk, final String path, final byte[] data, final int version)
    throws InterruptedException, KeeperException {
  zk.retryOperation(new ZooKeeperOperation<Boolean>() {
    @Override
    public Boolean execute() throws KeeperException, InterruptedException {
      zk.setData(path, data, version);
      return null;
    }
  });
}

代码示例来源：origin: com.ngdata/hbase-sep-impl-common

/**
 * Updates data on a zookeeper node.
 * 
 * <p>
 * The supplied data is used for the last node in the path. The path must
 * already exist. It is not checked if the data is changed or not. This will
 * cause the version of the node to be increased.
 * <p>
 * This operation is retried until it succeeds.
 */
public static void update(final ZooKeeperItf zk, final String path, final byte[] data, final int version)
    throws InterruptedException, KeeperException {
  zk.retryOperation(new ZooKeeperOperation<Boolean>() {
    @Override
    public Boolean execute() throws KeeperException, InterruptedException {
      zk.setData(path, data, version);
      return null;
    }
  });
}

代码示例来源：origin: com.ngdata/hbase-indexer-model

@Override
public void unregisterIndexerProcess(final String indexerProcessId) {
  try {
    zk.retryOperation(new ZooKeeperOperation<Integer>() {
      @Override
      public Integer execute() throws KeeperException, InterruptedException {
        zk.delete(indexerProcessId, -1);
        return 0;
      }
    });
  } catch (Exception e) {
    throw new RuntimeException("Error unregistering indexer process " + indexerProcessId, e);
  }
}

代码示例来源：origin: NGDATA/hbase-indexer

@Override
public void unregisterIndexerProcess(final String indexerProcessId) {
  try {
    zk.retryOperation(new ZooKeeperOperation<Integer>() {
      @Override
      public Integer execute() throws KeeperException, InterruptedException {
        zk.delete(indexerProcessId, -1);
        return 0;
      }
    });
  } catch (Exception e) {
    throw new RuntimeException("Error unregistering indexer process " + indexerProcessId, e);
  }
}

代码示例来源：origin: com.ngdata/hbase-indexer-model

@Override
public List<IndexerProcess> getIndexerProcesses(final String indexerName) {
  
  try {
    return zk.retryOperation(new ZooKeeperOperation<List<IndexerProcess>>(){
      @Override
      public List<IndexerProcess> execute() throws KeeperException, InterruptedException {
        List<IndexerProcess> indexerProcesses = Lists.newArrayList();
        for (String childNode : zk.getChildren(zkBaseNode, false)) {
          List<String> nodeNameParts = Lists.newArrayList(Splitter.on(',').split(childNode));
          if (indexerName.equals(nodeNameParts.get(0))) {
            byte[] errorBytes = zk.getData(zkBaseNode + "/" + childNode, false, null);
            IndexerProcess indexerProcess = new IndexerProcess(indexerName,
                nodeNameParts.get(1),
                errorBytes == null  || errorBytes.length == 0? null : Bytes.toString(errorBytes));
            indexerProcesses.add(indexerProcess);
          }
        }
        return indexerProcesses;
        
      }});
  } catch (Exception e) {
    throw new RuntimeException("Error listing indexer processes for " + indexerName, e);
  }
}

代码示例来源：origin: NGDATA/hbase-indexer

@Override
public String registerIndexerProcess(String indexerName, String hostName) {
  // TODO Each indexer should have its own parent node for all its processes
  // TODO Make sure that commas in an indexer name won't cause issues
  final String zkNodePathBase = String.format("%s/%s,%s,", zkBaseNode, indexerName, hostName);
  try {
    return zk.retryOperation(new ZooKeeperOperation<String>() {
      @Override
      public String execute() throws KeeperException, InterruptedException {
        return zk.create(zkNodePathBase, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL_SEQUENTIAL);
      }
    });
  } catch (Exception e) {
    throw new RuntimeException("Error while registering indexer process", e);
  }
}

代码示例来源：origin: NGDATA/hbase-indexer

@Override
public List<IndexerProcess> getIndexerProcesses(final String indexerName) {
  
  try {
    return zk.retryOperation(new ZooKeeperOperation<List<IndexerProcess>>(){
      @Override
      public List<IndexerProcess> execute() throws KeeperException, InterruptedException {
        List<IndexerProcess> indexerProcesses = Lists.newArrayList();
        for (String childNode : zk.getChildren(zkBaseNode, false)) {
          List<String> nodeNameParts = Lists.newArrayList(Splitter.on(',').split(childNode));
          if (indexerName.equals(nodeNameParts.get(0))) {
            byte[] errorBytes = zk.getData(zkBaseNode + "/" + childNode, false, null);
            IndexerProcess indexerProcess = new IndexerProcess(indexerName,
                nodeNameParts.get(1),
                errorBytes == null  || errorBytes.length == 0? null : Bytes.toString(errorBytes));
            indexerProcesses.add(indexerProcess);
          }
        }
        return indexerProcesses;
        
      }});
  } catch (Exception e) {
    throw new RuntimeException("Error listing indexer processes for " + indexerName, e);
  }
}

代码示例来源：origin: com.ngdata/hbase-indexer-model

@Override
public String registerIndexerProcess(String indexerName, String hostName) {
  // TODO Each indexer should have its own parent node for all its processes
  // TODO Make sure that commas in an indexer name won't cause issues
  final String zkNodePathBase = String.format("%s/%s,%s,", zkBaseNode, indexerName, hostName);
  try {
    return zk.retryOperation(new ZooKeeperOperation<String>() {
      @Override
      public String execute() throws KeeperException, InterruptedException {
        return zk.create(zkNodePathBase, new byte[0],
            CreateMode.EPHEMERAL_SEQUENTIAL);
      }
    });
  } catch (Exception e) {
    throw new RuntimeException("Error while registering indexer process", e);
  }
}

代码示例来源：origin: NGDATA/hbase-indexer

/**
   * Gets data from a zookeeper node.
   * <p>
   * This operation is retried until it succeeds.
   */
  public static byte[] getData(final ZooKeeperItf zk, final String path, final Watcher watcher, final Stat stat)
      throws InterruptedException, KeeperException {
    final List<byte[]> data = new ArrayList<byte[]>(1);
    zk.retryOperation(new ZooKeeperOperation<Boolean>() {
      @Override
      public Boolean execute() throws KeeperException, InterruptedException {
        data.add(zk.getData(path, watcher, stat));
        return null;
      }
    });
    return data.get(0);
  }
}

代码示例来源：origin: com.ngdata/hbase-sep-impl-common

/**
   * Gets data from a zookeeper node.
   * <p>
   * This operation is retried until it succeeds.
   */
  public static byte[] getData(final ZooKeeperItf zk, final String path, final Watcher watcher, final Stat stat)
      throws InterruptedException, KeeperException {
    final List<byte[]> data = new ArrayList<byte[]>(1);
    zk.retryOperation(new ZooKeeperOperation<Boolean>() {
      @Override
      public Boolean execute() throws KeeperException, InterruptedException {
        data.add(zk.getData(path, watcher, stat));
        return null;
      }
    });
    return data.get(0);
  }
}

代码示例来源：origin: NGDATA/hbase-indexer

@Override
public void setErrorStatus(final String indexerProcessId, Throwable error) {
  final String stackTrace = ExceptionUtils.getStackTrace(error);
  try {
    zk.retryOperation(new ZooKeeperOperation<Integer>() {
      @Override
      public Integer execute() throws KeeperException, InterruptedException {
        zk.setData(indexerProcessId, Bytes.toBytes(stackTrace), -1);
        return 0;
      }
    });
  } catch (Exception e) {
    throw new RuntimeException("Error while setting error status on indexer node " + indexerProcessId, e);
  }
}

代码示例来源：origin: com.ngdata/hbase-indexer-model

@Override
public void setErrorStatus(final String indexerProcessId, Throwable error) {
  final String stackTrace = ExceptionUtils.getStackTrace(error);
  try {
    zk.retryOperation(new ZooKeeperOperation<Integer>() {
      @Override
      public Integer execute() throws KeeperException, InterruptedException {
        zk.setData(indexerProcessId, Bytes.toBytes(stackTrace), -1);
        return 0;
      }
    });
  } catch (Exception e) {
    throw new RuntimeException("Error while setting error status on indexer node " + indexerProcessId, e);
  }
}

代码示例来源：origin: NGDATA/hbase-indexer

private void connectWithZooKeeper() throws IOException, KeeperException, InterruptedException {
  int zkSessionTimeout = HBaseIndexerConfiguration.getSessionTimeout(conf);
  zk = new StateWatchingZooKeeper(zkConnectionString, zkSessionTimeout);
  final String zkRoot = conf.get("hbaseindexer.zookeeper.znode.parent");
  boolean indexerNodeExists = zk.retryOperation(new ZooKeeperOperation<Boolean>() {
    @Override
    public Boolean execute() throws KeeperException, InterruptedException {
      return zk.exists(zkRoot, false) != null;
    }
  });
  if (!indexerNodeExists) {
    System.err.println();
    System.err.println("WARNING: No " + zkRoot + " node found in ZooKeeper.");
    System.err.println();
  }
}

代码示例来源：origin: NGDATA/hbase-indexer

/**
 * Verifies that the specified lockId is the owner of the lock.
 */
public static boolean ownsLock(final ZooKeeperItf zk, final String lockId) throws ZkLockException {
  if (zk.isCurrentThreadEventThread()) {
    throw new RuntimeException("ZkLock should not be used from within the ZooKeeper event thread.");
  }
  try {
    int lastSlashPos = lockId.lastIndexOf('/');
    final String lockPath = lockId.substring(0, lastSlashPos);
    String lockName = lockId.substring(lastSlashPos + 1);
    List<String> children = zk.retryOperation(new ZooKeeperOperation<List<String>>() {
      @Override
      public List<String> execute() throws KeeperException, InterruptedException {
        return zk.getChildren(lockPath, null);
      }
    });
    if (children.isEmpty())
      return false;
    SortedSet<String> sortedChildren = new TreeSet<String>(children);
    return sortedChildren.first().equals(lockName);
  } catch (Throwable t) {
    throw new ZkLockException("Error checking lock, path: " + lockId, t);
  }
}

代码示例来源：origin: com.ngdata/hbase-indexer-cli

private void connectWithZooKeeper() throws IOException, KeeperException, InterruptedException {
  int zkSessionTimeout = HBaseIndexerConfiguration.getSessionTimeout(conf);
  zk = new StateWatchingZooKeeper(zkConnectionString, zkSessionTimeout, new DefaultACLProvider());
  final String zkRoot = conf.get("hbaseindexer.zookeeper.znode.parent");
  boolean indexerNodeExists = zk.retryOperation(new ZooKeeperOperation<Boolean>() {
    @Override
    public Boolean execute() throws KeeperException, InterruptedException {
      return zk.exists(zkRoot, false) != null;
    }
  });
  if (!indexerNodeExists) {
    System.err.println();
    System.err.println("WARNING: No " + zkRoot + " node found in ZooKeeper.");
    System.err.println();
  }
}

代码示例来源：origin: NGDATA/hbase-indexer

private void proposeAsLeader() throws LeaderElectionSetupException, InterruptedException, KeeperException {
  ZkUtil.createPath(zk, electionPath);
  try {
    // In case of connection loss, a node might have been created for us (we do not know it). Therefore,
    // retrying upon connection loss is important, so that we can continue with watching the leaders.
    // Later on, we do not look at the name of the node we created here, but at the owner.
    zk.retryOperation(new ZooKeeperOperation<String>() {
      @Override
      public String execute() throws KeeperException, InterruptedException {
        return zk.create(electionPath + "/n_", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
      }
    });
  } catch (KeeperException e) {
    throw new LeaderElectionSetupException("Error creating leader election zookeeper node below " +
        electionPath, e);
  }
  watchLeaders();
}

代码示例来源：origin: com.ngdata/hbase-indexer-common

private void proposeAsLeader() throws LeaderElectionSetupException, InterruptedException, KeeperException {
  ZkUtil.createPath(zk, electionPath);
  try {
    // In case of connection loss, a node might have been created for us (we do not know it). Therefore,
    // retrying upon connection loss is important, so that we can continue with watching the leaders.
    // Later on, we do not look at the name of the node we created here, but at the owner.
    zk.retryOperation(new ZooKeeperOperation<String>() {
      @Override
      public String execute() throws KeeperException, InterruptedException {
        return zk.create(electionPath + "/n_", null, CreateMode.EPHEMERAL_SEQUENTIAL);
      }
    });
  } catch (KeeperException e) {
    throw new LeaderElectionSetupException("Error creating leader election zookeeper node below " +
        electionPath, e);
  }
  watchLeaders();
}

代码示例来源：origin: NGDATA/hbase-indexer

private IndexerDefinition loadIndexer(String indexerName, boolean forCache)
    throws InterruptedException, KeeperException, IndexerNotFoundException {
  final String childPath = indexerCollectionPath + "/" + indexerName;
  final Stat stat = new Stat();
  byte[] data;
  try {
    if (forCache) {
      // do not retry, install watcher
      data = zk.getData(childPath, watcher, stat);
    } else {
      // do retry, do not install watcher
      data = zk.retryOperation(new ZooKeeperOperation<byte[]>() {
        @Override
        public byte[] execute() throws KeeperException, InterruptedException {
          return zk.getData(childPath, false, stat);
        }
      });
    }
  } catch (KeeperException.NoNodeException e) {
    throw new IndexerNotFoundException(indexerName);
  }
  IndexerDefinitionBuilder builder = IndexerDefinitionJsonSerDeser.INSTANCE.fromJsonBytes(data);
  builder.name(indexerName);
  builder.occVersion(stat.getVersion());
  return builder.build();
}

代码示例来源：origin: com.ngdata/hbase-indexer-model

@Override
public void updateIndexerInternal(final IndexerDefinition indexer) throws InterruptedException, KeeperException,
    IndexerNotFoundException, IndexerConcurrentModificationException, IndexerValidityException {
  assertValid(indexer);
  final byte[] newData = IndexerDefinitionJsonSerDeser.INSTANCE.toJsonBytes(indexer);
  try {
    zk.retryOperation(new ZooKeeperOperation<Stat>() {
      @Override
      public Stat execute() throws KeeperException, InterruptedException {
        return zk.setData(indexerCollectionPathSlash + indexer.getName(), newData, indexer.getOccVersion());
      }
    });
  } catch (KeeperException.NoNodeException e) {
    throw new IndexerNotFoundException(indexer.getName());
  } catch (KeeperException.BadVersionException e) {
    throw new IndexerConcurrentModificationException(indexer.getName());
  }
}

代码示例来源：origin: NGDATA/hbase-indexer

@Override
public void updateIndexerInternal(final IndexerDefinition indexer) throws InterruptedException, KeeperException,
    IndexerNotFoundException, IndexerConcurrentModificationException, IndexerValidityException {
  assertValid(indexer);
  final byte[] newData = IndexerDefinitionJsonSerDeser.INSTANCE.toJsonBytes(indexer);
  try {
    zk.retryOperation(new ZooKeeperOperation<Stat>() {
      @Override
      public Stat execute() throws KeeperException, InterruptedException {
        return zk.setData(indexerCollectionPathSlash + indexer.getName(), newData, indexer.getOccVersion());
      }
    });
  } catch (KeeperException.NoNodeException e) {
    throw new IndexerNotFoundException(indexer.getName());
  } catch (KeeperException.BadVersionException e) {
    throw new IndexerConcurrentModificationException(indexer.getName());
  }
}