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

ZKStoreHelper介绍

暂无

代码示例

代码示例来源：origin: pravega/pravega

@Override
public CompletableFuture<Void> createScope() {
  return store.addNode(scopePath);
}

代码示例来源：origin: pravega/pravega

@Override
public CompletableFuture<String> getScopeConfiguration(final String scopeName) {
  return storeHelper.checkExists(String.format("/store/%s", scopeName))
      .thenApply(scopeExists -> {
        if (scopeExists) {
          return scopeName;
        } else {
          throw StoreException.create(StoreException.Type.DATA_NOT_FOUND, scopeName);
        }
      });
}

代码示例来源：origin: pravega/pravega

CompletableFuture<Integer> createZNodeIfNotExist(final String path, final byte[] data) {
  return createZNodeIfNotExist(path, data, true);
}

代码示例来源：origin: pravega/pravega

private CompletableFuture<Void> gcCompletedTxn() {
  return storeHelper.getChildren(COMPLETED_TX_BATCH_ROOT_PATH)
      .thenApply(children -> {
            // retain latest two and delete remainder.
            List<Long> list = children.stream().map(Long::parseLong).sorted().collect(Collectors.toList());
            if (list.size() > 2) {
              return list.subList(0, list.size() - 2);
            } else {
              return new ArrayList<Long>();
            }
          }
      )
      .thenCompose(toDeleteList -> {
        log.debug("deleting batches {} on new scheme" + toDeleteList);
        // delete all those marked for toDelete.
        return Futures.allOf(toDeleteList.stream()
            .map(toDelete -> storeHelper.deleteTree(String.format(COMPLETED_TX_BATCH_PATH, toDelete)))
            .collect(Collectors.toList()));
      });
}

代码示例来源：origin: pravega/pravega

@Test
  public void testEphemeralNode() {
    CuratorFramework cli2 = CuratorFrameworkFactory.newClient(zkServer.getConnectString(), new RetryNTimes(0, 0));
    cli2.start();
    ZKStoreHelper zkStoreHelper2 = new ZKStoreHelper(cli2, executor);

    Assert.assertTrue(zkStoreHelper2.createEphemeralZNode("/testEphemeral", new byte[0]).join());
    Assert.assertNotNull(zkStoreHelper2.getData("/testEphemeral").join());
    zkStoreHelper2.getClient().close();
    // let session get expired.
    // now read the data again. Verify that node no longer exists
    AssertExtensions.assertFutureThrows("", Futures.delayedFuture(() -> zkStoreHelper.getData("/testEphemeral"), 1000, executor),
        e -> e instanceof StoreException.DataNotFoundException);
  }
}

代码示例来源：origin: pravega/pravega

@Override
public CompletableFuture<Void> addUpdateStreamForAutoStreamCut(final String scope, final String stream, final RetentionPolicy retentionPolicy,
                                final OperationContext context, final Executor executor) {
  Preconditions.checkNotNull(retentionPolicy);
  int bucket = getBucket(scope, stream);
  String retentionPath = String.format(RETENTION_PATH, bucket, encodedScopedStreamName(scope, stream));
  byte[] serialize = SerializationUtils.serialize(retentionPolicy);
  return storeHelper.getData(retentionPath)
      .exceptionally(e -> {
        if (e instanceof StoreException.DataNotFoundException) {
          return null;
        } else {
          throw new CompletionException(e);
        }
      }).thenCompose(data -> {
        if (data == null) {
          return Futures.toVoid(storeHelper.createZNodeIfNotExist(retentionPath, serialize));
        } else {
          return Futures.toVoid(storeHelper.setData(retentionPath, new Data(serialize, data.getVersion())));
        }
      });
}

代码示例来源：origin: pravega/pravega

