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

ZKStoreHelper.getData介绍

暂无

代码示例

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

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

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

@Override
CompletableFuture<Data> getCommitTxnRecord() {
  return store.getData(committingTxnsPath);
}

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

@Override
CompletableFuture<Data> getRetentionSetData() {
  return store.getData(retentionSetPath);
}

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

@Override
CompletableFuture<Data> getWaitingRequestNode() {
  return store.getData(waitingRequestProcessorPath);
}

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

@Override
CompletableFuture<Data> getSealedSegmentSizesMapShardData(int shard) {
  String path = String.format(segmentsSealedSizeMapShardPathFormat, shard);
  return store.getData(path);
}

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

@Override
CompletableFuture<Data> getActiveTx(final int epoch, final UUID txId) {
  final String activeTxPath = getActiveTxPath(epoch, txId.toString());
  return store.getData(activeTxPath);
}

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

@Override
CompletableFuture<Data> getMarkerData(long segmentId) {
  final CompletableFuture<Data> result = new CompletableFuture<>();
  final String path = ZKPaths.makePath(markerPath, String.format("%d", segmentId));
  store.getData(path)
     .whenComplete((res, ex) -> {
       if (ex != null) {
         Throwable cause = Exceptions.unwrap(ex);
         if (cause instanceof StoreException.DataNotFoundException) {
           result.complete(null);
         } else {
           result.completeExceptionally(cause);
         }
       } else {
         result.complete(res);
       }
     });
  return result;
}

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

@VisibleForTesting
CompletableFuture<Void> fetchVersion() {
  return zkStoreHelper.getData(guardPath)
      .thenAccept(data -> latestVersion.set(data.getVersion().asIntVersion().getIntValue()));
}

代码示例来源：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

@Override
public CompletableFuture<Integer> getSafeStartingSegmentNumberFor(final String scopeName, final String streamName) {
  return storeHelper.getData(String.format(DELETED_STREAMS_PATH, getScopedStreamName(scopeName, streamName)))
           .handleAsync((data, ex) -> {
             if (ex == null) {
               return BitConverter.readInt(data.getData(), 0) + 1;
             } else if (ex instanceof StoreException.DataNotFoundException) {
               return 0;
             } else {
               log.error("Problem found while getting a safe starting segment number for {}.",
                   getScopedStreamName(scopeName, streamName), ex);
               throw new CompletionException(ex);
             }
           });
}

代码示例来源：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

@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 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

@VisibleForTesting
CompletableFuture<Void> getRefreshFuture() {
  return storeHelper.createZNodeIfNotExist(COUNTER_PATH, Int96.ZERO.toBytes())
      .thenCompose(v -> storeHelper.getData(COUNTER_PATH)
          .thenCompose(data -> {
            Int96 previous = Int96.fromBytes(data.getData());
            Int96 nextLimit = previous.add(COUNTER_RANGE);
            return storeHelper.setData(COUNTER_PATH, new Data(nextLimit.toBytes(), data.getVersion()))
                .thenAccept(x -> {
                  // Received new range, we should reset the counter and limit under the lock
                  // and then reset refreshfutureref to null
                  synchronized (lock) {
                    // Note: counter is set to previous range's highest value. Always get the
                    // next counter by calling counter.incrementAndGet otherwise there will
                    // be a collision with counter used by someone else.
                    counter.set(previous.getMsb(), previous.getLsb());
                    limit.set(nextLimit.getMsb(), nextLimit.getLsb());
                    refreshFutureRef = null;
                    log.info("Refreshed counter range. Current counter is {}. Current limit is {}", counter.get(), limit.get());
                  }
                });
          }));
}

代码示例来源：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

doReturn(CompletableFuture.completedFuture(data)).when(storeHelper).getData(COUNTER_PATH);