@Test public void testFindEntriesSince() throws Exception { setupCloudStore(false, true, defaultCacheLimit, true); long maxTotalSize = 1000000; long blobSize = 200000; int numBlobsFound = 5; CosmosChangeFeedFindToken cosmosChangeFeedFindToken = new CosmosChangeFeedFindToken(blobSize * numBlobsFound, "start", "end", 0, numBlobsFound, UUID.randomUUID().toString()); when(dest.findEntriesSince(anyString(), any(CosmosChangeFeedFindToken.class), anyLong())).thenReturn( new FindResult(Collections.emptyList(), cosmosChangeFeedFindToken)); CosmosChangeFeedFindToken startToken = new CosmosChangeFeedFindToken(); FindInfo findInfo = store.findEntriesSince(startToken, maxTotalSize); CosmosChangeFeedFindToken outputToken = (CosmosChangeFeedFindToken) findInfo.getFindToken(); assertEquals(blobSize * numBlobsFound, outputToken.getBytesRead()); assertEquals(numBlobsFound, outputToken.getTotalItems()); assertEquals(0, outputToken.getIndex()); cosmosChangeFeedFindToken = new CosmosChangeFeedFindToken(blobSize * 2 * numBlobsFound, "start2", "end2", 0, numBlobsFound, UUID.randomUUID().toString()); when(dest.findEntriesSince(anyString(), any(CosmosChangeFeedFindToken.class), anyLong())).thenReturn( new FindResult(Collections.emptyList(), cosmosChangeFeedFindToken)); findInfo = store.findEntriesSince(outputToken, maxTotalSize); outputToken = (CosmosChangeFeedFindToken) findInfo.getFindToken(); assertEquals(blobSize * 2 * numBlobsFound, outputToken.getBytesRead()); assertEquals(numBlobsFound, outputToken.getTotalItems()); assertEquals(0, outputToken.getIndex()); when(dest.findEntriesSince(anyString(), any(CosmosChangeFeedFindToken.class), anyLong())).thenReturn( new FindResult(Collections.emptyList(), outputToken)); findInfo = store.findEntriesSince(outputToken, maxTotalSize); assertTrue(findInfo.getMessageEntries().isEmpty()); FindToken finalToken = findInfo.getFindToken(); assertEquals(outputToken, finalToken); }
@Override public FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries) throws StoreException { try { FindResult findResult = requestAgent.doWithRetries( () -> cloudDestination.findEntriesSince(partitionId.toPathString(), token, maxTotalSizeOfEntries), "FindEntriesSince", partitionId.toPathString()); if (findResult.getMetadataList().isEmpty()) { return new FindInfo(Collections.emptyList(), findResult.getUpdatedFindToken()); } List<MessageInfo> messageEntries = new ArrayList<>(); for (CloudBlobMetadata metadata : findResult.getMetadataList()) { messageEntries.add(getMessageInfoFromMetadata(metadata)); } return new FindInfo(messageEntries, findResult.getUpdatedFindToken()); } catch (CloudStorageException | IOException ex) { throw new StoreException(ex, StoreErrorCodes.IOError); } }
CloudBlobStore implements Store { @Override public FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries) throws StoreException { try { FindResult findResult = requestAgent.doWithRetries( () -> cloudDestination.findEntriesSince(partitionId.toPathString(), token, maxTotalSizeOfEntries), "FindEntriesSince", partitionId.toPathString()); if (findResult.getMetadataList().isEmpty()) { return new FindInfo(Collections.emptyList(), findResult.getUpdatedFindToken()); } List<MessageInfo> messageEntries = new ArrayList<>(); for (CloudBlobMetadata metadata : findResult.getMetadataList()) { messageEntries.add(getMessageInfoFromMetadata(metadata)); } return new FindInfo(messageEntries, findResult.getUpdatedFindToken()); } catch (CloudStorageException | IOException ex) { throw new StoreException(ex, StoreErrorCodes.IOError); } } }
CloudBlobStore implements Store { @Override public FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries) throws StoreException { try { FindResult findResult = requestAgent.doWithRetries( () -> cloudDestination.findEntriesSince(partitionId.toPathString(), token, maxTotalSizeOfEntries), "FindEntriesSince", partitionId.toPathString()); if (findResult.getMetadataList().isEmpty()) { return new FindInfo(Collections.emptyList(), findResult.getUpdatedFindToken()); } List<MessageInfo> messageEntries = new ArrayList<>(); for (CloudBlobMetadata metadata : findResult.getMetadataList()) { messageEntries.add(getMessageInfoFromMetadata(metadata)); } return new FindInfo(messageEntries, findResult.getUpdatedFindToken()); } catch (CloudStorageException | IOException ex) { throw new StoreException(ex, StoreErrorCodes.IOError); } } CloudBlobStore(VerifiableProperties properties, PartitionId partitionId, CloudDestination cloudDestination,
      ClusterMap clusterMap, VcrMetrics vcrMetrics); }