@Test(timeout = 10000)
public void testGetActiveTxn() throws Exception {
  ZKStoreHelper storeHelper = spy(new ZKStoreHelper(cli, executor));
  ZKStream stream = new ZKStream("scope", "stream", storeHelper);
  final int startingSegmentNumber = 0;
  storeHelper.createZNodeIfNotExist("/store/scope").join();
  final ScalingPolicy policy1 = ScalingPolicy.fixed(2);
  final StreamConfiguration configuration1 = StreamConfiguration.builder().scalingPolicy(policy1).build();
  stream.create(configuration1, System.currentTimeMillis(), startingSegmentNumber).join();
  stream.updateState(State.ACTIVE).join();
  UUID txId = stream.generateNewTxnId(0, 0L).join();
  stream.createTransaction(txId, 1000L, 1000L).join();
  String activeTxPath = stream.getActiveTxPath(0, txId.toString());
  // throw DataNotFoundException for txn path
  doReturn(Futures.failedFuture(StoreException.create(StoreException.Type.DATA_NOT_FOUND, "txn data not found")))
      .when(storeHelper).getData(eq(activeTxPath));
  Map<String, Data> result = stream.getCurrentTxns().join();
  // verify that call succeeds and no active txns were found
  assertTrue(result.isEmpty());
  // throw generic exception for txn path
  doReturn(Futures.failedFuture(new RuntimeException())).when(storeHelper).getData(eq(activeTxPath));
  ZKStream stream2 = new ZKStream("scope", "stream", storeHelper);
  // verify that the call fails
  AssertExtensions.assertFutureThrows("", stream2.getCurrentTxns(), e -> Exceptions.unwrap(e) instanceof RuntimeException);
  reset(storeHelper);
  ZKStream stream3 = new ZKStream("scope", "stream", storeHelper);
  result = stream3.getCurrentTxns().join();
  assertEquals(1, result.size());
}

代码示例来源：origin: pravega/pravega

@Override
CompletableFuture<Data> getEpochTransitionNode() {
  return store.getData(epochTransitionPath);
}

代码示例来源：origin: pravega/pravega

/**
 * List Scopes in the cluster.
 *
 * @return A list of scopes.
 */
public CompletableFuture<List<String>> listScopes() {
  return getChildren("/store");
}

代码示例来源：origin: pravega/pravega

@Override
public CompletableFuture<Map<String, Data>> getTxnInEpoch(int epoch) {
  return Futures.exceptionallyExpecting(store.getChildren(getEpochPath(epoch)),
      e -> Exceptions.unwrap(e) instanceof StoreException.DataNotFoundException, Collections.emptyList())
         .thenCompose(txIds -> Futures.allOfWithResults(txIds.stream().collect(
             Collectors.toMap(txId -> txId, txId -> Futures.exceptionallyExpecting(store.getData(getActiveTxPath(epoch, txId)),
                 e -> Exceptions.unwrap(e) instanceof StoreException.DataNotFoundException, EMPTY_DATA)))
             ).thenApply(txnMap -> txnMap.entrySet().stream().filter(x -> !x.getValue().equals(EMPTY_DATA))
                           .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)))
         );
}

代码示例来源：origin: pravega/pravega

@Override
public CompletableFuture<Boolean> takeBucketOwnership(int bucket, String processId, Executor executor) {
  Preconditions.checkArgument(bucket < bucketCount);
  // try creating an ephemeral node
  String bucketPath = ZKPaths.makePath(BUCKET_OWNERSHIP_PATH, String.valueOf(bucket));
  return storeHelper.createEphemeralZNode(bucketPath, SerializationUtils.serialize(processId))
      .thenCompose(created -> {
        if (!created) {
          // Note: data may disappear by the time we do a getData. Let exception be thrown from here
          // so that caller may retry.
          return storeHelper.getData(bucketPath)
              .thenApply(data -> (SerializationUtils.deserialize(data.getData())).equals(processId));
        } else {
          return CompletableFuture.completedFuture(true);
        }
      });
}

代码示例来源：origin: pravega/pravega

