@Test public void testCreateAll() throws Exception { startZK(1); ZookeeperClient client = createClient(SES_TIMEOUT); client.createIfNeeded("/apacheIgnite", null, CreateMode.PERSISTENT); List<String> paths = new ArrayList<>(); paths.add("/apacheIgnite/1"); paths.add("/apacheIgnite/2"); paths.add("/apacheIgnite/3"); client.createAll(paths, CreateMode.PERSISTENT); assertEquals(3, client.getChildren("/apacheIgnite").size()); }
void createAll(List<String> paths, CreateMode createMode) throws ZookeeperClientFailedException, InterruptedException { if (paths.isEmpty()) return; List<List<Op>> batches = new LinkedList<>(); int batchSize = 0; List<Op> batch = new LinkedList<>(); for (String path : paths) { int size = requestOverhead(path) + 48 ; assert size <= MAX_REQ_SIZE; if (batchSize + size > MAX_REQ_SIZE) { batches.add(batch); batch = new LinkedList<>(); batchSize = 0; } batch.add(Op.create(path, EMPTY_BYTES, ZK_ACL, createMode)); batchSize += size; } batches.add(batch); for (List<Op> ops : batches) { for (;;) { long connStartTime = this.connStartTime; try { zk.multi(ops); break; } catch (KeeperException.NodeExistsException e) { if (log.isDebugEnabled()) log.debug("Failed to create nodes using bulk operation: " + e); for (Op op : ops) createIfNeeded(op.getPath(), null, createMode); break; } catch (Exception e) { onZookeeperError(connStartTime, e); } } } }
ZookeeperClient implements Watcher { void createAll(List<String> paths, CreateMode createMode) throws ZookeeperClientFailedException, InterruptedException { if (paths.isEmpty()) return; List<List<Op>> batches = new LinkedList<>(); int batchSize = 0; List<Op> batch = new LinkedList<>(); for (String path : paths) { int size = requestOverhead(path) + 48 ; assert size <= MAX_REQ_SIZE; if (batchSize + size > MAX_REQ_SIZE) { batches.add(batch); batch = new LinkedList<>(); batchSize = 0; } batch.add(Op.create(path, EMPTY_BYTES, ZK_ACL, createMode)); batchSize += size; } batches.add(batch); for (List<Op> ops : batches) { for (;;) { long connStartTime = this.connStartTime; try { zk.multi(ops); break; } catch (KeeperException.NodeExistsException e) { if (log.isDebugEnabled()) log.debug("Failed to create nodes using bulk operation: " + e); for (Op op : ops) createIfNeeded(op.getPath(), null, createMode); break; } catch (Exception e) { onZookeeperError(connStartTime, e); } } } } }
ZookeeperClient implements Watcher { void createAll(List<String> paths, CreateMode createMode) throws ZookeeperClientFailedException, InterruptedException { if (paths.isEmpty()) return; List<List<Op>> batches = new LinkedList<>(); int batchSize = 0; List<Op> batch = new LinkedList<>(); for (String path : paths) { int size = requestOverhead(path) + 48 ; assert size <= MAX_REQ_SIZE; if (batchSize + size > MAX_REQ_SIZE) { batches.add(batch); batch = new LinkedList<>(); batchSize = 0; } batch.add(Op.create(path, EMPTY_BYTES, ZK_ACL, createMode)); batchSize += size; } batches.add(batch); for (List<Op> ops : batches) { for (;;) { long connStartTime = this.connStartTime; try { zk.multi(ops); break; } catch (KeeperException.NodeExistsException e) { if (log.isDebugEnabled()) log.debug("Failed to create nodes using bulk operation: " + e); for (Op op : ops) createIfNeeded(op.getPath(), null, createMode); break; } catch (Exception e) { onZookeeperError(connStartTime, e); } } } } ZookeeperClient(IgniteLogger log, String connectString, int sesTimeout, IgniteRunnable connLostC);   ZookeeperClient(String igniteInstanceName,
        IgniteLogger log,
        String connectString,
        int sesTimeout,
        IgniteRunnable connLostC); }
ZookeeperClient implements Watcher { void createAll(List<String> paths, CreateMode createMode) throws ZookeeperClientFailedException, InterruptedException { if (paths.isEmpty()) return; List<List<Op>> batches = new LinkedList<>(); int batchSize = 0; List<Op> batch = new LinkedList<>(); for (String path : paths) { int size = requestOverhead(path) + 48 ; assert size <= MAX_REQ_SIZE; if (batchSize + size > MAX_REQ_SIZE) { batches.add(batch); batch = new LinkedList<>(); batchSize = 0; } batch.add(Op.create(path, EMPTY_BYTES, ZK_ACL, createMode)); batchSize += size; } batches.add(batch); for (List<Op> ops : batches) { for (;;) { long connStartTime = this.connStartTime; try { zk.multi(ops); break; } catch (KeeperException.NodeExistsException e) { if (log.isDebugEnabled()) log.debug("Failed to create nodes using bulk operation: " + e); for (Op op : ops) createIfNeeded(op.getPath(), null, createMode); break; } catch (Exception e) { onZookeeperError(connStartTime, e); } } } } ZookeeperClient(IgniteLogger log, String connectString, int sesTimeout, IgniteRunnable connLostC);   ZookeeperClient(String igniteInstanceName,
        IgniteLogger log,
        String connectString,
        int sesTimeout,
        IgniteRunnable connLostC); @Override void process(WatchedEvent evt); void close(); }
ZookeeperClient implements Watcher { void createAll(List<String> paths, CreateMode createMode) throws ZookeeperClientFailedException, InterruptedException { if (paths.isEmpty()) return; List<List<Op>> batches = new LinkedList<>(); int batchSize = 0; List<Op> batch = new LinkedList<>(); for (String path : paths) { int size = requestOverhead(path) + 48 ; assert size <= MAX_REQ_SIZE; if (batchSize + size > MAX_REQ_SIZE) { batches.add(batch); batch = new LinkedList<>(); batchSize = 0; } batch.add(Op.create(path, EMPTY_BYTES, ZK_ACL, createMode)); batchSize += size; } batches.add(batch); for (List<Op> ops : batches) { for (;;) { long connStartTime = this.connStartTime; try { zk.multi(ops); break; } catch (KeeperException.NodeExistsException e) { if (log.isDebugEnabled()) log.debug("Failed to create nodes using bulk operation: " + e); for (Op op : ops) createIfNeeded(op.getPath(), null, createMode); break; } catch (Exception e) { onZookeeperError(connStartTime, e); } } } } ZookeeperClient(IgniteLogger log, String connectString, int sesTimeout, IgniteRunnable connLostC);   ZookeeperClient(String igniteInstanceName,
        IgniteLogger log,
        String connectString,
        int sesTimeout,
        IgniteRunnable connLostC); @Override void process(WatchedEvent evt); void close();  }