CloudBlobStore implements Store { @Override public FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries) throws StoreException { try { FindResult findResult = requestAgent.doWithRetries( () -> cloudDestination.findEntriesSince(partitionId.toPathString(), token, maxTotalSizeOfEntries), "FindEntriesSince", partitionId.toPathString()); if (findResult.getMetadataList().isEmpty()) { return new FindInfo(Collections.emptyList(), findResult.getUpdatedFindToken()); } List<MessageInfo> messageEntries = new ArrayList<>(); for (CloudBlobMetadata metadata : findResult.getMetadataList()) { messageEntries.add(getMessageInfoFromMetadata(metadata)); } return new FindInfo(messageEntries, findResult.getUpdatedFindToken()); } catch (CloudStorageException | IOException ex) { throw new StoreException(ex, StoreErrorCodes.IOError); } } CloudBlobStore(VerifiableProperties properties, PartitionId partitionId, CloudDestination cloudDestination,
      ClusterMap clusterMap, VcrMetrics vcrMetrics); @Override void start(); @Override StoreInfo get(List<? extends StoreKey> ids, EnumSet<StoreGetOptions> storeGetOptions); @Override void put(MessageWriteSet messageSetToWrite); @Override void delete(List<MessageInfo> infos); @Override short undelete(MessageInfo info); @Override void updateTtl(List<MessageInfo> infos); @Override FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries); @Override Set<StoreKey> findMissingKeys(List<StoreKey> keys); @Override MessageInfo findKey(StoreKey key); @Override StoreStats getStoreStats(); @Override boolean isKeyDeleted(StoreKey key); @Override long getSizeInBytes(); @Override boolean isEmpty(); @Override boolean isBootstrapInProgress(); @Override boolean isDecommissionInProgress(); @Override void completeBootstrap(); @Override void setCurrentState(ReplicaState state); @Override ReplicaState getCurrentState(); @Override long getEndPositionOfLastPut(); @Override boolean recoverFromDecommission(); @Override boolean disabledOnError(); @Override void shutdown(); @Override boolean isStarted(); @Override String toString(); }
CloudBlobStore implements Store { @Override public FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries) throws StoreException { try { FindResult findResult = requestAgent.doWithRetries( () -> cloudDestination.findEntriesSince(partitionId.toPathString(), token, maxTotalSizeOfEntries), "FindEntriesSince", partitionId.toPathString()); if (findResult.getMetadataList().isEmpty()) { return new FindInfo(Collections.emptyList(), findResult.getUpdatedFindToken()); } List<MessageInfo> messageEntries = new ArrayList<>(); for (CloudBlobMetadata metadata : findResult.getMetadataList()) { messageEntries.add(getMessageInfoFromMetadata(metadata)); } return new FindInfo(messageEntries, findResult.getUpdatedFindToken()); } catch (CloudStorageException | IOException ex) { throw new StoreException(ex, StoreErrorCodes.IOError); } } CloudBlobStore(VerifiableProperties properties, PartitionId partitionId, CloudDestination cloudDestination,
      ClusterMap clusterMap, VcrMetrics vcrMetrics); @Override void start(); @Override StoreInfo get(List<? extends StoreKey> ids, EnumSet<StoreGetOptions> storeGetOptions); @Override void put(MessageWriteSet messageSetToWrite); @Override void delete(List<MessageInfo> infos); @Override short undelete(MessageInfo info); @Override void updateTtl(List<MessageInfo> infos); @Override FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries); @Override Set<StoreKey> findMissingKeys(List<StoreKey> keys); @Override MessageInfo findKey(StoreKey key); @Override StoreStats getStoreStats(); @Override boolean isKeyDeleted(StoreKey key); @Override long getSizeInBytes(); @Override boolean isEmpty(); @Override boolean isBootstrapInProgress(); @Override boolean isDecommissionInProgress(); @Override void completeBootstrap(); @Override void setCurrentState(ReplicaState state); @Override ReplicaState getCurrentState(); @Override long getEndPositionOfLastPut(); @Override boolean recoverFromDecommission(); @Override boolean disabledOnError(); @Override void shutdown(); @Override boolean isStarted(); @Override String toString();  }