@VisibleForTesting
ZKStreamMetadataStore(CuratorFramework client, int bucketCount, Executor executor, Duration gcPeriod) {
  super(new ZKHostIndex(client, "/hostTxnIndex", executor), bucketCount);
  storeHelper = new ZKStoreHelper(client, executor);
  bucketCacheMap = new ConcurrentHashMap<>();
  bucketOwnershipCacheRef = new AtomicReference<>();
  this.lock = new Object();
  this.counter = new AtomicInt96();
  this.limit = new AtomicInt96();
  this.refreshFutureRef = null;
  this.completedTxnGC = new ZKGarbageCollector(COMPLETED_TXN_GC_NAME, storeHelper, this::gcCompletedTxn, gcPeriod);
  this.completedTxnGC.startAsync();
  this.completedTxnGC.awaitRunning();
}

代码示例来源：origin: pravega/pravega

@Test
public void testDeleteNode() throws ExecutionException, InterruptedException, IOException {
  Assert.assertNull(zkStoreHelper.addNode("/test/test1").get());
  Assert.assertNull(zkStoreHelper.addNode("/test/test1/test2").get());
  AssertExtensions.assertFutureThrows("Should throw NodeNotEmptyException", zkStoreHelper.deleteNode("/test/test1"),
      e -> e instanceof StoreException.DataNotEmptyException);
  Assert.assertNull(zkStoreHelper.deleteNode("/test/test1/test2").get());
  Assert.assertNull(zkStoreHelper.deleteNode("/test/test1").get());
  AssertExtensions.assertFutureThrows("Should throw NodeNotFoundException", zkStoreHelper.deleteNode("/test/test1"),
      e -> e instanceof StoreException.DataNotFoundException);
  zkServer.stop();
  AssertExtensions.assertFutureThrows("Should throw UnknownException", zkStoreHelper.deleteNode("/test/test1"),
      e -> e instanceof StoreException.StoreConnectionException);
}

代码示例来源：origin: pravega/pravega

@Test
public void verifyBucketInitialization() {
  ZKStoreHelper zkStoreHelper = new ZKStoreHelper(cli, executor);
  // Verify that buckets are not initialized.
  assertFalse(zkStoreHelper.checkExists(ZKStreamMetadataStore.BUCKET_ROOT_PATH).join());
  // Execute the initialization of buckets in ZKStreamMetadataStore.
  store.createBucketsRoot().join();
  // Verify that the expected buckets are created after the execution of createBucketsRoot().
  assertTrue(zkStoreHelper.checkExists(ZKStreamMetadataStore.BUCKET_ROOT_PATH).join());
  assertTrue(zkStoreHelper.checkExists(ZKStreamMetadataStore.BUCKET_OWNERSHIP_PATH).join());
  for (int i = 0; i < ((AbstractStreamMetadataStore) store).getBucketCount(); i++) {
    assertTrue(zkStoreHelper.checkExists(String.format(ZKStreamMetadataStore.BUCKET_PATH, i)).join());
  }
}

代码示例来源：origin: pravega/pravega

@Test
public void testCounterConcurrentUpdates() {
  ZKStoreHelper storeHelper = spy(new ZKStoreHelper(cli, executor));
  storeHelper.createZNodeIfNotExist("/store/scope").join();
  ZKStreamMetadataStore zkStore = spy((ZKStreamMetadataStore) this.store);
  ZKStreamMetadataStore zkStore2 = spy((ZKStreamMetadataStore) this.store);
  ZKStreamMetadataStore zkStore3 = spy((ZKStreamMetadataStore) this.store);
  zkStore.setStoreHelperForTesting(storeHelper);
  // first call should get the new range from store
  Int96 counter = zkStore.getNextCounter().join();
  // verify that the generated counter is from new range
  assertEquals(0, counter.getMsb());
  assertEquals(1L, counter.getLsb());
  assertEquals(zkStore.getCounterForTesting(), counter);
  Int96 limit = zkStore.getLimitForTesting();
  assertEquals(ZKStreamMetadataStore.COUNTER_RANGE, limit.getLsb());
  zkStore3.getRefreshFuture().join();
  assertEquals(ZKStreamMetadataStore.COUNTER_RANGE, zkStore3.getCounterForTesting().getLsb());
  assertEquals(ZKStreamMetadataStore.COUNTER_RANGE * 2, zkStore3.getLimitForTesting().getLsb());
  zkStore2.getRefreshFuture().join();
  assertEquals(ZKStreamMetadataStore.COUNTER_RANGE * 2, zkStore2.getCounterForTesting().getLsb());
  assertEquals(ZKStreamMetadataStore.COUNTER_RANGE * 3, zkStore2.getLimitForTesting().getLsb());
  zkStore.getRefreshFuture().join();
  assertEquals(ZKStreamMetadataStore.COUNTER_RANGE * 3, zkStore.getCounterForTesting().getLsb());
  assertEquals(ZKStreamMetadataStore.COUNTER_RANGE * 4, zkStore.getLimitForTesting().getLsb());
}

代码示例来源：origin: pravega/pravega

CompletableFuture<Void> addNode(final String path) {
  final CompletableFuture<Void> result = new CompletableFuture<>();
  try {
    client.create().creatingParentsIfNeeded().inBackground(
        callback(x -> result.complete(null), result::completeExceptionally, path), executor).forPath(path);
  } catch (Exception e) {
    result.completeExceptionally(StoreException.create(StoreException.Type.UNKNOWN, e, path));
  }
  return result;
}

代码示例来源：origin: pravega/pravega

@SneakyThrows(Exception.class)
private NodeCache registerWatch(String watchPath) {
  NodeCache nodeCache = new NodeCache(zkStoreHelper.getClient(), watchPath);
  NodeCacheListener watchListener = () -> {
    currentBatch.set(nodeCache.getCurrentData().getStat().getVersion());
    log.debug("Current batch for {} changed to {}", gcName, currentBatch.get());
  };
  nodeCache.getListenable().addListener(watchListener);
  nodeCache.start();
  return nodeCache;
}

代码示例来源：origin: pravega/pravega

@Override
public CompletableFuture<Void> deleteScope() {
  return store.deleteNode(scopePath);
}

代码示例来源：origin: pravega/pravega

@Override
CompletableFuture<Void> recordLastStreamSegment(final String scope, final String stream, final int lastActiveSegment,
                        OperationContext context, final Executor executor) {
  final String deletePath = String.format(DELETED_STREAMS_PATH, getScopedStreamName(scope, stream));
  byte[] maxSegmentNumberBytes = new byte[Integer.BYTES];
  BitConverter.writeInt(maxSegmentNumberBytes, 0, lastActiveSegment);
  return storeHelper.getData(deletePath)
           .exceptionally(e -> {
             if (e instanceof StoreException.DataNotFoundException) {
               return null;
             } else {
               throw new CompletionException(e);
             }
           })
           .thenCompose(data -> {
             log.debug("Recording last segment {} for stream {}/{} on deletion.", lastActiveSegment, scope, stream);
             if (data == null) {
               return Futures.toVoid(storeHelper.createZNodeIfNotExist(deletePath, maxSegmentNumberBytes));
             } else {
               final int oldLastActiveSegment = BitConverter.readInt(data.getData(), 0);
               Preconditions.checkArgument(lastActiveSegment >= oldLastActiveSegment,
                   "Old last active segment ({}) for {}/{} is higher than current one {}.",
                   oldLastActiveSegment, scope, stream, lastActiveSegment);
               return Futures.toVoid(storeHelper.setData(deletePath, new Data(maxSegmentNumberBytes, data.getVersion())));
             }
           });
}

代码示例来源：origin: pravega/pravega

@Test
public void testCounter() throws Exception {
  ZKStoreHelper storeHelper = spy(new ZKStoreHelper(cli, executor));
  storeHelper.createZNodeIfNotExist("/store/scope").join();
  doReturn(CompletableFuture.completedFuture(data)).when(storeHelper).getData(COUNTER_PATH);